chiark / gitweb /
c99174c59317482768167b6f520e2e278471af6d
[innduct.git] / conn.c
1 /*
2  *  innduct
3  *  tailing reliable realtime streaming feeder for inn
4  *  conn.c - connection establishment and teardown
5  *
6  *  Copyright (C) 2010 Ian Jackson <ijackson@chiark.greenend.org.uk>
7  * 
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.
12  * 
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.
17  * 
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/>.
20  *
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.)
25  */
26
27 #include "innduct.h"
28
29 /*========== management of connections ==========*/
30
31 static void reconnect_blocking_event(void) {
32   until_connect= reconnect_delay_periods;
33 }
34
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));
39 }
40
41 int conn_busy(Conn *conn) {
42   return
43     conn->waiting.count ||
44     conn->priority.count ||
45     conn->sent.count ||
46     conn->xmitu;
47 }
48
49 void conn_dispose(Conn *conn) {
50   if (!conn) return;
51   if (conn->rd) {
52     oop_rd_cancel(conn->rd);
53     oop_rd_delete_kill(conn->rd);
54     conn->rd= 0;
55   }
56   if (conn->fd) {
57     loop->cancel_fd(loop, conn->fd, OOP_WRITE);
58     loop->cancel_fd(loop, conn->fd, OOP_EXCEPTION);
59   }
60   conn_closefd(conn,"");
61   free(conn);
62 }
63
64 static void *conn_exception(oop_source *lp, int fd,
65                             oop_event ev, void *conn_v) {
66   Conn *conn= conn_v;
67   unsigned char ch;
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);
74   return OOP_CONTINUE;
75 }  
76
77 void vconnfail(Conn *conn, const char *fmt, va_list al) {
78   int requeue[art_MaxState];
79   memset(requeue,0,sizeof(requeue));
80
81   Article *art;
82   
83   while ((art= LIST_REMHEAD(conn->priority)))
84     LIST_ADDTAIL(art->ipf->queue, art);
85
86   while ((art= LIST_REMHEAD(conn->waiting)))
87     LIST_ADDTAIL(art->ipf->queue, art);
88
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);
94   }
95
96   int i;
97   XmitDetails *d;
98   for (i=0, d=conn->xmitd; i<conn->xmitu; i++, d++)
99     xmit_free(d);
100
101   LIST_REMOVE(conns,conn);
102
103   char *m= xvasprintf(fmt,al);
104   warn("C%d (now %d) connection failed requeueing " RCI_TRIPLE_FMT_BASE ": %s",
105        conn->fd, conns.count, RCI_TRIPLE_VALS_BASE(requeue, /*nothing*/), m);
106   free(m);
107
108   reconnect_blocking_event();
109   conn_dispose(conn);
110   check_assign_articles();
111 }
112
113 void connfail(Conn *conn, const char *fmt, ...) {
114   va_list al;
115   va_start(al,fmt);
116   vconnfail(conn,fmt,al);
117   va_end(al);
118 }
119
120 static void conn_idle_close(Conn *conn, const char *why) {
121   static const char quitcmd[]= "QUIT\r\n";
122   int todo= sizeof(quitcmd)-1;
123   const char *p= quitcmd;
124   for (;;) {
125     int r= write(conn->fd, p, todo);
126     if (r<0) {
127       if (isewouldblock(errno))
128         connfail(conn, "blocked writing QUIT to idle connection");
129       else
130         connfail(conn, "failed to write QUIT to idle connection: %s",
131                  strerror(errno));
132       break;
133     }
134     assert(r<=todo);
135     todo -= r;
136     if (!todo) {
137       conn->quitting= why;
138       conn->since_activity= 0;
139       dbg("C%d is idle (%s), quitting", conn->fd, why);
140       break;
141     }
142   }
143 }
144
145 /*
146  * For our last connection, we also shut it down if we have had
147  * less than K in the last L
148  */
149 void check_idle_conns(void) {
150   Conn *conn;
151
152   int volthisperiod= lowvol_perperiod[lowvol_circptr];
153   lowvol_circptr++;
154   lowvol_circptr %= lowvol_periods;
155   lowvol_total += volthisperiod;
156   lowvol_total -= lowvol_perperiod[lowvol_circptr];
157   lowvol_perperiod[lowvol_circptr]= 0;
158
159   FOR_CONN(conn)
160     conn->since_activity++;
161
162  search_again:
163   FOR_CONN(conn) {
164     if (conn->since_activity <= need_activity_periods) continue;
165
166     /* We need to shut this down */
167     if (conn->quitting)
168       connfail(conn,"timed out waiting for response to QUIT (%s)",
169                conn->quitting);
170     else if (conn->sent.count)
171       connfail(conn,"timed out waiting for responses");
172     else if (conn->waiting.count || conn->priority.count)
173       connfail(conn,"BUG IN INNDUCT conn has queue but nothing sent");
174     else if (conn->xmitu)
175       connfail(conn,"peer has been sending responses"
176                " before receiving our commands!");
177     else
178       conn_idle_close(conn, "no activity");
179     
180     goto search_again;
181   }
182
183   conn= LIST_HEAD(conns);
184   if (!volthisperiod &&
185       conns.count==1 &&
186       lowvol_total < lowvol_thresh &&
187       !conn_busy(conn))
188     conn_idle_close(conn, "low volume");
189 }  
190
191 /*---------- making new connections ----------*/
192
193 pid_t connecting_child;
194 int connecting_fdpass_sock;
195
196 static void connect_attempt_discard(void) {
197   if (connecting_child) {
198     int status= xwaitpid(&connecting_child, "connect");
199     if (!(WIFEXITED(status) ||
200           (WIFSIGNALED(status) && WTERMSIG(status) == SIGKILL)))
201       report_child_status("connect", status);
202   }
203   if (connecting_fdpass_sock) {
204     cancel_fd_read_except(connecting_fdpass_sock);
205     xclose_perhaps(&connecting_fdpass_sock, "connecting fdpass socket",0);
206   }
207 }
208
209 #define PREP_DECL_MSG_CMSG(msg)                 \
210   char msgbyte= 0;                              \
211   struct iovec msgiov;                          \
212   msgiov.iov_base= &msgbyte;                    \
213   msgiov.iov_len= 1;                            \
214   struct msghdr msg;                            \
215   memset(&msg,0,sizeof(msg));                   \
216   char msg##cbuf[CMSG_SPACE(sizeof(int))];      \
217   msg.msg_iov= &msgiov;                         \
218   msg.msg_iovlen= 1;                            \
219   msg.msg_control= msg##cbuf;                   \
220   msg.msg_controllen= sizeof(msg##cbuf);
221
222 static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
223   Conn *conn= 0;
224
225   assert(fd == connecting_fdpass_sock);
226
227   PREP_DECL_MSG_CMSG(msg);
228   
229   ssize_t rs= recvmsg(fd, &msg, 0);
230   if (rs<0) {
231     if (isewouldblock(errno)) return OOP_CONTINUE;
232     syswarn("failed to read socket from connecting child");
233     goto x;
234   }
235
236   NEW(conn);
237   LIST_INIT(conn->waiting);
238   LIST_INIT(conn->priority);
239   LIST_INIT(conn->sent);
240
241   struct cmsghdr *h= 0;
242   if (rs >= 0) h= CMSG_FIRSTHDR(&msg);
243   if (!h) {
244     int status= xwaitpid(&connecting_child, "connect child (broken)");
245
246     if (WIFEXITED(status)) {
247       if (WEXITSTATUS(status) != 0 &&
248           WEXITSTATUS(status) != CONNCHILD_ESTATUS_STREAM &&
249           WEXITSTATUS(status) != CONNCHILD_ESTATUS_NOSTREAM)
250         /* child already reported the problem */;
251       else {
252         if (e == OOP_EXCEPTION)
253           warn("connect: connection child exited code %d but"
254                " unexpected exception on fdpass socket",
255                WEXITSTATUS(status));
256         else
257           warn("connect: connection child exited code %d but"
258                " no cmsg (rs=%d)",
259                WEXITSTATUS(status), (int)rs);
260       }
261     } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALRM) {
262       warn("connect: connection attempt timed out");
263     } else {
264       report_child_status("connect", status);
265     }
266     goto x;
267   }
268
269 #define CHK(field, val)                                                    \
270   if (h->cmsg_##field != val) {                                            \
271     crash("connect: child sent cmsg with cmsg_" #field "=%d, expected %d", \
272           h->cmsg_##field, val);                                           \
273     goto x;                                                                \
274   }
275   CHK(level, SOL_SOCKET);
276   CHK(type,  SCM_RIGHTS);
277   CHK(len,   CMSG_LEN(sizeof(conn->fd)));
278 #undef CHK
279
280   if (CMSG_NXTHDR(&msg,h)) crash("connect: child sent many cmsgs");
281
282   memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
283
284   int status;
285   pid_t got= waitpid(connecting_child, &status, 0);
286   if (got==-1) syscrash("connect: real wait for child");
287   assert(got == connecting_child);
288   connecting_child= 0;
289
290   if (!WIFEXITED(status)) { report_child_status("connect",status); goto x; }
291   int es= WEXITSTATUS(status);
292   switch (es) {
293   case CONNCHILD_ESTATUS_STREAM:    conn->stream= 1;   break;
294   case CONNCHILD_ESTATUS_NOSTREAM:  conn->stream= 0;   break;
295   default:
296     die("connect: child gave unexpected exit status %d", es);
297   }
298
299   /* Phew! */
300   conn->max_queue= conn->stream ? max_queue_per_conn : 1;
301
302   loop->on_fd(loop, conn->fd, OOP_EXCEPTION, conn_exception, conn);
303   conn->rd= oop_rd_new_fd(loop,conn->fd, 0, 0); /* sets nonblocking, too */
304   if (!conn->fd) crash("oop_rd_new_fd conn failed (fd=%d)",conn->fd);
305   int r= oop_rd_read(conn->rd, &peer_rd_style, NNTP_STRLEN,
306                      &peer_rd_ok, conn,
307                      &peer_rd_err, conn);
308   if (r) syscrash("oop_rd_read for peer (fd=%d)",conn->fd);
309
310   LIST_ADDHEAD(conns, conn);
311   notice("C%d (now %d) connected %s",
312          conn->fd, conns.count, conn->stream ? "streaming" : "plain");
313
314   connect_attempt_discard();
315   check_assign_articles();
316   return OOP_CONTINUE;
317
318  x:
319   conn_dispose(conn);
320   connect_attempt_discard();
321   reconnect_blocking_event();
322   return OOP_CONTINUE;
323 }
324
325 int allow_connect_start(void) {
326   return conns.count < max_connections
327     && !connecting_child
328     && !until_connect;
329 }
330
331 void connect_start(void) {
332   assert(!connecting_child);
333   assert(!connecting_fdpass_sock);
334
335   info("starting connection attempt");
336   int ok_until_connect= until_connect;
337   reconnect_blocking_event();
338
339   int socks[2];
340   int r= socketpair(AF_UNIX, SOCK_STREAM, 0, socks);
341   if (r) { syswarn("connect: cannot create socketpair for child"); return; }
342
343   connecting_child= xfork("connection");
344
345   if (!connecting_child) {
346     FILE *cn_from, *cn_to;
347     char buf[NNTP_STRLEN+100];
348     int exitstatus= CONNCHILD_ESTATUS_NOSTREAM;
349
350     xclose(socks[0], "(in child) parent's connection fdpass socket",0);
351
352     alarm(connection_setup_timeout);
353     if (NNTPconnect((char*)remote_host, port, &cn_from, &cn_to, buf) < 0) {
354       int l= strlen(buf);
355       int stripped=0;
356       while (l>0) {
357         unsigned char c= buf[l-1];
358         if (!isspace(c)) break;
359         if (c=='\n' || c=='\r') stripped=1;
360         --l;
361       }
362       if (!buf[0]) {
363         sysdie("connect: connection attempt failed");
364       } else {
365         buf[l]= 0;
366         die("connect: %s: %s", stripped ? "rejected" : "failed",
367             sanitise(buf,-1));
368       }
369     }
370     if (NNTPsendpassword((char*)remote_host, cn_from, cn_to) < 0)
371       sysdie("connect: authentication failed");
372     if (try_stream) {
373       if (fputs("MODE STREAM\r\n", cn_to)==EOF ||
374           fflush(cn_to))
375         sysdie("connect: could not send MODE STREAM");
376       buf[sizeof(buf)-1]= 0;
377       if (!fgets(buf, sizeof(buf)-1, cn_from)) {
378         if (ferror(cn_from))
379           sysdie("connect: could not read response to MODE STREAM");
380         else
381           die("connect: connection close in response to MODE STREAM");
382       }
383       int l= strlen(buf);
384       assert(l>=1);
385       if (buf[l-1]!='\n')
386         die("connect: response to MODE STREAM is too long: %.100s...",
387             sanitise(buf,-1));
388       l--;  if (l>0 && buf[l-1]=='\r') l--;
389       buf[l]= 0;
390       char *ep;
391       int rcode= strtoul(buf,&ep,10);
392       if (ep != &buf[3])
393         die("connect: bad response to MODE STREAM: %.50s", sanitise(buf,-1));
394
395       switch (rcode) {
396       case 203:
397         exitstatus= CONNCHILD_ESTATUS_STREAM;
398         break;
399       case 480:
400       case 500:
401         break;
402       default:
403         warn("connect: unexpected response to MODE STREAM: %.50s",
404              sanitise(buf,-1));
405         exitstatus= 2;
406         break;
407       }
408     }
409     int fd= fileno(cn_from);
410
411     PREP_DECL_MSG_CMSG(msg);
412     struct cmsghdr *cmsg= CMSG_FIRSTHDR(&msg);
413     cmsg->cmsg_level= SOL_SOCKET;
414     cmsg->cmsg_type=  SCM_RIGHTS;
415     cmsg->cmsg_len=   CMSG_LEN(sizeof(fd));
416     memcpy(CMSG_DATA(cmsg), &fd, sizeof(fd));
417
418     msg.msg_controllen= cmsg->cmsg_len;
419     r= sendmsg(socks[1], &msg, 0);
420     if (r<0) syscrash("sendmsg failed for new connection");
421     if (r!=1) crash("sendmsg for new connection gave wrong result %d",r);
422
423     _exit(exitstatus);
424   }
425
426   xclose(socks[1], "connecting fdpass child's socket",0);
427   connecting_fdpass_sock= socks[0];
428   xsetnonblock(connecting_fdpass_sock, 1);
429   on_fd_read_except(connecting_fdpass_sock, connchild_event);
430
431   if (!conns.count)
432     until_connect= ok_until_connect;
433 }
434