3 * tailing reliable realtime streaming feeder for inn
4 * conn.c - connection establishment and teardown
6 * Copyright (C) 2010 Ian Jackson <ijackson@chiark.greenend.org.uk>
8 * This program is free software: you can redistribute it and/or modify
9 * it under the terms of the GNU General Public License as published by
10 * the Free Software Foundation, either version 3 of the License, or
11 * (at your option) any later version.
13 * This program is distributed in the hope that it will be useful,
14 * but WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 * GNU General Public License for more details.
18 * You should have received a copy of the GNU General Public License
19 * along with this program. If not, see <http://www.gnu.org/licenses/>.
21 * (I believe that when you compile and link this as part of the inn2
22 * build, with the Makefile runes I have provided, all the libraries
23 * and files which end up included in innduct are licence-compatible
24 * with GPLv3. If not then please let me know. -Ian Jackson.)
29 /*========== management of connections ==========*/
31 static void reconnect_blocking_event(void) {
32 until_connect= reconnect_delay_periods;
35 void conn_closefd(Conn *conn, const char *msgprefix) {
36 int r= close_perhaps(&conn->fd);
37 if (r) info("C%d %serror closing socket: %s",
38 conn->fd, msgprefix, strerror(errno));
41 int conn_busy(Conn *conn) {
43 conn->waiting.count ||
44 conn->priority.count ||
49 void conn_dispose(Conn *conn) {
52 oop_rd_cancel(conn->rd);
53 oop_rd_delete_kill(conn->rd);
57 loop->cancel_fd(loop, conn->fd, OOP_WRITE);
58 loop->cancel_fd(loop, conn->fd, OOP_EXCEPTION);
60 conn_closefd(conn,"");
64 static void *conn_exception(oop_source *lp, int fd,
65 oop_event ev, void *conn_v) {
68 assert(fd == conn->fd);
69 assert(ev == OOP_EXCEPTION);
70 int r= read(conn->fd, &ch, 1);
71 if (r<0) connfail(conn,"read failed: %s",strerror(errno));
72 else connfail(conn,"exceptional condition on socket (peer sent urgent"
73 " data? read(,&ch,1)=%d,ch='\\x%02x')",r,ch);
77 void vconnfail(Conn *conn, const char *fmt, va_list al) {
78 int requeue[art_MaxState];
79 memset(requeue,0,sizeof(requeue));
83 while ((art= LIST_REMHEAD(conn->priority)))
84 LIST_ADDTAIL(art->ipf->queue, art);
86 while ((art= LIST_REMHEAD(conn->waiting)))
87 LIST_ADDTAIL(art->ipf->queue, art);
89 while ((art= LIST_REMHEAD(conn->sent))) {
90 requeue[art->state]++;
91 if (art->state==art_Unsolicited) art->state= art_Unchecked;
92 LIST_ADDTAIL(art->ipf->queue,art);
93 check_reading_pause_resume(art->ipf);
98 for (i=0, d=conn->xmitd; i<conn->xmitu; i++, d++)
101 LIST_REMOVE(conns,conn);
103 char *m= mvasprintf(fmt,al);
104 warn("C%d (now %d) connection failed "
105 "(requeueing " RCI_TRIPLE_FMT_BASE "): %s",
106 conn->fd, conns.count, RCI_TRIPLE_VALS_BASE(requeue, /*nothing*/), m);
109 reconnect_blocking_event();
111 check_assign_articles();
114 void connfail(Conn *conn, const char *fmt, ...) {
117 vconnfail(conn,fmt,al);
121 static void conn_idle_close(Conn *conn, const char *why) {
122 static const char quitcmd[]= "QUIT\r\n";
123 int todo= sizeof(quitcmd)-1;
124 const char *p= quitcmd;
126 int r= write(conn->fd, p, todo);
128 if (isewouldblock(errno))
129 connfail(conn, "blocked writing QUIT to idle connection");
131 connfail(conn, "failed to write QUIT to idle connection: %s",
139 conn->since_activity= 0;
140 dbg("C%d is idle (%s), quitting", conn->fd, why);
147 * For our last connection, we also shut it down if we have had
148 * less than K in the last L
150 void check_idle_conns(void) {
153 int volthisperiod= lowvol_perperiod[lowvol_circptr];
155 lowvol_circptr %= lowvol_periods;
156 lowvol_total += volthisperiod;
157 lowvol_total -= lowvol_perperiod[lowvol_circptr];
158 lowvol_perperiod[lowvol_circptr]= 0;
161 conn->since_activity++;
165 if (conn->since_activity <= need_activity_periods) continue;
167 /* We need to shut this down */
169 connfail(conn,"timed out waiting for response to QUIT (%s)",
171 else if (conn->sent.count)
172 connfail(conn,"timed out waiting for responses");
173 else if (conn->waiting.count || conn->priority.count)
174 connfail(conn,"BUG IN INNDUCT conn has queue but nothing sent");
175 else if (conn->xmitu)
176 connfail(conn,"peer has been sending responses"
177 " before receiving our commands!");
179 conn_idle_close(conn, "no activity");
184 conn= LIST_HEAD(conns);
185 if (!volthisperiod &&
187 lowvol_total < lowvol_thresh &&
189 conn_idle_close(conn, "low volume");
192 /*---------- reporting numbers of connections ----------*/
194 static int conns_max_reported, conns_idle_reported;
196 void notice_conns_more(const char *new_kind) {
197 if (conns.count > conns_max_reported) {
198 notice("up to %d connection(s) (%s)", conns.count, new_kind);
199 conns_max_reported= conns.count;
203 void notice_conns_fewer(void) {
204 if (!conns.count && !conns_idle_reported) {
205 notice("low volume, using intermittent connection");
206 conns_idle_reported= 1;
210 void notice_conns_stats(void) {
211 notice("currently %d connection(s)", conns.count);
212 conns_max_reported= conns.count;
213 conns_idle_reported= 0;
216 /*---------- making new connections ----------*/
218 pid_t connecting_child;
219 int connecting_fdpass_sock;
221 static void connect_attempt_discard(void) {
222 if (connecting_child) {
223 int status= xwaitpid(&connecting_child, "connect");
224 if (!(WIFEXITED(status) ||
225 (WIFSIGNALED(status) && WTERMSIG(status) == SIGKILL)))
226 report_child_status("connect", status);
228 if (connecting_fdpass_sock) {
229 cancel_fd_read_except(connecting_fdpass_sock);
230 xclose_perhaps(&connecting_fdpass_sock, "connecting fdpass socket",0);
234 #define PREP_DECL_MSG_CMSG(msg) \
236 struct iovec msgiov; \
237 msgiov.iov_base= &msgbyte; \
240 memset(&msg,0,sizeof(msg)); \
241 char msg##cbuf[CMSG_SPACE(sizeof(int))]; \
242 msg.msg_iov= &msgiov; \
244 msg.msg_control= msg##cbuf; \
245 msg.msg_controllen= sizeof(msg##cbuf);
247 static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
250 assert(fd == connecting_fdpass_sock);
252 PREP_DECL_MSG_CMSG(msg);
254 ssize_t rs= recvmsg(fd, &msg, 0);
256 if (isewouldblock(errno)) return OOP_CONTINUE;
257 syswarn("failed to read socket from connecting child");
262 LIST_INIT(conn->waiting);
263 LIST_INIT(conn->priority);
264 LIST_INIT(conn->sent);
266 struct cmsghdr *h= 0;
267 if (rs >= 0) h= CMSG_FIRSTHDR(&msg);
269 int status= xwaitpid(&connecting_child, "connect child (broken)");
271 if (WIFEXITED(status)) {
272 if (WEXITSTATUS(status) != 0 &&
273 WEXITSTATUS(status) != CONNCHILD_ESTATUS_STREAM &&
274 WEXITSTATUS(status) != CONNCHILD_ESTATUS_NOSTREAM)
275 /* child already reported the problem */;
277 if (e == OOP_EXCEPTION)
278 warn("connect: connection child exited code %d but"
279 " unexpected exception on fdpass socket",
280 WEXITSTATUS(status));
282 warn("connect: connection child exited code %d but"
284 WEXITSTATUS(status), (int)rs);
286 } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALRM) {
287 warn("connect: connection attempt timed out");
289 report_child_status("connect", status);
294 #define CHK(field, val) \
295 if (h->cmsg_##field != val) { \
296 crash("connect: child sent cmsg with cmsg_" #field "=%ld, expected %ld", \
297 (long)h->cmsg_##field, (long)val); \
300 CHK(level, SOL_SOCKET);
301 CHK(type, SCM_RIGHTS);
302 CHK(len, CMSG_LEN(sizeof(conn->fd)));
305 if (CMSG_NXTHDR(&msg,h)) crash("connect: child sent many cmsgs");
307 memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
310 pid_t got= waitpid(connecting_child, &status, 0);
311 if (got==-1) syscrash("connect: real wait for child");
312 assert(got == connecting_child);
315 if (!WIFEXITED(status)) { report_child_status("connect",status); goto x; }
316 int es= WEXITSTATUS(status);
318 case CONNCHILD_ESTATUS_STREAM: conn->stream= 1; break;
319 case CONNCHILD_ESTATUS_NOSTREAM: conn->stream= 0; break;
321 die("connect: child gave unexpected exit status %d", es);
325 conn->max_queue= conn->stream ? max_queue_per_conn : 1;
327 loop->on_fd(loop, conn->fd, OOP_EXCEPTION, conn_exception, conn);
328 conn->rd= oop_rd_new_fd(loop,conn->fd, 0, 0); /* sets nonblocking, too */
329 if (!conn->fd) crash("oop_rd_new_fd conn failed (fd=%d)",conn->fd);
330 int r= oop_rd_read(conn->rd, &peer_rd_style, NNTP_MAXLEN_COMMAND+1,
333 if (r) syscrash("oop_rd_read for peer (fd=%d)",conn->fd);
335 LIST_ADDHEAD(conns, conn);
336 const char *streamdesc= conn->stream ? "streaming" : "plain";
337 info("C%d (now %d) connected %s", conn->fd, conns.count, streamdesc);
338 notice_conns_more(streamdesc);
340 connect_attempt_discard();
341 check_assign_articles();
346 connect_attempt_discard();
347 reconnect_blocking_event();
351 int allow_connect_start(void) {
352 return conns.count < max_connections
357 void connect_start(void) {
358 assert(!connecting_child);
359 assert(!connecting_fdpass_sock);
361 info("starting connection attempt");
362 int ok_until_connect= until_connect;
363 reconnect_blocking_event();
366 int r= socketpair(AF_UNIX, SOCK_STREAM, 0, socks);
367 if (r) { syswarn("connect: cannot create socketpair for child"); return; }
369 connecting_child= xfork("connection");
371 if (!connecting_child) {
372 FILE *cn_from, *cn_to;
373 char buf[NNTP_MAXLEN_COMMAND+100];
374 int exitstatus= CONNCHILD_ESTATUS_NOSTREAM;
376 xclose(socks[0], "(in child) parent's connection fdpass socket",0);
378 alarm(connection_setup_timeout);
379 buf[sizeof(buf)-1] = 0;
380 if (NNTPconnect(remote_host, port, &cn_from, &cn_to,
381 buf, sizeof(buf)-1) < 0) {
385 unsigned char c= buf[l-1];
386 if (!isspace(c)) break;
387 if (c=='\n' || c=='\r') stripped=1;
391 sysdie("connect: connection attempt failed");
394 die("connect: %s: %s", stripped ? "rejected" : "failed",
398 if (NNTPsendpassword((char*)remote_host, cn_from, cn_to) < 0)
399 sysdie("connect: authentication failed");
401 if (fputs("MODE STREAM\r\n", cn_to)==EOF ||
403 sysdie("connect: could not send MODE STREAM");
404 buf[sizeof(buf)-1]= 0;
405 if (!fgets(buf, sizeof(buf)-1, cn_from)) {
407 sysdie("connect: could not read response to MODE STREAM");
409 die("connect: connection close in response to MODE STREAM");
414 die("connect: response to MODE STREAM is too long: %.100s...",
416 l--; if (l>0 && buf[l-1]=='\r') l--;
419 int rcode= strtoul(buf,&ep,10);
421 die("connect: bad response to MODE STREAM: %.50s", sanitise(buf,-1));
425 exitstatus= CONNCHILD_ESTATUS_STREAM;
431 warn("connect: unexpected response to MODE STREAM: %.50s",
433 exitstatus= CONNCHILD_ESTATUS_NOSTREAM;
437 int fd= fileno(cn_from);
439 PREP_DECL_MSG_CMSG(msg);
440 struct cmsghdr *cmsg= CMSG_FIRSTHDR(&msg);
441 cmsg->cmsg_level= SOL_SOCKET;
442 cmsg->cmsg_type= SCM_RIGHTS;
443 cmsg->cmsg_len= CMSG_LEN(sizeof(fd));
444 memcpy(CMSG_DATA(cmsg), &fd, sizeof(fd));
446 msg.msg_controllen= cmsg->cmsg_len;
447 r= sendmsg(socks[1], &msg, 0);
448 if (r<0) syscrash("sendmsg failed for new connection");
449 if (r!=1) crash("sendmsg for new connection gave wrong result %d",r);
454 xclose(socks[1], "connecting fdpass child's socket",0);
455 connecting_fdpass_sock= socks[0];
456 xsetnonblock(connecting_fdpass_sock, 1);
457 on_fd_read_except(connecting_fdpass_sock, connchild_event);
460 until_connect= ok_until_connect;