backends/innduct.c

   1 /*
   2  * Four files full of
   3  *    token article
   4  *
   5  *   site.name_duct.lock       lock preventing multiple ducts
   6  *                                holder of this lock is "duct"
   7  * F site.name                 main feed file
   8  *                                opened/created, then written, by innd
   9  *                                read by duct
  10  *                                unlinked by duct
  11  *                                tokens blanked out by duct when processed
  12  * D site.name_duct            temporary feed file during flush (or crash)
  13  *                                hardlink created by duct
  14  *                                unlinked by duct
  15  *   site.name_duct.defer      431'd articles, still being written,
  16  *                                created, written, used by duct
  17  *   site.name_backlog.lock    lock taken out by innxmit wrapper
  18  *                                holder and its child are "xmit"
  19  *   site.name_backlog_<inum>  431'd articles, ready for innxmit
  20  *                                created (link/mv) by duct
  21  *                                read by xmit
  22  *                                unlinked by xmit
  23  *   site.name_backlog_<letters> eg
  24  *   site.name_backlog_manual
  25  *                             anything the sysadmin likes (eg, feed files
  26  *                             from old feeds to be merged into this one)
  27  *                                created (link/mv) by admin
  28  *                                read by xmit
  29  *                                unlinked by xmit
  30
  31
  32    OVERALL STATES:
  33
  34                                                                 START
  35                                                                   |
  36                                                              check D, F
  37                                                                   |
  38                           <--------------------------------------'|
  39         Nothing                            F, D both ENOENT       |
  40          F: ENOENT                                                |
  41          D: ENOENT                                                |
  42          duct: not not reading anything                           |
  43            |                                                      |
  44            |`---------------------.                               |
  45            |                      | duct times out waiting for F  |
  46            V  innd creates F      | duct exits                    |
  47            |                      V                               |
  48         Noduct                    GO TO Dropped                   |
  49          F: innd writing                                          |
  50          D: ENOENT                                                |
  51          duct: not running or not reading anything                |
  52            |                                                      |
  53            |                                                      |
  54      ,-->--+                   <---------------------------------'|
  55      |     |  duct opens F                         F exists       |
  56      |     |                                       D ENOENT       |
  57      |     V                                                      |
  58      |  Normal                                                    |
  59      |   F: innd writing, duct reading                            |
  60      |   D: ENOENT                                                |
  61      |     |                                                      |
  62      |     |  duct decides time to flush                          |
  63      |     |  duct makes hardlink                                 |
  64      |     |                                                      |
  65      |     V                            <------------------------'|
  66      |  Hardlinked                                  F==D          |
  67      |   F == D: innd writing, duct reading         both exist    |
  68      ^     |                                                      |
  69      |     |  duct unlinks F                                      |
  70      |     V                                                      |
  71      |  Moved                               <----+------------<--'|
  72      |   F: ENOENT                               |  F ENOENT      |
  73      |   D: innd writing, duct reading           |  D exists      |
  74      |     |                                     |                |
  75      |     |  duct requests flush of feed        |                |
  76      |     |   (others can too, harmlessly)      |                |
  77      |     V                                     |                |
  78      |  Flushing                                 |                |
  79      |   F: ENOENT                               |                |
  80      |   D: innd flushing, duct reading          |                |
  81      |     |                                     |                |
  82      |     |   inndcomm flush fails              |                |
  83      |     |`-------------------------->---------'                |
  84      |     |                                                      |
  85      |     |   inndcomm reports no such site                      |
  86      |     |`---------------------------------------------------- | -.
  87      |     |                                                      |  |
  88      |     |  innd finishes writing D, creates F                  |  |
  89      |     |  inndcomm reports flush successful                   |  |
  90      |     |  duct opens F too                                    |  |
  91      |     V                                                      |  |
  92      |  Flushed                                  <----------------'  |
  93      |   F: innd writing, duct reading              F!=D            /
  94      |   D: duct reading                             both exist    /
  95      |     |                                                      /
  96      |     |  duct gets to the end of D                          /
  97      |     V  duct finishes processing D                        /
  98      |     |  duct unlinks D                                   /
  99      |     |                                                  |
 100      `--<--'                                                  V
 101                                                         Dropping
 102                                                          F: ENOENT
 103                                                          D: duct reading
 104                                                               |
 105                                                               | duct finishes
 106                                                               |  processing D
 107                                                               | duct unlinks D
 108                                                               | duct exits
 109                                                               V
 110                                                         Dropped
 111                                                          F: ENOENT
 112                                                          D: ENOENT
 113                                                          duct not running
 114
 115    "duct reading" means innduct is reading the file but also
 116    overwriting processed tokens.
 117
 118  *
 119  *
 120  */
 121
 122 #define PERIOD_SECONDS 60
 123
 124 static char *feedfile;
 125 static int max_connections, max_queue_per_conn;
 126 static int connection_setup_timeout, port, try_stream;
 127 static const char *remote_host;
 128
 129 #define ISNODE(T)    T *next, *back;
 130 #define LIST(T)      struct { T *head, *tail, *tailpred; int count; }
 131
 132 #define NODE(n) ((struct node*)&(n)->head)
 133
 134 #define LIST_ADDHEAD(l,n)                                               \
 135  (list_addhead((struct list*)&(l), NODE((n))), (void)(l).count++)
 136 #define LIST_ADDTAIL(l,n)                                               \
 137  (list_addtail((struct list*)&(l), NODE((n))), (void)(l).count++)
 138
 139 #define LIST_REMHEAD(l)                                                   \
 140  ((l).count ? ((l).count--, (void*)list_remhead((struct list*)&(l))) : 0)
 141 #define LIST_REMTAIL(l)                                                   \
 142  ((l).count ? ((l).count--, (void*)list_remtail((struct list*)&(l))) : 0)
 143 #define LIST_REMOVE(l,n)                        \
 144  (list_remove(NODE((n))), (void)(l).count--)
 145 #define LIST_INSERT(l,n,pred) \
 146  (list_insert((struct list*)&(l), NODE((n)), NODE((pred))), (void)(l).count++)
 147
 148 struct Article {
 149   char *mid;
 150   int midlen;
 151   int checked, sentbody;
 152   fd and offset for blanking token or mid;
 153 };
 154
 155 #define CONNIOVS 128
 156
 157 #define CN "<%d> "
 158
 159 typedef struct Conn Conn;
 160
 161 typedef enum {
 162   Malloc, Const, Artdata;
 163 } XmitKind;
 164
 165 typedef struct {
 166   XmitKind kind;
 167   union {
 168     char *malloc_tofree;
 169     ARTHANDLE *sm_art;
 170   } info;
 171 } XmitDetails;
 172
 173 struct Conn {
 174   ISNODE(Conn);
 175   int fd, max_queue, stream;
 176   LIST(Article) queue; /* not yet told peer, or CHECK said send it */
 177   LIST(Article) sent; /* offered/transmitted - in xmit or waiting reply */
 178   struct iovec xmit[CONNIOVS];
 179   XmitDetails xmitd[CONNIOVS];
 180   int xmitu;
 181 };
 182
 183 static int filemon_init(void);
 184 static void filemon_setfile(int mainfeed_fd, const char *mainfeed_path);
 185 static void filemon_callback(void);
 186
 187
 188 #define CHILD_ESTATUS_STREAM   4
 189 #define CHILD_ESTATUS_NOSTREAM 5
 190
 191 static int since_connect_attempt;
 192 static int nconns;
 193 static LIST(Conn) idle, working, full;
 194
 195 static LIST(Article) *queue;
 196
 197 static void perhaps_close(int *fd) { if (*fd) { close(*fd); fd=0; } }
 198
 199 /*========== making new connections ==========*/
 200
 201 static int connecting_sockets[2]= {-1,-1};
 202 static pid_t connecting_child;
 203
 204 static void report_child_status(const char *what, int status) {
 205   if (WIFEXITED(status)) {
 206     int es= WEXITSTATUS(status);
 207     if (es)
 208       warn("%s: child died with error exit status %d",es);
 209   } else if (WIFSIGNALED(status)) {
 210     int sig= WTERMSIG(status);
 211     const char *sigstr= strsignal(sig);
 212     const char *coredump= WCOREDUMP(status) ? " (core dumped)" : "";
 213     if (sigstr)
 214       warn("%s: child died due to fatal signal %s%s", what, sigstr, coredump);
 215     else
 216       warn("%s: child died due to unknown fatal signal %d%s",
 217            what, sig, coredump);
 218   } else {
 219     warn("%s: child died with unknown wait status %d", status);
 220   }
 221 }
 222
 223 static void connect_attempt_discard(void) {
 224   if (connecting_sockets[0]) {
 225     cancel_fd(loop, connecting_sockets[0], OOP_READ);
 226     cancel_fd(loop, connecting_sockets[0], OOP_EXCEPTION);
 227   }
 228   perhaps_close(&connecting_sockets[0]);
 229   perhaps_close(&connecting_sockets[1]);
 230
 231   if (connecting_child) {
 232     int status;
 233     r= kill(connecting_child, SIGKILL);
 234     if (r) sysdie("cannot kill connect child");
 235
 236     pid_t got= waitpid(connecting_child, &status, WNOHANG);
 237     if (got==-1) sysdie("cannot reap connect child");
 238
 239     if (!(WIFEXITED(status) ||
 240           (WIFSIGNALED(status) && WTERMSIG(status) == SIGKILL))) {
 241       report_child_status("connect"
 242     }
 243     connecting_child= 0;
 244   }
 245 }
 246
 247 #define PREP_DECL_MSG_CMSG(msg)                 \
 248   struct msghdr msg;                            \
 249   memset(&msg,0,sizeof(msg));                   \
 250   char msg##cbuf[CMSG_SPACE(sizeof(fd))];       \
 251   msg.msg_control= msg##cbuf;                   \
 252   msg.msg_controllen= sizeof(msg##cbuf);
 253
 254 static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
 255   Conn *conn= 0;
 256
 257   conn= xcalloc(sizeof(*conn));
 258
 259   DECL_MSG_CMSG(msg);
 260   struct cmsghdr *h= 0;
 261   ssize_t rs= recvmsg(fd, &msg, MSG_DONTWAIT);
 262   if (rs >= 0) h= CMSG_FIRSTHDR(&msg);
 263   if (!h) {
 264     int status;
 265     pid_t got= waitpid(connecting_child, &status, WNOHANG);
 266     if (got != -1) {
 267       assert(got==connecting_child);
 268       connecting_child= 0;
 269       if (WIFEXITED(status) &&
 270           (WEXITSTATUS(status) != 0
 271            WEXITSTATUS(status) != CHILD_ESTATUS_STREAM &&
 272            WEXITSTATUS(status) != CHILD_ESTATUS_NOSTREAM)) {
 273         /* child already reported the problem */
 274       } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALARM) {
 275         warn("connect: connection attempt timed out");
 276       } else if (!WIFEXITED(status)) {
 277         report_child_status("connect", status);
 278         /* that's probably the root cause then */
 279       }
 280     } else {
 281       /* child is still running apparently, report the socket problem */
 282       if (rs < 0)
 283         syswarn("connect: read from child socket failed");
 284       else if (e == OOP_EXCEPTIONN)
 285         warn("connect: unexpected exception on child socket");
 286       else
 287         warn("connect: unexpected EOF on child socket");
 288     }
 289     goto x;
 290   }
 291
 292 #define CHK(field, val)                                                   \
 293   if (h->cmsg_##field != val) {                                           \
 294     die("connect: child sent cmsg with cmsg_" #field "=%d, expected %d"); \
 295     goto x;                                                               \
 296   }
 297   CHK(level, SOL_SOCKET);
 298   CHK(type,  SCM_RIGHTS);
 299   CHK(len,   CMSG_LEN(sizeof(conn-b>fd)));
 300 #undef CHK
 301
 302   if (CMSG_NXTHDR,&msg,h) { die("connect: child sent many cmsgs"); goto x; }
 303
 304   memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
 305
 306   pid_t got= waitpid(connecting_child, &status, 0);
 307   if (got==-1) sysdie("connect: real wait for child");
 308   assert(got == connecting_child);
 309   connecting_child= 0;
 310
 311   if (!WIFEXITED(status)) { report_child_status("connect",status); goto x; }
 312   int es= WEXITSTATUS(status);
 313   switch (es) {
 314   case CHILD_ESTATUS_STREAM:    conn->stream= 1;   break;
 315   case CHILD_ESTATUS_NOSTREAM:  conn->stream= 0;   break;
 316   default:
 317     die("connect: child gave unexpected exit status %d", es);
 318   }
 319
 320   set nonblocking;
 321
 322   /* Phew! */
 323   LIST_ADDHEAD(idle, conn);
 324   notice(CN "connected %s", conn->fd, conn->stream ? "streaming" : "plain");
 325   connect_attempt_discard();
 326   check_master_queue();
 327   return 0;
 328
 329  x:
 330   if (conn) {
 331     perhaps_close(&conn->fd);
 332     free(conn);
 333   }
 334   connect_attempt_discard();
 335 }
 336
 337 static void connect_start() {
 338   assert(!connecting_sockets[0]);
 339   assert(!connecting_sockets[1]);
 340   assert(!connecting_child);
 341
 342   notice("starting connection attempt");
 343
 344   r= socketpair(AF_UNIX, SOCK_STREAM, 0, connecting_sockets);
 345   if (r) { syswarn("connect: cannot create socketpair for child"); goto x; }
 346
 347   connecting_child= fork();
 348   if (connecting_child==-1) { syswarn("connect: cannot fork"); goto x; }
 349
 350   if (!connecting_child) {
 351     FILE *cn_from, *cn_to;
 352     char buf[NNTP_STRLEN+100];
 353     int exitstatus= CHILD_ESTATUS_NOSTREAM;
 354
 355     put sigpipe back;
 356     close unwanted fds;
 357
 358     r= close(connecting_sockets[0]);
 359     if (r) sysdie("connect: close parent socket in child");
 360
 361     alarm(connection_setup_timeout);
 362     if (NNTPconnect(remote_host, port, &cn_from, &cn_to, buf) < 0) {
 363       if (buf[0]) {
 364         sanitise_inplace(buf);
 365         die("connect: rejected: %s", buf);
 366       } else {
 367         sysdie("connect: connection attempt failed");
 368       }
 369     }
 370     if (NNTPsendpassword(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) ||
 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[-1]!='\n') {
 386         sanitise_inplace(buf);
 387         die("connect: response to MODE STREAM is too long: %.100s...",
 388             remote_host, buf);
 389       }
 390       l--;  if (l>0 && buf[1-]=='\r') l--;
 391       buf[l]= 0;
 392       char *ep;
 393       int rcode= strtoul(buf,&ep,10);
 394       if (ep != buf[3]) {
 395         sanitise_inplace(buf);
 396         die("connect: bad response to MODE STREAM: %.50s", buf);
 397       }
 398       switch (rcode) {
 399       case 203:
 400         exitstatus= CHILD_ESTATUS_STREAM;
 401         break;
 402       case 480:
 403       case 500:
 404         break;
 405       default:
 406         sanitise_inplace(buf);
 407         warn("connect: unexpected response to MODE STREAM: %.50s", buf);
 408         exitstatus= 2;
 409         break;
 410       }
 411     }
 412     int fd= fileno(cn_from);
 413
 414     PREP_DECL_MSG_CMSG(msg);
 415     struct cmsghdr *cmsg= CMSG_FIRSTHDR(&msg);
 416     cmsg->cmsg_level= SOL_SOCKET;
 417     cmsg->cmsg_type=  SCM_RIGHTS;
 418     cmsg->cmsg_len=   CMSG_LEN(sizeof(fd));
 419     memcpy(CMSG_DATA(cmsg), &fd, sizeof(fd));
 420
 421     msg.msg_controllen= cmsg->cmsg_len;
 422     r= sendmsg(connecting_sockets[1], &msg, 0);
 423     if (r) sysdie("sendmsg failed for new connection");
 424
 425     _exit(exitstatus);
 426   }
 427
 428   r= close(connecting_sockets[1]);  connecting_sockets[1]= 0;
 429   if (r) sysdie("connect: close child socket in parent");
 430
 431   loop->on_fd(loop, connecting_sockets[0], OOP_READ,      connchild_event, 0);
 432   loop->on_fd(loop, connecting_sockets[0], OOP_EXCEPTION, connchild_event, 0);
 433   return OOP_CONTINUE;
 434
 435  x:
 436   connect_attempt_discard();
 437 }
 438
 439 /*========== overall control of article flow ==========*/
 440
 441 static void conn_check_work(Conn *conn);
 442
 443 static void check_master_queue(void) {
 444   try reading current feed file;
 445
 446   if (!queue.count)
 447     return;
 448
 449   Conn *last_assigned=0;
 450   for (;;) {
 451     if (working.head) {
 452       conn_assign_one_article(&working, &last_assigned);
 453     } else if (idle.head) {
 454       conn_assign_one_article(&idle, &last_assigned);
 455     } else if (nconns < maxconns && queue.count >= max_queue_per_conn &&
 456                !connecting_child && !connect_delay) {
 457       connect_delay= reconnect_delay_periods;
 458       connect_start();
 459     } else {
 460       break;
 461     }
 462   }
 463   conn_check_work(last_assigned);
 464 }
 465
 466 static void conn_assign_one_article(LIST(Conn) *connlist,
 467                                     Conn **last_assigned) {
 468   Conn *conn= connlist->head;
 469
 470   LIST_REMOVE(*connlist, conn);
 471   Article *art= LIST_REMHEAD(queue);
 472   LIST_ADDTAIL(conn->queue, art);
 473   LIST_ADD(*conn_determine_right_list(conn), conn);
 474
 475   /* This slightly odd arrangement is so that we call conn_check_work
 476    * once after filling the queue for a new connection in
 477    * check_master_queue, rather than for each article. */
 478   if (conn != *last_assigned && *last_assigned)
 479     conn_check_work(*last_assigned);
 480   *last_assigned= conn;
 481 }
 482
 483 static int conn_total_queued_articles(Conn *conn) {
 484   return conn->sent.count + conn->queue.count;
 485 }
 486
 487 static LIST(Conn) *conn_determine_right_list(Conn *conn) {
 488   int inqueue= conn_total_queued_articles(conn);
 489   assert(inqueue <= max_queue);
 490   if (inqueue == 0) return &idle;
 491   if (inqueue == conn->max_queue) return &full;
 492   return &working;
 493 }
 494
 495 static void *conn_writeable(oop_source *l, int fd, int ev, void *u) {
 496   check_conn_work(u);
 497   return OOP_CONTINUE;
 498 }
 499
 500 static void conn_check_work(Conn *conn)  {
 501   void *rp= 0;
 502   for (;;) {
 503     conn_make_some_xmits(conn);
 504     if (!conn->xmitu) {
 505       loop->cancel_fd(loop, conn->fd, OOP_WRITE);
 506       return;
 507     }
 508
 509     void *rp= conn_write_some_xmits(conn);
 510     if (rp==OOP_CONTINUE) {
 511       loop->on_fd(loop, conn->fd, OOP_WRITE, conn_writeable, conn);
 512       return;
 513     } else if (rp==OOP_HALT) {
 514       return;
 515     } else if (!rp) {
 516       /* transmitted everything */
 517     } else {
 518       abort();
 519     }
 520   }
 521 }
 522
 523 /*========== article transmission ==========*/
 524
 525 static void *conn_write_some_xmits(Conn *conn) {
 526   /* return values:
 527    *      0:            nothing more to write, no need to call us again
 528    *      OOP_CONTINUE: more to write but fd not writeable
 529    *      OOP_HALT:     disaster, have destroyed conn
 530    */
 531   for (;;) {
 532     int count= conn->xmitu;
 533     if (!count) return 0;
 534
 535     if (count > IOV_MAX) count= IOV_MAX;
 536     ssize_t rs= writev(conn->fd, conn->xmit, count);
 537     if (rs < 0) {
 538       if (errno == EAGAIN) return OOP_CONTINUE;
 539       syswarn(CN "write failed", conn->fd);
 540       conn_failed(conn);
 541       return OOP_HALT;
 542     }
 543     assert(rs > 0);
 544
 545     for (done=0; rs && done<xmitu; done++) {
 546       struct iovec *vp= &conn->xmit[done];
 547       XmitDetails *dp= &conn->xmitd[done];
 548       if (rs > vp->iov_len) {
 549         rs -= vp->iov_len;
 550         xmit_free(dp);
 551       } else {
 552         vp->iov_base += rs;
 553         vp->iov_len -= rs;
 554       }
 555     }
 556     int newu= conn->xmitu - done;
 557     memmove(conn->xmit,  conn->xmit  + done, newu * sizeof(*conn->xmit));
 558     memmove(conn->xmitd, conn->xmitd + done, newu * sizeof(*conn->xmitd));
 559     conn->xmitu= newu;
 560   }
 561 }
 562
 563 static void conn_make_some_xmits(Conn *conn) {
 564   for (;;) {
 565     if (conn->xmitu+5 > CONNIOVS)
 566       break;
 567
 568     Article *art= LIST_REMHEAD(queue);
 569     if (!art) break;
 570
 571     if (art->checked || conn->nocheck) {
 572       /* actually send it */
 573
 574       ARTHANDLE *artdata= SMretrieve(somehow);
 575
 576       if (conn->stream) {
 577         if (artdata) {
 578           XMIT_LITERAL("TAKETHIS ");
 579           xmit_noalloc(art->mid, art->midlen);
 580           XMIT_LITERAL("\r\n");
 581           xmit_artbody(artdata);
 582         }
 583       } else {
 584         /* we got 235 from IHAVE */
 585         if (artdata) {
 586           xmit_artbody(artdata);
 587         } else {
 588           XMIT_LITERAL(".\r\n");
 589         }
 590       }
 591       art->sent= 1;
 592       LIST_ADDTAIL(conn->sent, art);
 593
 594     } else {
 595       /* check it */
 596
 597       if (conn->stream)
 598         XMIT_LITERAL("IHAVE ");
 599       else
 600         XMIT_LITERAL("CHECK ");
 601       xmit_noalloc(art->mid, art->midlen);
 602       XMIT_LITERAL("\r\n");
 603
 604       LIST_ADDTAIL(conn->sent, art);
 605     }
 606   }
 607 }
 608
 609 /*========== responses from peer ==========*/
 610
 611 static const oop_rd_style peer_rd_style= {
 612   OOP_RD_DELIM_STRIP, '\n',
 613   OOP_RD_NUL_FORBID,
 614   OOP_RD_SHORTREC_FORBID
 615 };
 616
 617 static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_event ev,
 618                         const char *errmsg, int errnoval,
 619                         const char *data, size_t recsz, void *conn_v) {
 620   Conn *conn= conn_v;
 621
 622   if (ev == OOP_RD_EOF) {
 623     warn("unexpected EOF from peer");
 624     conn_failed(conn);
 625     return;
 626   }
 627   assert(ev == OOP_RD_OK);
 628
 629   char *ep;
 630   unsigned long code= strtoul(data, &ep, 10);
 631   if (ep != data+3 || *ep != ' ' || data[0]=='0') {
 632     char sanibuf[100];
 633     const char *p= data;
 634     char *q= sanibuf;
 635     *q++= '`';
 636     for (;;) {
 637       if (q > sanibuf+sizeof(sanibuf)-8) { strcpy(q,"..."); break; }
 638       int c= *p++;
 639       if (!c) { *q++= '\''; break; }
 640       if (c>=' ' && c<=126 && c!='\\') { *q++= c; continue; }
 641       sprintf(q,"\\x%02x",c);
 642       q += 4;
 643     }
 644     warn("badly formatted response from peer: %s", sanibuf);
 645     conn_failed(conn);
 646     return;
 647   }
 648
 649   if (conn->quitting) {
 650     if (code!=205) {
 651       warn("peer gave failure response to QUIT: %s", sani);
 652       conn_failed(conn);
 653       return;
 654     }
 655     conn close ok;
 656     return;
 657   }
 658
 659   switch (code) {
 660   case 438: /* CHECK says they have it */
 661   case 435: /* IHAVE says they have it */
 662     ARTICLE_DEALTWITH(1,unwanted);
 663     break;
 664
 665   case 238: /* CHECK says send it */
 666   case 335: /* IHAVE says send it */
 667     count_checkedwanted++;
 668     Article *art= LIST_REMHEAD(conn->sent);
 669     art->checked= 1;
 670     LIST_ADDTAIL(conn->queue);
 671     break;
 672
 673   case 235: /* IHAVE says thanks */
 674   case 239: /* TAKETHIS says thanks */
 675     ARTICLE_DEALTWITH(1,accepted);
 676     break;
 677
 678   case 439: /* TAKETHIS says rejected */
 679   case 437: /* IHAVE says rejected */
 680     ARTICLE_DEALTWITH(1,rejected);
 681     break;
 682
 683   case 431: /* CHECK or TAKETHIS says try later */
 684   case 436: /* IHAVE says try later */
 685     ARTICLE_DEALTWITH(0,deferred);
 686     break;
 687
 688   case 400: warn("peer has stopped accepting articles: %s", sani); goto failed;
 689   case 503: warn("peer timed us out: %s", sani);                   goto failed;
 690   default:  warn("peer sent unexpected message: %s", sani);
 691   failed:
 692     conn_failed(conn);
 693     return OOP_CONTINUE;;
 694   }
 695
 696   return OOP_CONTINUE;
 697 }
 698
 699 /*========== monitoring of input file ==========*/
 700
 701 /*---------- tailing input file ----------*/
 702
 703
 704
 705 /*---------- filemon implemented with inotify ----------*/
 706
 707 #if defined(HAVE_INOTIFY) && !defined(HAVE_FILEMON)
 708 #define HAVE_FILEMON
 709
 710 #include <linux/inotify.h>
 711
 712 static int filemon_inotify_fd;
 713 static int filemon_inotify_wdmax;
 714 static InputFile **filemon_inotify_wd2ipf;
 715
 716 typedef struct Filemon_Perfile {
 717   int wd;
 718 } Filemon_Inotify_Perfile;
 719
 720 static void filemon_startfile(InputFile *ipf) {
 721   int wd= inotify_add_watch(filemon_inotify_fd, ipf->path, IN_MODIFY);
 722   if (wd < 0) sysdie("inotify_add_watch %s", ipf->path);
 723
 724   if (wd >= filemon_inotify_wdmax) {
 725     int newmax= wd+2;
 726     filemon_inotify_wd= xrealloc(filemon_inotify_wd2ipf,
 727                                  sizeof(*filemon_inotify_wd2ipf) * newmax);
 728     memset(filemon_inotify_wd2ipf + filemon_inotify_wdmax, 0,
 729            sizeof(*filemon_inotify_wd2ipf) * (newmax - filemon_inotify_wdmax));
 730     filemon_inotify_wdmax= newmax;
 731   }
 732
 733   assert(!filemon_inotify_wd2ipf[wd]);
 734   filemon_inotify_wd2ipf[wd]= ipf;
 735
 736   assert(!ipf->filemon);
 737   ipf->filemon= xmalloc(sizeof(*ipf->filemon));
 738   ipf->filemon->wd= wd;
 739 }
 740
 741 static void filemon_stopfile(InputFile *ipf) {
 742   int wd= ipf->filemon->wd;
 743   int r= inotify_rm_watch(filemon_inotify_fd, filemon_inotify_wd);
 744   if (r) sysdie("inotify_rm_watch");
 745   filemon_inotify_wd2ipf[wd]= 0;
 746   free(ipf->filemon);
 747   ipf->filemon= 0;
 748 }
 749
 750 static void *filemon_inotify_readable(oop_source *lp, int fd,
 751                                       oop_event e, void *u) {
 752   struct inotify_event iev;
 753   for (;;) {
 754     int r= read(filemon_inotify_fd, &iev, sizeof(iev));
 755     if (r==-1) {
 756       if (errno==EAGAIN) break;
 757       sysdie("read from inotify master");
 758     } else if (r==sizeof(iev)) {
 759       assert(iev.wd >= 0 && iev.wd < filemon_inotify_wdmax);
 760     } else {
 761       die("inotify read %d bytes wanted struct of %d", r, (int)sizeof(iev));
 762     }
 763     filemon_callback(filemon_inotify_wd2ipf[iev.wd]);
 764   }
 765   return OOP_CONTINUE;
 766 }
 767
 768 static int filemon_init(void) {
 769   filemon_inotify_fd= inotify_init();
 770   if (filemon_inotify_fd<0) {
 771     syswarn("could not initialise inotify: inotify_init failed");
 772     return 0;
 773   }
 774   set nonblock;
 775   loop->on_fd(loop, filemon_inotify_fd, OOP_READ, filemon_inotify_readable);
 776
 777   return 1;
 778 }
 779
 780 #endif /* HAVE_INOTIFY && !HAVE_FILEMON *//
 781
 782 /*---------- filemon dummy implementation ----------*/
 783
 784 #if !defined(HAVE_FILEMON)
 785
 786 static int filemon_init(void) { return 0; }
 787 static void filemon_startfile(InputFile *ipf) { }
 788 static void filemon_stopfile(InputFile *ipf) { }
 789
 790 #endif
 791
 792 /*========== interaction with innd ==========*/
 793
 794 /* See official state diagram at top of file.  We implement
 795  * this as follows:
 796  *
 797            ================
 798             WAITING
 799            [Nothing/Noduct]
 800             poll for F
 801            ================
 802                 |
 803                 |     TIMEOUT
 804                 |`--------------------------.
 805                 |                           | install defer as backlog
 806      ,--------->|                           | exit
 807      |          | OPEN F SUCCEEDS           V
 808      |          V                         =========
 809      |     ========                        (ESRCH)
 810      |      NORMAL                        [Dropped]
 811      |     [Normal]                       =========
 812      |      read F
 813      |     ========
 814      |          |
 815      |          | F IS SO BIG WE SHOULD FLUSH
 816      ^          | hardlink F to D
 817      |     [Hardlinked]
 818      |          | unlink F
 819      |          | our handle onto F is now onto D
 820      |     [Moved]
 821      |          |
 822      |          |<---------------------------------------------------.
 823      |          |                                                    |
 824      |          | spawn inndcomm flush                               |
 825      |          V                                                    |
 826      |     ==========                                                |
 827      |      FLUSHING                                                 |
 828      |     [Flushing]                                                |
 829      |      read D                                                   |
 830      |     ==========                                                |
 831      |          |                                                    |
 832      |          |   INNDCOMM FLUSH FAILS                             ^
 833      |          |`----------------------->--------.                  |
 834      |          |                                 |                  |
 835      |          |   NO SUCH SITE                  V                  |
 836      ^          |`----------------.            =========             |
 837      |          |                 |            FLUSHFAIL             |
 838      |          |                 V            [Moved]               |
 839      |          |            ==========        read D                |
 840      |          |             DROPPING         =========             |
 841      |          |            [Dropping]           |                  |
 842      |          |             read D              | TIME TO RETRY    |
 843      |          |            ==========           `------------------'
 844      |          | FLUSH OK        |
 845      |          | open F          | AT EOF OF D AND ALL PROCESSED
 846      |          V                 | install defer as backlog
 847      |     ==========             | unlink D
 848      |      FLUSHED               | exit
 849      |     [Flushed]              V
 850      |      read D, F         ==========
 851      |     ==========          (ESRCH)
 852      |          |             [Droppped]
 853      |          |             ==========
 854      |          |
 855      |          | AT EOF OF D AND ALL D PROCESSED
 856      ^          V unlink D
 857      |          | close D
 858      |          | install defer as backlog
 859      |          | start new defer
 860      |          |
 861      `----------'
 862
 863  *
 864  *  duct state
 865  *   WAITING
 866  *   NORMAL
 867  *   FLUSHING
 868  *   FLUSHED
 869  *   FLUSHFAIL
 870  *   DROPPING
 871  */
 872
 873 static char *path_ductlock, *path_duct, *path_ductdefer;
 874
 875 typedef struct {
 876   /* This is an instance of struct oop_readable */
 877   struct oop_readable readable;
 878   oop_readable_call *readable_callback;
 879
 880   int fd;
 881   const char *path; /* ptr copy of path_<foo> or feedfile */
 882   struct Filemon_Perfile *filemon;
 883
 884   oop_read *rd;
 885   long inprogress; /* no. of articles read but not processed */
 886 } InputFile;
 887
 888 static void statemc_init(void) {
 889   path_ductlock=  xasprintf("%s_duct.lock",  feedfile);
 890   path_duct=      xasprintf("%s_duct",       feedfile);
 891   path_ductdefer= xasprintf("%s_duct.defer", feedfile);
 892
 893   int lockfd= open(path_ductlock, O_CREAT|O_RDWR, 0600);
 894   if (lockfd<0) sysdie("open lockfile %s", path_ductlock);
 895
 896   struct flock fl;
 897   memset(&fl,0,sizeof(fl));
 898   fl.l_type= F_WRLCK;
 899   fl.l_whence= SEEK_SET;
 900   r= fcntl(lockfd, F_SETLK, &fl);
 901   if (r==-1) {
 902     if (errno==EACCES || errno==EAGAIN)
 903       die("another duct holds the lockfile");
 904     sysdie("fcntl F_SETLK lockfile %s", path_ductlock);
 905   }
 906 }
 907
 908 static void statemc_poll(void) {
 909   if (tailing_fd>=0) return;
 910
 911   int d_fd= open(path_duct, O_RDWR);
 912   if (d_fd<0)
 913     if (errno!=ENOENT) sysdie("open duct file %s", path_duct);
 914
 915   int f_fd= open(feedfile, O_RDWR);
 916   if (f_fd<0)
 917     if (errno!=ENOENT) sysdie("open feed file %s", feedfile);
 918
 919   if (d_fd<0) {
 920     if (f_fd>=0)
 921       start_tailing(f_fd);
 922     return;
 923   }
 924
 925
 926
 927 /*========== main program ==========*/
 928
 929 #define EVERY(what, interval, body)                                          \
 930   static const struct timeval what##_timeout = { 5, 0 };                     \
 931   static void what##_schedule(void);                                         \
 932   static void *what##_timedout(oop_source *lp, struct timeval tv, void *u) { \
 933     { body }                                                                 \
 934     what##_schedule();                                                       \
 935   }                                                                          \
 936   static void what##_schedule(void) {                                        \
 937     loop->on_time(loop, what##_timeout, what##_timedout, 0);                 \
 938   }
 939
 940 EVERY(filepoll, {5,0}, { check_master_queue(); })
 941
 942 EVERY(period, {PERIOD_SECONDS,0}, {
 943   if (connect_delay) connect_delay--;
 944   statemc_poll();
 945   check_master_queue();
 946 });
 947
 948 main {
 949   ignore sigpipe;
 950   if (!filemon_init())
 951     filepoll_schedule();
 952   period_schedule();
 953 };