chiark / gitweb /
comment for dealing with wedge problems
[inn-innduct.git] / backends / innduct.c
index 8e92f08..18ce508 100644 (file)
@@ -1,18 +1,28 @@
 /*
- * todo
- *  - rename defraise to raise_default
- *  - xmalloc + memset -> xcalloc
- *  - macro for conn iteration
- *  - skipping_long offset calculation is wrong
- *  - reset signals TERM and INT (and HUP) in children
- *
- *  - manpage: document control master stuff
- *  - admin-initiated flush
- *
  * debugging rune:
- *  build-lfs/backends/innduct --no-daemon -C ../inn.conf -f `pwd`/fee sit localhost
+ *  build-lfs/backends/innduct --connection-timeout=30 --no-daemon -C ../inn.conf -f `pwd`/fee sit localhost
  */
 
+/*--
+flow control notes
+to ensure articles go away eventually
+separate queue for each input file
+  queue expiry
+    every period, check head of backlog queue for expiry with SMretrieve
+      if too old: discard, and check next article
+    also check every backlog article as we read it
+  flush expiry
+    after too long in SEPARATED/DROPPING ie Separated/Finishing/Dropping
+    one-off: eat queued articles from flushing and write them to defer
+    one-off: connfail all connections which have any articles from flushing
+    newly read articles from flushing go straight to defer
+    this should take care of it and get us out of this state
+to avoid filling up ram needlessly
+  input control
+    limit number of queued articles for each ipf
+    pause/resume inputfile tailing
+--*/
+
 /*
  * Newsfeeds file entries should look like this:
  *     host.name.of.site[/exclude,exclude,...]\
@@ -212,9 +222,16 @@ perl -ne 'print if m/-8\<-/..m/-\>8-/; print "\f" if m/-\^L-/' backends/innduct.
 #define VA                va_list al;  va_start(al,fmt)
 #define PRINTF(f,a)       __attribute__((__format__(printf,f,a)))
 #define NORET_PRINTF(f,a) __attribute__((__noreturn__,__format__(printf,f,a)))
+#define NORET             __attribute__((__noreturn__))
+
+#define NEW(ptr)              ((ptr)= zxmalloc(sizeof(*(ptr))))
+#define NEW_DECL(type,ptr) type ptr = zxmalloc(sizeof(*(ptr)))
 
 #define DUMPV(fmt,pfx,v) fprintf(f, " " #v "=" fmt, pfx v);
 
+#define FOR_CONN(conn) \
+  for ((conn)=LIST_HEAD(conns); (conn); (conn)=LIST_NEXT((conn)))
+
 /*----- doubly linked lists -----*/
 
 #define ISNODE(T)   struct node list_node
@@ -296,6 +313,7 @@ static void statemc_setstate(StateMachineState newsms, int periods,
 
 static void statemc_start_flush(const char *why); /* Normal => Flushing */
 static void spawn_inndcomm_flush(const char *why); /* Moved => Flushing */
+static int trigger_flush_ok(void); /* => Flushing,FLUSHING, ret 1; or ret 0 */
 
 static void article_done(Conn *conn, Article *art, int whichcount);
 
@@ -312,7 +330,7 @@ static void open_defer(void);
 static void close_defer(void);
 static void search_backlog_file(void);
 static void preterminate(void);
-static void defraise(int signo);
+static void raise_default(int signo) NORET;
 static char *debug_report_ipf(InputFile *ipf);
 
 static void inputfile_reading_start(InputFile *ipf);
@@ -407,13 +425,12 @@ typedef enum {
 #define CONNIOVS 128
 
 typedef enum {
-  xk_Malloc, xk_Const, xk_Artdata
+  xk_Const, xk_Artdata
 } XmitKind;
 
 struct XmitDetails {
   XmitKind kind;
   union {
-    char *malloc_tofree;
     ARTHANDLE *sm_art;
   } info;
 };
@@ -498,7 +515,7 @@ static pid_t self_pid;
 
 /* statemc_init initialises */
 static StateMachineState sms;
-static int sm_period_counter;
+static int until_flush;
 static InputFile *main_input_file, *flushing_input_file, *backlog_input_file;
 static FILE *defer;
 
@@ -659,6 +676,12 @@ static int xwaitpid(pid_t *pid, const char *what) {
   return status;
 }
 
+static void *zxmalloc(size_t sz) {
+  void *p= xmalloc(sz);
+  memset(p,0,sz);
+  return p;
+}
+
 static void xunlink(const char *path, const char *what) {
   int r= unlink(path);
   if (r) sysdie("can't unlink %s %s", path, what);
@@ -670,9 +693,16 @@ static time_t xtime(void) {
   return now;
 }
 
-static void xsigaction(int s, const struct sigaction *sa) {
-  int r= sigaction(s,sa,0);
-  if (r) sysdie("sigaction failed for \"%s\"", strsignal(s));
+static void xsigaction(int signo, const struct sigaction *sa) {
+  int r= sigaction(signo,sa,0);
+  if (r) sysdie("sigaction failed for \"%s\"", strsignal(signo));
+}
+
+static void xsigsetdefault(int signo) {
+  struct sigaction sa;
+  memset(&sa,0,sizeof(sa));
+  sa.sa_handler= SIG_DFL;
+  xsigaction(signo,&sa);
 }
 
 static void xgettimeofday(struct timeval *tv_r) {
@@ -802,34 +832,45 @@ CCMD(help) {
   const ControlCommand *ccmd;
   for (ccmd=control_commands; ccmd->cmd; ccmd++)
     fprintf(cc->out, " %s\n", ccmd->cmd);
+  fputs("NB: permissible arguments are not shown above."
+       "  Not all commands listed are safe.  See innduct(8).\n", cc->out);
 }
 
-CCMD(period) { period(); }
-CCMD(setintarg) { *(int*)c->xdata= atoi(arg); }
-CCMD(setint) { *(int*)c->xdata= c->xval; }
-CCMD(setint_period) { *(int*)c->xdata= c->xval; period(); }
-CCMD(dump);
+CCMD(flush) {
+  int ok= trigger_flush_ok();
+  if (!ok) fprintf(cc->out,"already flushing (state is %s)\n", sms_names[sms]);
+}
 
 CCMD(stop) {
   preterminate();
   notice("terminating (CTRL%d)",cc->fd);
-  defraise(SIGTERM);
+  raise_default(SIGTERM);
   abort();
 }
 
+CCMD(dump);
+
+/* messing with our head: */
+CCMD(period) { period(); }
+CCMD(setintarg) { *(int*)c->xdata= atoi(arg); }
+CCMD(setint) { *(int*)c->xdata= c->xval; }
+CCMD(setint_period) { *(int*)c->xdata= c->xval; period(); }
+
 static const ControlCommand control_commands[]= {
   { "h",             ccmd_help      },
-  { "p",             ccmd_period    },
+  { "flush",         ccmd_flush     },
   { "stop",          ccmd_stop      },
   { "dump q",        ccmd_dump, 0,0 },
   { "dump a",        ccmd_dump, 0,1 },
 
+  { "p",             ccmd_period    },
+
 #define POKES(cmd,func)                                                        \
-  { cmd "sm",        func,           &sm_period_counter,       1 },    \
+  { cmd "flush",     func,           &until_flush,             1 },    \
   { cmd "conn",      func,           &until_connect,           0 },    \
   { cmd "blscan",    func,           &until_backlog_nextscan,  0 },
-POKES("prod ", ccmd_setint_period)
 POKES("next ", ccmd_setint)
+POKES("prod ", ccmd_setint_period)
 
   { "pretend flush", ccmd_setintarg, &simulate_flush             },
   { "wedge blscan",  ccmd_setint,    &until_backlog_nextscan, -1 },
@@ -903,8 +944,7 @@ static void control_stdio_destroy(ControlConn *cc) {
 }
 
 static void control_stdio(void) {
-  ControlConn *cc= xmalloc(sizeof(*cc));
-  memset(cc,0,sizeof(*cc));
+  NEW_DECL(ControlConn *,cc);
   cc->destroy= control_stdio_destroy;
 
   cc->fd= 0;
@@ -925,8 +965,7 @@ static void control_accepted_destroy(ControlConn *cc) {
 
 static void *control_master_readable(oop_source *lp, int master,
                                     oop_event ev, void *u) {
-  ControlConn *cc= xmalloc(sizeof(*cc));
-  memset(cc,0,sizeof(*cc));
+  NEW_DECL(ControlConn *,cc);
   cc->destroy= control_accepted_destroy;
 
   cc->salen= sizeof(cc->sa);
@@ -1113,10 +1152,10 @@ static void connfail(Conn *conn, const char *fmt, ...) {
 
 static void check_idle_conns(void) {
   Conn *conn;
-  for (conn=LIST_HEAD(conns); conn; conn=LIST_NEXT(conn))
+  FOR_CONN(conn)
     conn->since_activity++;
  search_again:
-  for (conn=LIST_HEAD(conns); conn; conn=LIST_NEXT(conn)) {
+  FOR_CONN(conn) {
     if (conn->since_activity <= need_activity_periods) continue;
 
     /* We need to shut this down */
@@ -1202,8 +1241,7 @@ static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
     goto x;
   }
 
-  conn= xmalloc(sizeof(*conn));
-  memset(conn,0,sizeof(*conn));
+  NEW(conn);
   LIST_INIT(conn->waiting);
   LIST_INIT(conn->priority);
   LIST_INIT(conn->sent);
@@ -1411,7 +1449,7 @@ static void check_assign_articles(void) {
      * connections in order.  That way if we have too many
      * connections, the spare ones will go away eventually.
      */
-    for (walk=LIST_HEAD(conns); walk; walk=LIST_NEXT(walk)) {
+    FOR_CONN(walk) {
       if (walk->quitting) continue;
       inqueue= walk->sent.count + walk->priority.count
             + walk->waiting.count;
@@ -1491,7 +1529,6 @@ static void xmit_artbody(Conn *conn, ARTHANDLE *ah /* consumed */) {
 
 static void xmit_free(XmitDetails *d) {
   switch (d->kind) {
-  case xk_Malloc:  free(d->info.malloc_tofree);   break;
   case xk_Artdata: SMfreearticle(d->info.sm_art); break;
   case xk_Const:                                  break;
   default: abort();
@@ -1915,7 +1952,7 @@ static void *feedfile_got_article(oop_source *lp, oop_read *rd,
   if (!data) { feedfile_eof(ipf); return OOP_CONTINUE; }
 
   off_t old_offset= ipf->offset;
-  ipf->offset += recsz + 1;
+  ipf->offset += recsz + !!(ev == OOP_RD_OK);
 
 #define X_BAD_DATA(m) return feedfile_got_bad_data(ipf,old_offset,data,m);
 
@@ -1955,7 +1992,7 @@ static void *feedfile_got_article(oop_source *lp, oop_read *rd,
   ipf->readcount_ok++;
 
   art= xmalloc(sizeof(*art) - 1 + midlen + 1);
-  memset(art,0,sizeof(art));
+  memset(art,0,sizeof(*art));
   art->state= art_Unchecked;
   art->midlen= midlen;
   art->ipf= ipf;  ipf->inprogress++;
@@ -2146,8 +2183,7 @@ static void filemon_method_dump_info(FILE *f) { fprintf(f,"dummy\n"); }
 static void filemon_start(InputFile *ipf) {
   assert(!ipf->filemon);
 
-  ipf->filemon= xmalloc(sizeof(*ipf->filemon));
-  memset(ipf->filemon, 0, sizeof(*ipf->filemon));
+  NEW(ipf->filemon);
   filemon_method_startfile(ipf, ipf->filemon);
 }
 
@@ -2261,7 +2297,7 @@ static void inputfile_reading_stop(InputFile *ipf) {
      |      flsh->rd!=0       |     |              flsh->rd!=0
      |     [Separated]        |     |             [Dropping]
      |      main F idle       |     |              main none
-     |      old D tail        |     |              old D tail
+     |      flsh D tail       |     |              flsh D tail
      |     =============      |     |             ============
      |          |             |     | install       |
      ^          | EOF ON D    |     |  defer        | EOF ON D
@@ -2271,7 +2307,7 @@ static void inputfile_reading_stop(InputFile *ipf) {
      |      flsh->rd==0       |     V              flsh->rd==0
      |     [Finishing]        |     |             [Dropping]
      |      main F tail              |     `.             main none
-     |      old D closed      |       `.           old D closed
+     |      flsh D closed     |       `.           flsh D closed
      |     ===============    V                `.        ===============
      |          |                        `.          |
      |          | ALL D PROCESSED           `.        | ALL D PROCESSED
@@ -2284,7 +2320,7 @@ static void inputfile_reading_stop(InputFile *ipf) {
                                                 DROPPED
                                                [Dropped]
                                                 main none
-                                                old none
+                                                flsh none
                                                 some backlog
                                                ==============
                                                      |
@@ -2400,6 +2436,9 @@ static void statemc_init(void) {
     if (file_d) {
       debug("startup: F!=D => Separated");
       startup_set_input_file(file_d);
+      flushing_input_file= main_input_file;
+      main_input_file= open_input_file(feedfile);
+      if (!main_input_file) die("feedfile vanished during startup");
       SMS(SEPARATED, 0, "found both old and current feed files");
     } else {
       debug("startup: F exists, D ENOENT => Normal");
@@ -2418,7 +2457,7 @@ static void statemc_start_flush(const char *why) { /* Normal => Flushing */
        why,
        (unsigned long)(main_input_file ? main_input_file->offset : 0),
        (unsigned long)target_max_feedfile_size,
-       sm_period_counter);
+       until_flush);
 
   int r= link(feedfile, path_flushing);
   if (r) sysfatal("link feedfile %s to flushing file %s",
@@ -2431,24 +2470,29 @@ static void statemc_start_flush(const char *why) { /* Normal => Flushing */
   spawn_inndcomm_flush(why); /* => Flushing FLUSHING */
 }
 
-static void statemc_period_poll(void) {
-  if (!sm_period_counter) return;
-  sm_period_counter--;
-  assert(sm_period_counter>=0);
-
-  if (sm_period_counter) return;
+static int trigger_flush_ok(void) { /* => Flushing,FLUSHING, ret 1; or ret 0 */
   switch (sms) {
   case sm_NORMAL:
     statemc_start_flush("periodic"); /* Normal => Flushing; => FLUSHING */
-    break;
+    return 1;
   case sm_FLUSHFAILED:
     spawn_inndcomm_flush("retry"); /* Moved => Flushing; => FLUSHING */
-    break;
+    return 1;
   default:
-    abort();
+    return 0;
   }
 }
 
+static void statemc_period_poll(void) {
+  if (!until_flush) return;
+  until_flush--;
+  assert(until_flush>=0);
+
+  if (until_flush) return;
+  int ok= trigger_flush_ok();
+  assert(ok);
+}
+
 static int inputfile_is_done(InputFile *ipf) {
   if (!ipf) return 0;
   if (ipf->inprogress) return 0; /* new article in the meantime */
@@ -2554,7 +2598,7 @@ static void queue_check_input_done(void) {
 static void statemc_setstate(StateMachineState newsms, int periods,
                             const char *forlog, const char *why) {
   sms= newsms;
-  sm_period_counter= periods;
+  until_flush= periods;
 
   const char *xtra= "";
   switch (sms) {
@@ -2772,7 +2816,7 @@ static void search_backlog_file(void) {
 static void preterminate(void) {
   if (in_child) return;
   notice_processed(main_input_file,0,"feedfile","");
-  notice_processed(flushing_input_file,0,"flushing file","");
+  notice_processed(flushing_input_file,0,"flushing","");
   if (backlog_input_file)
     notice_processed(backlog_input_file,0, "backlog file ",
                     backlog_input_file->path);
@@ -2781,12 +2825,10 @@ static void preterminate(void) {
 static int signal_self_pipe[2];
 static sig_atomic_t terminate_sig_flag;
 
-static void defraise(int signo) {
-  struct sigaction sa;
-  memset(&sa,0,sizeof(sa));
-  sa.sa_handler= SIG_DFL;
-  xsigaction(signo,&sa);
+static void raise_default(int signo) {
+  xsigsetdefault(signo);
   raise(signo);
+  abort();
 }
 
 static void *sigarrived_event(oop_source *lp, int fd, oop_event e, void *u) {
@@ -2798,8 +2840,7 @@ static void *sigarrived_event(oop_source *lp, int fd, oop_event e, void *u) {
   if (terminate_sig_flag) {
     preterminate();
     notice("terminating (%s)", strsignal(terminate_sig_flag));
-    defraise(terminate_sig_flag);
-    abort();
+    raise_default(terminate_sig_flag);
   }
   return OOP_CONTINUE;
 }
@@ -2807,7 +2848,8 @@ static void *sigarrived_event(oop_source *lp, int fd, oop_event e, void *u) {
 static void sigarrived_handler(int signum) {
   static char x;
   switch (signum) {
-  case SIGINT: case SIGTERM:
+  case SIGTERM:
+  case SIGINT:
     if (!terminate_sig_flag) terminate_sig_flag= signum;
     break;
   default:
@@ -2979,14 +3021,16 @@ static void postfork_stdio(FILE *f, const char *what, const char *what2) {
 static void postfork(void) {
   in_child= 1;
 
-  if (signal(SIGPIPE, SIG_DFL) == SIG_ERR)
-    sysdie("(in child) failed to reset SIGPIPE");
+  xsigsetdefault(SIGTERM);
+  xsigsetdefault(SIGINT);
+  xsigsetdefault(SIGPIPE);
+  if (terminate_sig_flag) raise(terminate_sig_flag);
 
   postfork_inputfile(main_input_file);
   postfork_inputfile(flushing_input_file);
 
   Conn *conn;
-  for (conn=LIST_HEAD(conns); conn; conn=LIST_NEXT(conn))
+  FOR_CONN(conn)
     conn_closefd(conn,"(in child) ");
 
   postfork_stdio(defer, "defer file ", path_defer);
@@ -3016,7 +3060,7 @@ static void every_schedule(Every *e, struct timeval base) {
 }
 
 static void every(int interval, int fixed_rate, void (*f)(void)) {
-  Every *e= xmalloc(sizeof(*e));
+  NEW_DECL(Every *,e);
   e->interval.tv_sec= interval;
   e->interval.tv_usec= 0;
   e->fixed_rate= fixed_rate;
@@ -3055,7 +3099,7 @@ static void period(void) {
        " input_files main:%s flushing:%s backlog:%s"
        " children connecting=%ld inndcomm=%ld"
        ,
-       sms_names[sms], sm_period_counter,
+       sms_names[sms], until_flush,
          conns.count, queue.count, until_connect,
        dipf_main, dipf_flushing, dipf_backlog,
        (long)connecting_child, (long)inndcomm_child
@@ -3126,7 +3170,7 @@ CCMD(dump) {
 
   fprintf(f,"general");
   DUMPV("%s", sms_names,[sms]);
-  DUMPV("%d", ,sm_period_counter);
+  DUMPV("%d", ,until_flush);
   DUMPV("%ld", (long),self_pid);
   DUMPV("%p", , defer);
   DUMPV("%d", , until_connect);
@@ -3154,7 +3198,7 @@ CCMD(dump) {
   fprintf(f,"conns count=%d\n", conns.count);
 
   Conn *conn;
-  for (conn=LIST_HEAD(conns); conn; conn=LIST_NEXT(conn)) {
+  FOR_CONN(conn) {
 
     fprintf(f,"C%d",conn->fd);
     DUMPV("%p",conn->,rd);             DUMPV("%d",conn->,max_queue);
@@ -3173,16 +3217,8 @@ CCMD(dump) {
       char *dinfo;
       long diff;
       switch (xd->kind) {
-      case xk_Malloc:
-       diff= xd->info.malloc_tofree - (char*)iv->iov_base;
-       dinfo= xasprintf("M%5ld", diff);
-       break;
-      case xk_Const:
-       dinfo= xasprintf("Const");
-       break;
-      case xk_Artdata:
-       dinfo= xasprintf("A%p", xd->info.sm_art);
-       break;
+      case xk_Const:    dinfo= xasprintf("Const");                 break;
+      case xk_Artdata:  dinfo= xasprintf("A%p", xd->info.sm_art);  break;
       default:
        abort();
       }