conn.c

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