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= 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);
 107   free(m);
 108
 109   reconnect_blocking_event();
 110   conn_dispose(conn);
 111   check_assign_articles();
 112 }
 113
 114 void connfail(Conn *conn, const char *fmt, ...) {
 115   va_list al;
 116   va_start(al,fmt);
 117   vconnfail(conn,fmt,al);
 118   va_end(al);
 119 }
 120
 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;
 125   for (;;) {
 126     int r= write(conn->fd, p, todo);
 127     if (r<0) {
 128       if (isewouldblock(errno))
 129         connfail(conn, "blocked writing QUIT to idle connection");
 130       else
 131         connfail(conn, "failed to write QUIT to idle connection: %s",
 132                  strerror(errno));
 133       break;
 134     }
 135     assert(r<=todo);
 136     todo -= r;
 137     if (!todo) {
 138       conn->quitting= why;
 139       conn->since_activity= 0;
 140       dbg("C%d is idle (%s), quitting", conn->fd, why);
 141       break;
 142     }
 143   }
 144 }
 145
 146 /*
 147  * For our last connection, we also shut it down if we have had
 148  * less than K in the last L
 149  */
 150 void check_idle_conns(void) {
 151   Conn *conn;
 152
 153   int volthisperiod= lowvol_perperiod[lowvol_circptr];
 154   lowvol_circptr++;
 155   lowvol_circptr %= lowvol_periods;
 156   lowvol_total += volthisperiod;
 157   lowvol_total -= lowvol_perperiod[lowvol_circptr];
 158   lowvol_perperiod[lowvol_circptr]= 0;
 159
 160   FOR_CONN(conn)
 161     conn->since_activity++;
 162
 163  search_again:
 164   FOR_CONN(conn) {
 165     if (conn->since_activity <= need_activity_periods) continue;
 166
 167     /* We need to shut this down */
 168     if (conn->quitting)
 169       connfail(conn,"timed out waiting for response to QUIT (%s)",
 170                conn->quitting);
 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!");
 178     else
 179       conn_idle_close(conn, "no activity");
 180
 181     goto search_again;
 182   }
 183
 184   conn= LIST_HEAD(conns);
 185   if (!volthisperiod &&
 186       conns.count==1 &&
 187       lowvol_total < lowvol_thresh &&
 188       !conn_busy(conn))
 189     conn_idle_close(conn, "low volume");
 190 }
 191
 192 /*---------- reporting numbers of connections ----------*/
 193
 194 static int conns_max_reported, conns_idle_reported;
 195
 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;
 200   }
 201 }
 202
 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;
 207   }
 208 }
 209
 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;
 214 }
 215
 216 /*---------- making new connections ----------*/
 217
 218 pid_t connecting_child;
 219 int connecting_fdpass_sock;
 220
 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);
 227   }
 228   if (connecting_fdpass_sock) {
 229     cancel_fd_read_except(connecting_fdpass_sock);
 230     xclose_perhaps(&connecting_fdpass_sock, "connecting fdpass socket",0);
 231   }
 232 }
 233
 234 #define PREP_DECL_MSG_CMSG(msg)                 \
 235   char msgbyte= 0;                              \
 236   struct iovec msgiov;                          \
 237   msgiov.iov_base= &msgbyte;                    \
 238   msgiov.iov_len= 1;                            \
 239   struct msghdr msg;                            \
 240   memset(&msg,0,sizeof(msg));                   \
 241   char msg##cbuf[CMSG_SPACE(sizeof(int))];      \
 242   msg.msg_iov= &msgiov;                         \
 243   msg.msg_iovlen= 1;                            \
 244   msg.msg_control= msg##cbuf;                   \
 245   msg.msg_controllen= sizeof(msg##cbuf);
 246
 247 static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
 248   Conn *conn= 0;
 249
 250   assert(fd == connecting_fdpass_sock);
 251
 252   PREP_DECL_MSG_CMSG(msg);
 253
 254   ssize_t rs= recvmsg(fd, &msg, 0);
 255   if (rs<0) {
 256     if (isewouldblock(errno)) return OOP_CONTINUE;
 257     syswarn("failed to read socket from connecting child");
 258     goto x;
 259   }
 260
 261   NEW(conn);
 262   LIST_INIT(conn->waiting);
 263   LIST_INIT(conn->priority);
 264   LIST_INIT(conn->sent);
 265
 266   struct cmsghdr *h= 0;
 267   if (rs >= 0) h= CMSG_FIRSTHDR(&msg);
 268   if (!h) {
 269     int status= xwaitpid(&connecting_child, "connect child (broken)");
 270
 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 */;
 276       else {
 277         if (e == OOP_EXCEPTION)
 278           warn("connect: connection child exited code %d but"
 279                " unexpected exception on fdpass socket",
 280                WEXITSTATUS(status));
 281         else
 282           warn("connect: connection child exited code %d but"
 283                " no cmsg (rs=%d)",
 284                WEXITSTATUS(status), (int)rs);
 285       }
 286     } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALRM) {
 287       warn("connect: connection attempt timed out");
 288     } else {
 289       report_child_status("connect", status);
 290     }
 291     goto x;
 292   }
 293
 294 #define CHK(field, val)                                                    \
 295   if (h->cmsg_##field != val) {                                            \
 296     crash("connect: child sent cmsg with cmsg_" #field "=%d, expected %d", \
 297           h->cmsg_##field, val);                                           \
 298     goto x;                                                                \
 299   }
 300   CHK(level, SOL_SOCKET);
 301   CHK(type,  SCM_RIGHTS);
 302   CHK(len,   CMSG_LEN(sizeof(conn->fd)));
 303 #undef CHK
 304
 305   if (CMSG_NXTHDR(&msg,h)) crash("connect: child sent many cmsgs");
 306
 307   memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
 308
 309   int status;
 310   pid_t got= waitpid(connecting_child, &status, 0);
 311   if (got==-1) syscrash("connect: real wait for child");
 312   assert(got == connecting_child);
 313   connecting_child= 0;
 314
 315   if (!WIFEXITED(status)) { report_child_status("connect",status); goto x; }
 316   int es= WEXITSTATUS(status);
 317   switch (es) {
 318   case CONNCHILD_ESTATUS_STREAM:    conn->stream= 1;   break;
 319   case CONNCHILD_ESTATUS_NOSTREAM:  conn->stream= 0;   break;
 320   default:
 321     die("connect: child gave unexpected exit status %d", es);
 322   }
 323
 324   /* Phew! */
 325   conn->max_queue= conn->stream ? max_queue_per_conn : 1;
 326
 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,
 331                      &peer_rd_ok, conn,
 332                      &peer_rd_err, conn);
 333   if (r) syscrash("oop_rd_read for peer (fd=%d)",conn->fd);
 334
 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);
 339
 340   connect_attempt_discard();
 341   check_assign_articles();
 342   return OOP_CONTINUE;
 343
 344  x:
 345   conn_dispose(conn);
 346   connect_attempt_discard();
 347   reconnect_blocking_event();
 348   return OOP_CONTINUE;
 349 }
 350
 351 int allow_connect_start(void) {
 352   return conns.count < max_connections
 353     && !connecting_child
 354     && !until_connect;
 355 }
 356
 357 void connect_start(void) {
 358   assert(!connecting_child);
 359   assert(!connecting_fdpass_sock);
 360
 361   info("starting connection attempt");
 362   int ok_until_connect= until_connect;
 363   reconnect_blocking_event();
 364
 365   int socks[2];
 366   int r= socketpair(AF_UNIX, SOCK_STREAM, 0, socks);
 367   if (r) { syswarn("connect: cannot create socketpair for child"); return; }
 368
 369   connecting_child= xfork("connection");
 370
 371   if (!connecting_child) {
 372     FILE *cn_from, *cn_to;
 373     char buf[NNTP_MAXLEN_COMMAND+100];
 374     int exitstatus= CONNCHILD_ESTATUS_NOSTREAM;
 375
 376     xclose(socks[0], "(in child) parent's connection fdpass socket",0);
 377
 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) {
 382       int l= strlen(buf);
 383       int stripped=0;
 384       while (l>0) {
 385         unsigned char c= buf[l-1];
 386         if (!isspace(c)) break;
 387         if (c=='\n' || c=='\r') stripped=1;
 388         --l;
 389       }
 390       if (!buf[0]) {
 391         sysdie("connect: connection attempt failed");
 392       } else {
 393         buf[l]= 0;
 394         die("connect: %s: %s", stripped ? "rejected" : "failed",
 395             sanitise(buf,-1));
 396       }
 397     }
 398     if (NNTPsendpassword((char*)remote_host, cn_from, cn_to) < 0)
 399       sysdie("connect: authentication failed");
 400     if (try_stream) {
 401       if (fputs("MODE STREAM\r\n", cn_to)==EOF ||
 402           fflush(cn_to))
 403         sysdie("connect: could not send MODE STREAM");
 404       buf[sizeof(buf)-1]= 0;
 405       if (!fgets(buf, sizeof(buf)-1, cn_from)) {
 406         if (ferror(cn_from))
 407           sysdie("connect: could not read response to MODE STREAM");
 408         else
 409           die("connect: connection close in response to MODE STREAM");
 410       }
 411       int l= strlen(buf);
 412       assert(l>=1);
 413       if (buf[l-1]!='\n')
 414         die("connect: response to MODE STREAM is too long: %.100s...",
 415             sanitise(buf,-1));
 416       l--;  if (l>0 && buf[l-1]=='\r') l--;
 417       buf[l]= 0;
 418       char *ep;
 419       int rcode= strtoul(buf,&ep,10);
 420       if (ep != &buf[3])
 421         die("connect: bad response to MODE STREAM: %.50s", sanitise(buf,-1));
 422
 423       switch (rcode) {
 424       case 203:
 425         exitstatus= CONNCHILD_ESTATUS_STREAM;
 426         break;
 427       case 480:
 428       case 500:
 429         break;
 430       default:
 431         warn("connect: unexpected response to MODE STREAM: %.50s",
 432              sanitise(buf,-1));
 433         exitstatus= CONNCHILD_ESTATUS_NOSTREAM;
 434         break;
 435       }
 436     }
 437     int fd= fileno(cn_from);
 438
 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));
 445
 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);
 450
 451     _exit(exitstatus);
 452   }
 453
 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);
 458
 459   if (!conns.count)
 460     until_connect= ok_until_connect;
 461 }
 462