backends/innduct.c

   1 /*
   2  * Four files full of
   3  *    token article
   4  *
   5  *   site.name_ductlock        lock taken out by innduct
   6  * F site.name                 written by innd
   7  * D site.name_duct            moved aside by innduct, for flush
   8  *   site.name_deferwork       431'd articles, still being written
   9  *   site.name_defergo_<inum>  431'd articles, ready for innxmit
  10  *   site.name_deferlock       lock taken out by innxmit wrapper
  11  *
  12  *
  13  *
  14  * OVERALL STATES:
  15  *
  16  *                                                   START
  17  *   ,-->--.                                           |
  18  *   |     |                                         stat D
  19  *   |     |                                         /   |
  20  *   |     |                                  ENOENT/    |exists
  21  *   |     V                            <----------'     |
  22  *   |  Normal                                         stat F
  23  *   |   F: innd writing, duct reading                  /|\
  24  *   |   D: ENOENT                                     / | \
  25  *   |     |                                          /  |  \
  26  *   |     |  duct decides time to flush      same   /   |   |
  27  *   |     |  duct makes hardlink             as D  /    |   |
  28  *   |     |                                       /     |   |
  29  *   |     V                            <---------'      |   |
  30  *   |  Hardlinked                                       |   |
  31  *   |   F == D: innd writing, duct reading              |   |
  32  *   ^     |                                             |   |
  33  *   |     |  duct unlinks F                            /    |
  34  *   |     V                                  ENOENT   /     |
  35  *   |  Moved                            <------------'      |
  36  *   |   F: ENOENT                                           |
  37  *   |   D: innd writing, duct reading                       |
  38  *   |     |                                                 |
  39  *   |     |  duct flushes feed                              |
  40  *   |     |   (others can too, harmlessly)                  |
  41  *   |     V                                                 |
  42  *   |  Separated                          <-----------------'
  43  *   |   F: innd writing                        different to D
  44  *   |   D: duct reading
  45  *   |     |
  46  *   |     V  duct completes processing of D
  47  *   |     |  duct unlinks D
  48  *   |     |
  49  *   `--<--'
  50  *
  51  */
  52
  53 static int max_connections, max_queue_per_conn;
  54 static int connection_setup_timeout, port, try_stream;
  55 static const char *remote_host;
  56
  57 #define ISNODE(T)    T *next, *back;
  58 #define LIST(T)      struct { T *head, *tail, *tailpred; int count; }
  59
  60 #define NODE(n) ((struct node*)&(n)->head)
  61
  62 #define LIST_ADDHEAD(l,n)                                               \
  63  (list_addhead((struct list*)&(l), NODE((n))), (void)(l).count++)
  64 #define LIST_ADDTAIL(l,n)                                               \
  65  (list_addtail((struct list*)&(l), NODE((n))), (void)(l).count++)
  66
  67 #define LIST_REMHEAD(l)                                                   \
  68  ((l).count ? ((l).count--, (void*)list_remhead((struct list*)&(l))) : 0)
  69 #define LIST_REMTAIL(l)                                                   \
  70  ((l).count ? ((l).count--, (void*)list_remtail((struct list*)&(l))) : 0)
  71 #define LIST_REMOVE(l,n)                        \
  72  (list_remove(NODE((n))), (void)(l).count--)
  73 #define LIST_INSERT(l,n,pred) \
  74  (list_insert((struct list*)&(l), NODE((n)), NODE((pred))), (void)(l).count++)
  75
  76 struct Article {
  77   char *mid;
  78   int midlen;
  79   int checked;
  80 };
  81
  82 #define CONNBUFSZ 16384
  83
  84 #define CN "<%d> "
  85
  86 typedef struct Conn Conn;
  87
  88 typedef enum {
  89   Malloc, Const, Artdata;
  90 } XmitKind;
  91 typedef struct {
  92   XmitKind kind;
  93   union {
  94     char *malloc_tofree;
  95   } info;
  96 } XmitDetails;
  97
  98 struct Conn {
  99   ISNODE(Conn);
 100   int fd, max_queue, stream;
 101   LIST(Article) queue; /* not yet told peer, or said TAKETHIS */
 102   LIST(Article) sent; /* offered, in xmit, or transmitted waiting reply */
 103   Article send; /* partially transmitted */
 104   struct iovec *xmit;
 105   XmitDetails *xmitd;
 106   int xmitu;
 107 };
 108
 109
 110 #define CHILD_ESTATUS_STREAM   4
 111 #define CHILD_ESTATUS_NOSTREAM 5
 112
 113 static int since_connect_attempt;
 114 static int nconns;
 115 static LIST(Conn) idle, working, full;
 116
 117 static LIST(Article) *queue;
 118
 119 static void perhaps_close(int *fd) { if (*fd) { close(*fd); fd=0; } }
 120
 121
 122 /*========== making new connections ==========*/
 123
 124 static int connecting_sockets[2]= {-1,-1};
 125 static pid_t connecting_child;
 126
 127 static void report_child_status(const char *what, int status) {
 128   if (WIFEXITED(status)) {
 129     int es= WEXITSTATUS(status);
 130     if (es)
 131       warn("%s: child died with error exit status %d",es);
 132   } else if (WIFSIGNALED(status)) {
 133     int sig= WTERMSIG(status);
 134     const char *sigstr= strsignal(sig);
 135     const char *coredump= WCOREDUMP(status) ? " (core dumped)" : "";
 136     if (sigstr)
 137       warn("%s: child died due to fatal signal %s%s", what, sigstr, coredump);
 138     else
 139       warn("%s: child died due to unknown fatal signal %d%s",
 140            what, sig, coredump);
 141   } else {
 142     warn("%s: child died with unknown wait status %d", status);
 143   }
 144 }
 145
 146 static void connect_attempt_discard(void) {
 147   if (connecting_sockets[0]) {
 148     cancel_fd(loop, connecting_sockets[0], OOP_READ);
 149     cancel_fd(loop, connecting_sockets[0], OOP_EXCEPTION);
 150   }
 151   perhaps_close(&connecting_sockets[0]);
 152   perhaps_close(&connecting_sockets[1]);
 153
 154   if (connecting_child) {
 155     int status;
 156     r= kill(connecting_child, SIGKILL);
 157     if (r) sysdie("cannot kill connect child");
 158
 159     pid_t got= waitpid(connecting_child, &status, WNOHANG);
 160     if (got==-1) sysdie("cannot reap connect child");
 161
 162     if (!(WIFEXITED(status) ||
 163           (WIFSIGNALED(status) && WTERMSIG(status) == SIGKILL))) {
 164       report_child_status("connect"
 165     }
 166     connecting_child= 0;
 167   }
 168 }
 169
 170 #define PREP_DECL_MSG_CMSG(msg)                 \
 171   struct msghdr msg;                            \
 172   memset(&msg,0,sizeof(msg));                   \
 173   char msg##cbuf[CMSG_SPACE(sizeof(fd))];       \
 174   msg.msg_control= msg##cbuf;                   \
 175   msg.msg_controllen= sizeof(msg##cbuf);
 176
 177 static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
 178   Conn *conn= 0;
 179
 180   conn= xcalloc(sizeof(*conn));
 181
 182   DECL_MSG_CMSG(msg);
 183   struct cmsghdr *h= 0;
 184   ssize_t rs= recvmsg(fd, &msg, MSG_DONTWAIT);
 185   if (rs >= 0) h= CMSG_FIRSTHDR(&msg);
 186   if (!h) {
 187     int status;
 188     pid_t got= waitpid(connecting_child, &status, WNOHANG);
 189     if (got != -1) {
 190       assert(got==connecting_child);
 191       connecting_child= 0;
 192       if (WIFEXITED(status) &&
 193           (WEXITSTATUS(status) != 0
 194            WEXITSTATUS(status) != CHILD_ESTATUS_STREAM &&
 195            WEXITSTATUS(status) != CHILD_ESTATUS_NOSTREAM)) {
 196         /* child already reported the problem */
 197       } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALARM) {
 198         warn("connect: connection attempt timed out");
 199       } else if (!WIFEXITED(status)) {
 200         report_child_status("connect", status);
 201         /* that's probably the root cause then */
 202       }
 203     } else {
 204       /* child is still running apparently, report the socket problem */
 205       if (rs < 0)
 206         syswarn("connect: read from child socket failed");
 207       else if (e == OOP_EXCEPTIONN)
 208         warn("connect: unexpected exception on child socket");
 209       else
 210         warn("connect: unexpected EOF on child socket");
 211     }
 212     goto x;
 213   }
 214
 215 #define CHK(field, val)                                                   \
 216   if (h->cmsg_##field != val) {                                           \
 217     die("connect: child sent cmsg with cmsg_" #field "=%d, expected %d"); \
 218     goto x;                                                               \
 219   }
 220   CHK(level, SOL_SOCKET);
 221   CHK(type,  SCM_RIGHTS);
 222   CHK(len,   CMSG_LEN(sizeof(conn-b>fd)));
 223 #undef CHK
 224
 225   if (CMSG_NXTHDR,&msg,h) { die("connect: child sent many cmsgs"); goto x; }
 226
 227   memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
 228
 229   pid_t got= waitpid(connecting_child, &status, 0);
 230   if (got==-1) sysdie("connect: real wait for child");
 231   assert(got == connecting_child);
 232   connecting_child= 0;
 233
 234   if (!WIFEXITED(status)) { report_child_status("connect",status); goto x; }
 235   int es= WEXITSTATUS(status);
 236   switch (es) {
 237   case CHILD_ESTATUS_STREAM:    conn->stream= 1;   break;
 238   case CHILD_ESTATUS_NOSTREAM:  conn->stream= 0;   break;
 239   default:
 240     die("connect: child gave unexpected exit status %d", es);
 241   }
 242
 243   set nonblocking;
 244
 245   /* Phew! */
 246   LIST_ADDHEAD(idle, conn);
 247   notice(CN "connected %s", conn->fd, conn->stream ? "streaming" : "plain");
 248   connect_attempt_discard();
 249   process_queue();
 250   return 0;
 251
 252  x:
 253   if (conn) {
 254     perhaps_close(&conn->fd);
 255     free(conn);
 256   }
 257   connect_attempt_discard();
 258 }
 259
 260 static void connect_start() {
 261   assert(!connecting_sockets[0]);
 262   assert(!connecting_sockets[1]);
 263   assert(!connecting_child);
 264
 265   notice("starting connection attempt");
 266
 267   r= socketpair(AF_UNIX, SOCK_STREAM, 0, connecting_sockets);
 268   if (r) { syswarn("connect: cannot create socketpair for child"); goto x; }
 269
 270   connecting_child= fork();
 271   if (connecting_child==-1) { syswarn("connect: cannot fork"); goto x; }
 272
 273   if (!connecting_child) {
 274     FILE *cn_from, *cn_to;
 275     char buf[NNTP_STRLEN+100];
 276     int exitstatus= CHILD_ESTATUS_NOSTREAM;
 277
 278     put sigpipe back;
 279     close unwanted fds;
 280
 281     r= close(connecting_sockets[0]);
 282     if (r) sysdie("connect: close parent socket in child");
 283
 284     alarm(connection_setup_timeout);
 285     if (NNTPconnect(remote_host, port, &cn_from, &cn_to, buf) < 0) {
 286       if (buf[0]) {
 287         sanitise_inplace(buf);
 288         die("connect: rejected: %s", buf);
 289       } else {
 290         sysdie("connect: connection attempt failed");
 291       }
 292     }
 293     if (NNTPsendpassword(remote_host, cn_from, cn_to) < 0)
 294       sysdie("connect: authentication failed");
 295     if (try_stream) {
 296       if (fputs("MODE STREAM\r\n", cn_to) ||
 297           fflush(cn_to))
 298         sysdie("connect: could not send MODE STREAM");
 299       buf[sizeof(buf)-1]= 0;
 300       if (!fgets(buf, sizeof(buf)-1, cn_from)) {
 301         if (ferror(cn_from))
 302           sysdie("connect: could not read response to MODE STREAM");
 303         else
 304           die("connect: connection close in response to MODE STREAM");
 305       }
 306       int l= strlen(buf);
 307       assert(l>=1);
 308       if (buf[-1]!='\n') {
 309         sanitise_inplace(buf);
 310         die("connect: response to MODE STREAM is too long: %.100s...",
 311             remote_host, buf);
 312       }
 313       l--;  if (l>0 && buf[1-]=='\r') l--;
 314       buf[l]= 0;
 315       char *ep;
 316       int rcode= strtoul(buf,&ep,10);
 317       if (ep != buf[3]) {
 318         sanitise_inplace(buf);
 319         die("connect: bad response to MODE STREAM: %.50s", buf);
 320       }
 321       switch (rcode) {
 322       case 203:
 323         exitstatus= CHILD_ESTATUS_STREAM;
 324         break;
 325       case 480:
 326       case 500:
 327         break;
 328       default:
 329         sanitise_inplace(buf);
 330         warn("connect: unexpected response to MODE STREAM: %.50s", buf);
 331         exitstatus= 2;
 332         break;
 333       }
 334     }
 335     int fd= fileno(cn_from);
 336
 337     PREP_DECL_MSG_CMSG(msg);
 338     struct cmsghdr *cmsg= CMSG_FIRSTHDR(&msg);
 339     cmsg->cmsg_level= SOL_SOCKET;
 340     cmsg->cmsg_type=  SCM_RIGHTS;
 341     cmsg->cmsg_len=   CMSG_LEN(sizeof(fd));
 342     memcpy(CMSG_DATA(cmsg), &fd, sizeof(fd));
 343
 344     msg.msg_controllen= cmsg->cmsg_len;
 345     r= sendmsg(connecting_sockets[1], &msg, 0);
 346     if (r) sysdie("sendmsg failed for new connection");
 347
 348     _exit(exitstatus);
 349   }
 350
 351   r= close(connecting_sockets[1]);  connecting_sockets[1]= 0;
 352   if (r) sysdie("connect: close child socket in parent");
 353
 354   loop->on_fd(loop, connecting_sockets[0], OOP_READ,      connchild_event, 0);
 355   loop->on_fd(loop, connecting_sockets[0], OOP_EXCEPTION, connchild_event, 0);
 356   return OOP_CONTINUE;
 357
 358  x:
 359   connect_attempt_discard();
 360 }
 361
 362 /*========== overall control of article flow ==========*/
 363
 364 static void check_master_queue(void) {
 365   if (!queue.count)
 366     return;
 367
 368   if (working.head) {
 369     conn_assign_one_article(&working);
 370   } else if (idle.head) {
 371     conn_assign_one_article(&idle);
 372   } else if (nconns < maxconns && queue.count >= max_queue_per_conn &&
 373              !connecting_child && !connect_delay) {
 374     connect_delay= reconnect_delay_periods;
 375     connect_start();
 376   }
 377 }
 378
 379 static int conn_total_queued_articles(Conn *conn) {
 380   return conn->sent.count + !!conn->send + conn->queue.count;
 381 }
 382
 383 static void conn_assign_one_article(LIST(Conn) *connlist) {
 384   Conn *conn= connlist->head;
 385
 386   LIST_REMOVE(*connlist, conn);
 387   Article *art= LIST_REMHEAD(queue);
 388   LIST_ADDTAIL(conn->queue, art);
 389   LIST_ADD(*conn_determine_right_list(conn), conn);
 390
 391   check_conn_work(conn);
 392 }
 393
 394 static LIST(Conn) *conn_determine_right_list(Conn *conn) {
 395   int inqueue= conn_total_queued_articles(conn);
 396   assert(inqueue <= max_queue);
 397   if (inqueue == 0) return &idle;
 398   if (inqueue == conn->max_queue) return &full;
 399   return &working;
 400 }
 401
 402 static void check_conn_work(Conn *conn)  {
 403   void *rp;
 404   for (;;) {
 405     conn_make_some_xmits(conn);
 406
 407     void *rp= conn_write_some_xmits(conn);
 408     if (!rp) {
 409       loop->cancel_fd(loop, conn->fd, OOP_WRITE);
 410       return;
 411     } else if (rp==OOP_CONTINUE) {
 412       loop->on_fd(loop, conn->fd, OOP_WRITE;)
 413  else if (rp==OOP_HALT) {
 414       return;
 415     }
 416
 417
 418       if (
 419
 420     while (
 421
 422 }
 423
 424 /*========== article transmission ==========*/
 425
 426 static void *conn_writeable() {
 427   for (;;) {
 428
 429     if (!conn->xmitu) {
 430       perhaps_transmit_on(conn);
 431       if (!conn->xmitu) {
 432         unlink from readable;
 433         break;
 434       }
 435     }
 436
 437
 438 static void *conn_write_some_xmits(Conn *conn) {
 439   /* return values:
 440    *      0:            nothing more to write, no need to call us again
 441    *      OOP_CONTINUE: more to write but fd not writeable
 442    *      OOP_HALT:     disaster, have destroyed conn
 443    */
 444   for (;;) {
 445     int count= conn->xmitu;
 446     if (!count) return 0;
 447
 448     if (count > IOV_MAX) count= IOV_MAX;
 449     ssize_t rs= writev(conn->fd, conn->xmit, count);
 450     if (rs < 0) {
 451       if (errno == EAGAIN) return OOP_CONTINUE;
 452       syswarn(CN "write failed", conn->fd);
 453       conn_failed(conn);
 454       return OOP_HALT;
 455     }
 456     assert(rs > 0);
 457
 458     for (done=0; rs && done<xmitu; done++) {
 459       struct iovec *vp= &conn->xmit[done];
 460       XmitDetails *dp= &conn->xmitd[done];
 461       if (rs > vp->iov_len) {
 462         rs -= vp->iov_len;
 463         xmit_free(dp);
 464       } else {
 465         vp->iov_base += rs;
 466         vp->iov_len -= rs;
 467       }
 468     }
 469     int newu= conn->xmitu - done;
 470     memmove(conn->xmit,  conn->xmit  + done, newu * sizeof(*conn->xmit));
 471     memmove(conn->xmitd, conn->xmitd + done, newu * sizeof(*conn->xmitd));
 472     conn->xmitu= newu;
 473   }
 474 }
 475
 476 static void conn_make_some_xmits(Conn *conn) {
 477   for (;;) {
 478     if (conn->send) {
 479       do something about this article text;
 480       continue;
 481     }
 482
 483     if (conn->xmitu+3 > conn->xmita)
 484       /* no space for a CHECK even */
 485       break;
 486
 487     Article *art= LIST_REMHEAD(queue);
 488     if (!art) break;
 489
 490     if (art->checked || conn->nocheck) {
 491       if (conn->stream) {
 492         XMIT_LITERAL("TAKETHIS ");
 493         xmit_noalloc(art->mid, art->midlen);
 494         XMIT_LITERAL("\r\n");
 495       } else {
 496         /* we got 235 from IHAVE */
 497       }
 498       conn->send= art;
 499     } else {
 500       if (conn->stream)
 501         XMIT_LITERAL("IHAVE ");
 502       else
 503         XMIT_LITERAL("CHECK ");
 504       xmit_noalloc(art->mid, art->midlen);
 505       XMIT_LITERAL("\r\n");
 506       LIST_ADDTAIL(conn->sent, art);
 507     }
 508   }
 509 }
 510
 511
 512     if (conn->queue.head) {
 513       if (conn->queue.checked || conn->nocheck) {
 514
 515
 516       && conn->xmitu+3 <= xmita) {
 517       if (
 518       XMIT("
 519     if (conn->xmitu < xmita
 520
 521
 522   if (!queue
 523
 524
 525
 526
 527
 528     int circ_used= circ_write - circ_read;
 529     if (circ_used < 0) circ_used += CONNBUFSZ;
 530     writeable_moredata(conn, CONNBUFSZ-1 - circ_used);
 531
 532     if (conn->circ_read == conn->circ_write)
 533       return OOP_CONTINUE;
 534
 535     struct iovec iov[2];
 536     int niov= 1;
 537     iov[0].iov_base= conn->circ_buf + conn->circ_read;
 538     if (conn->circ_read > conn->circ_write) { /* wrapped */
 539       iov[0].iov_len= CONNBUFSZ - conn->circ_read;
 540       iov[1].iov_base= conn->circ_buf;
 541       iov[1].iov_len= conn->circ_write;
 542       if (niov[1].iov_len) niov= 2;
 543     } else {
 544       iov[0].iov_len= conn->circ_write - conn->circ_read;
 545     }
 546     ssize_t rs= writev(conn->fd, &iov, niov);
 547     if (rs < 0) {
 548
 549     }
 550     assert(rs > 0);
 551
 552     conn->circ_read += rs;
 553     if (conn->circ_read > CONNBUFSZ)
 554       conn->circ_read -= CONNBUFSZ;
 555   }
 556 }
 557
 558
 559 main {
 560   ignore sigpipe;
 561 };