chiark / gitweb /
Do not assume that loop->on_fd is idempotent
[inn-innduct.git] / backends / innduct.c
index b5fb0176cf5e8adc29ca33c5f5161994ff521ac0..29fe5c10ccdfe294d445269648900fc6edf673df 100644 (file)
@@ -365,7 +365,7 @@ static oop_rd_call peer_rd_err, peer_rd_ok;
 static const char *sitename, *remote_host;
 static const char *feedfile, *path_run, *path_cli, *path_cli_dir;
 static int quiet_multiple=0;
-static int become_daemon=1, try_filemon=1;
+static int interactive=0, try_filemon=1;
 static int try_stream=1;
 static int port=119;
 static const char *inndconffile;
@@ -373,7 +373,7 @@ static const char *inndconffile;
 static int max_connections=10;
 static int max_queue_per_conn=200;
 static int target_max_feedfile_size=100000;
-static int period_seconds=60;
+static int period_seconds=30;
 static int filepoll_seconds=5;
 static int max_queue_per_ipf=-1;
 
@@ -511,7 +511,9 @@ struct Conn {
   ISNODE(Conn);
   int fd; /* may be 0, meaning closed (during construction/destruction) */
   oop_read *rd; /* likewise */
-  int max_queue, stream, quitting;
+  int oopwriting; /* since on_fd is not idempotent */
+  int max_queue, stream;
+  const char *quitting;
   int since_activity; /* periods */
   ArticleList waiting; /* not yet told peer */
   ArticleList priority; /* peer says send it now */
@@ -532,7 +534,7 @@ static char *globpat_backlog;
 static pid_t self_pid;
 static int *lowvol_perperiod;
 static int lowvol_circptr;
-static int lowvol_total;
+static int lowvol_total; /* does not include current period */
 
 /* statemc_init initialises */
 static StateMachineState sms;
@@ -553,7 +555,7 @@ int simulate_flush= -1;
 static void logcore(int sysloglevel, const char *fmt, ...) PRINTF(2,3);
 static void logcore(int sysloglevel, const char *fmt, ...) {
   VA;
-  if (become_daemon) {
+  if (interactive < 2) {
     vsyslog(sysloglevel,fmt,al);
   } else {
     if (self_pid) fprintf(stderr,"[%lu] ",(unsigned long)self_pid);
@@ -1175,7 +1177,7 @@ static void conn_idle_close(Conn *conn, const char *why) {
     assert(r<=todo);
     todo -= r;
     if (!todo) {
-      conn->quitting= 1;
+      conn->quitting= why;
       conn->since_activity= 0;
       debug("C%d is idle (%s), quitting", conn->fd, why);
       break;
@@ -1190,9 +1192,10 @@ static void conn_idle_close(Conn *conn, const char *why) {
 static void check_idle_conns(void) {
   Conn *conn;
 
-  int verylowvol= lowvol_perperiod[lowvol_circptr];
+  int volthisperiod= lowvol_perperiod[lowvol_circptr];
   lowvol_circptr++;
   lowvol_circptr %= lowvol_periods;
+  lowvol_total += volthisperiod;
   lowvol_total -= lowvol_perperiod[lowvol_circptr];
   lowvol_perperiod[lowvol_circptr]= 0;
 
@@ -1205,7 +1208,8 @@ static void check_idle_conns(void) {
 
     /* We need to shut this down */
     if (conn->quitting)
-      connfail(conn,"timed out waiting for response to QUIT");
+      connfail(conn,"timed out waiting for response to QUIT (%s)",
+              conn->quitting);
     else if (conn->sent.count)
       connfail(conn,"timed out waiting for responses");
     else if (conn->waiting.count || conn->priority.count)
@@ -1220,9 +1224,9 @@ static void check_idle_conns(void) {
   }
 
   conn= LIST_HEAD(conns);
-  if (!verylowvol &&
+  if (!volthisperiod &&
       conns.count==1 &&
-      lowvol_total+verylowvol < lowvol_thresh &&
+      lowvol_total < lowvol_thresh &&
       !conn_busy(conn))
     conn_idle_close(conn, "low volume");
 }  
@@ -1372,7 +1376,7 @@ static void connect_start(void) {
   assert(!connecting_fdpass_sock);
 
   info("starting connection attempt");
-  int ok_reconnect_delay_periods= reconnect_delay_periods;
+  int ok_until_connect= until_connect;
   reconnect_blocking_event();
 
   int socks[2];
@@ -1468,7 +1472,7 @@ static void connect_start(void) {
   on_fd_read_except(connecting_fdpass_sock, connchild_event);
 
   if (!conns.count)
-    reconnect_delay_periods= ok_reconnect_delay_periods;
+    until_connect= ok_until_connect;
 }
 
 /*---------- assigning articles to conns, and transmitting ----------*/
@@ -1544,12 +1548,16 @@ static void conn_maybe_write(Conn *conn)  {
     conn_make_some_xmits(conn);
     if (!conn->xmitu) {
       loop->cancel_fd(loop, conn->fd, OOP_WRITE);
+      conn->oopwriting= 0;
       return;
     }
 
     void *rp= conn_write_some_xmits(conn);
     if (rp==OOP_CONTINUE) {
-      loop->on_fd(loop, conn->fd, OOP_WRITE, conn_writeable, conn);
+      if (!conn->oopwriting) {
+       loop->on_fd(loop, conn->fd, OOP_WRITE, conn_writeable, conn);
+       conn->oopwriting= 1;
+      }
       return;
     } else if (rp==OOP_HALT) {
       return;
@@ -1639,6 +1647,7 @@ static void autodefer_input_file_articles(InputFile *ipf) {
 }
 
 static void autodefer_input_file(InputFile *ipf) {
+  static const char *const abandon= "stuck";
   ipf->autodefer= 0;
 
   autodefer_input_file_articles(ipf);
@@ -1649,11 +1658,11 @@ static void autodefer_input_file(InputFile *ipf) {
       if (has_article_in(&walk->waiting,  ipf) ||
          has_article_in(&walk->priority, ipf) ||
          has_article_in(&walk->sent,     ipf))
-       walk->quitting= -1;
+       walk->quitting= abandon;
     }
     while (ipf->inprogress) {
       FOR_CONN(walk)
-       if (walk->quitting < 0) goto found;
+       if (walk->quitting == abandon) goto found;
       abort(); /* where are they ?? */
 
     found:
@@ -1951,11 +1960,12 @@ static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_rd_event ev,
 
   if (conn->quitting) {
     if (code!=205 && code!=400) {
-      connfail(conn, "peer gave unexpected response to QUIT: %s", sani);
+      connfail(conn, "peer gave unexpected response to QUIT (%s): %s",
+              conn->quitting, sani);
     } else {
       LIST_REMOVE(conns,conn);
-      notice("C%d (now %d) idle connection closed by us",
-            conn->fd, conns.count);
+      notice("C%d (now %d) idle connection closed (%s)",
+            conn->fd, conns.count, conn->quitting);
       assert(!busy);
       conn_dispose(conn);
     }
@@ -3307,11 +3317,11 @@ static void period(void) {
   debug("PERIOD"
        " sms=%s[%d] conns=%d until_connect=%d"
        " input_files main:%s flushing:%s backlog:%s[%d]"
-       " children connecting=%ld inndcomm=%ld"
+       " children connecting=%ld inndcomm=%ld lowvol_total=%d"
        ,
        sms_names[sms], until_flush, conns.count, until_connect,
        dipf_main, dipf_flushing, dipf_backlog, until_backlog_nextscan,
-       (long)connecting_child, (long)inndcomm_child
+       (long)connecting_child, (long)inndcomm_child, lowvol_total
        );
 
   free(dipf_main);
@@ -3402,6 +3412,17 @@ CCMD(dump) {
   DUMPV("%d", , cli_master);
   fprintf(f,"\n");
 
+  fprintf(f,"lowvol");
+  DUMPV("%d", , lowvol_circptr);
+  DUMPV("%d", , lowvol_total);
+  fprintf(f,":");
+  for (i=0; i<lowvol_periods; i++) {
+    fprintf(f," ");
+    if (i==lowvol_circptr) fprintf(f,"*");
+    fprintf(f,"%d",lowvol_perperiod[i]);
+  }
+  fprintf(f,"\n");
+
   fprintf(f,"filemon ");
   filemon_method_dump_info(f);
 
@@ -3416,7 +3437,7 @@ CCMD(dump) {
 
     fprintf(f,"C%d",conn->fd);
     DUMPV("%p",conn->,rd);             DUMPV("%d",conn->,max_queue);
-    DUMPV("%d",conn->,stream);         DUMPV("%d",conn->,quitting);
+    DUMPV("%d",conn->,stream);         DUMPV("\"%s\"",conn->,quitting);
     DUMPV("%d",conn->,since_activity);
     fprintf(f,"\n");
 
@@ -3465,7 +3486,7 @@ static void vbadusage(const char *fmt, va_list al) {
   fprintf(stderr, "bad usage: %s\n"
          "say --help for help, or read the manpage\n",
          m);
-  if (become_daemon)
+  if (interactive < 2)
     syslog(LOG_CRIT,"innduct: invoked with bad usage: %s",m);
   exit(8);
 }
@@ -3625,7 +3646,8 @@ static void help(const Option *o, const char *val);
 static const Option innduct_options[]= {
 {'f',"feedfile",         "F",     &feedfile,                 op_string      },
 {'q',"quiet-multiple",   0,       &quiet_multiple,           op_setint, 1   },
-{0,"no-daemon",          0,       &become_daemon,            op_setint, 0   },
+{0,"no-daemon",          0,       &interactive,              op_setint, 1   },
+{0,"interactive",        0,       &interactive,              op_setint, 2   },
 {0,"no-streaming",       0,       &try_stream,               op_setint, 0   },
 {0,"no-filemon",         0,       &try_filemon,              op_setint, 0   },
 {'C',"inndconf",         "F",     &inndconffile,             op_string      },
@@ -3770,6 +3792,7 @@ int main(int argc, char **argv) {
     lowvol_perperiod[i]= lowvol_thresh;
     lowvol_total += lowvol_thresh;
   }
+  lowvol_total -= lowvol_thresh;
 
   /* set things up */
 
@@ -3785,13 +3808,17 @@ int main(int argc, char **argv) {
 
   LIST_INIT(conns);
 
-  if (become_daemon) {
+  if (interactive < 1) {
     int i;
     for (i=3; i<255; i++)
       /* do this now before we open syslog, etc. */
       close(i);
+  }
+
+  if (interactive < 2)
     openlog("innduct",LOG_NDELAY|LOG_PID,LOG_NEWS);
 
+  if (interactive < 1) {
     int null= open("/dev/null",O_RDWR);
     if (null<0) sysfatal("failed to open /dev/null");
     dup2(null,0);
@@ -3825,7 +3852,7 @@ int main(int argc, char **argv) {
   r= SMsetup(SM_PREOPEN, &val); if (!r) warn("SMsetup SM_PREOPEN failed");
   r= SMinit(); if (!r) die("storage manager initialisation (SMinit) failed");
 
-  if (!become_daemon)
+  if (interactive >= 2)
     cli_stdio();
 
   cli_init();