chiark / gitweb /
check fd watches; other bugfixes
authorIan Jackson <ian@liberator.(none)>
Tue, 27 Apr 2010 13:04:53 +0000 (14:04 +0100)
committerIan Jackson <ian@liberator.(none)>
Tue, 27 Apr 2010 13:04:53 +0000 (14:04 +0100)
backends/innduct.c

index 1328d37..43ae915 100644 (file)
@@ -1,8 +1,8 @@
 /*
  * TODO
 /*
  * TODO
- *  - check all structs initialised
- *  - check all fd watches properly undone
  *  - check all init functions called
  *  - check all init functions called
+ *  - check ipf->inprogress managed properly
+ *  - xperhaps_close
  *  - actually implement badusage
  *  - options for all options
  *  - manpage
  *  - actually implement badusage
  *  - options for all options
  *  - manpage
@@ -238,6 +238,7 @@ perl -ne 'print if m/-8\<-/..m/-\>8-/; print "\f" if m/-\^L-/' backends/innduct.
 #define LIST_REMHEAD(l) LIST_REMSOMEHOW((l),list_remhead)
 #define LIST_REMTAIL(l) LIST_REMSOMEHOW((l),list_remtail)
 
 #define LIST_REMHEAD(l) LIST_REMSOMEHOW((l),list_remhead)
 #define LIST_REMTAIL(l) LIST_REMSOMEHOW((l),list_remtail)
 
+#define LIST_INIT(l) ((l).hd, list_new((struct list*)&(l)))
 #define LIST_HEAD(l) ((typeof((l).hd))(list_head((struct list*)&(l))))
 #define LIST_NEXT(n) ((typeof(n))list_succ(NODE((n))))
 #define LIST_BACK(n) ((typeof(n))list_pred(NODE((n))))
 #define LIST_HEAD(l) ((typeof((l).hd))(list_head((struct list*)&(l))))
 #define LIST_NEXT(n) ((typeof(n))list_succ(NODE((n))))
 #define LIST_BACK(n) ((typeof(n))list_pred(NODE((n))))
@@ -391,7 +392,7 @@ struct InputFile {
   oop_readable_call *readable_callback;
   void *readable_callback_user;
 
   oop_readable_call *readable_callback;
   void *readable_callback_user;
 
-  int fd;
+  int fd; /* may be 0, meaning closed */
   Filemon_Perfile *filemon;
 
   oop_read *rd;
   Filemon_Perfile *filemon;
 
   oop_read *rd;
@@ -436,7 +437,8 @@ static const char *sms_names[]= {
 
 struct Conn {
   ISNODE(Conn);
 
 struct Conn {
   ISNODE(Conn);
-  int fd, max_queue, stream, quitting;
+  int fd; /* may be 0, meaning closed (during construction/destruction) */
+  int max_queue, stream, quitting;
   int since_activity; /* periods */
   ArticleList waiting; /* not yet told peer */
   ArticleList priority; /* peer says send it now */
   int since_activity; /* periods */
   ArticleList waiting; /* not yet told peer */
   ArticleList priority; /* peer says send it now */
@@ -679,6 +681,10 @@ static int isewouldblock(int errnoval) {
 
 static void conn_dispose(Conn *conn) {
   if (!conn) return;
 
 static void conn_dispose(Conn *conn) {
   if (!conn) return;
+  if (conn->fd) {
+    loop->cancel_fd(loop, conn->fd, OOP_WRITE);
+    loop->cancel_fd(loop, conn->fd, OOP_EXCEPTION);
+  }
   perhaps_close(&conn->fd);
   free(conn);
   until_connect= reconnect_delay_periods;
   perhaps_close(&conn->fd);
   free(conn);
   until_connect= reconnect_delay_periods;
@@ -728,16 +734,18 @@ static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
     if (got != -1) {
       assert(got==connecting_child);
       connecting_child= 0;
     if (got != -1) {
       assert(got==connecting_child);
       connecting_child= 0;
-      if (WIFEXITED(status) &&
-         (WEXITSTATUS(status) != 0 &&
-          WEXITSTATUS(status) != CONNCHILD_ESTATUS_STREAM &&
-          WEXITSTATUS(status) != CONNCHILD_ESTATUS_NOSTREAM)) {
-       /* child already reported the problem */
+      if (WIFEXITED(status)) {
+       if (WEXITSTATUS(status) != 0 &&
+           WEXITSTATUS(status) != CONNCHILD_ESTATUS_STREAM &&
+           WEXITSTATUS(status) != CONNCHILD_ESTATUS_NOSTREAM)
+         /* child already reported the problem */;
+       else
+         warn("connect: connection child exited code %d but no cmsg",
+              WEXITSTATUS(status));
       } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALRM) {
        warn("connect: connection attempt timed out");
       } else if (WIFSIGNALED(status) && WTERMSIG(status) == SIGALRM) {
        warn("connect: connection attempt timed out");
-      } else if (!WIFEXITED(status)) {
+      } else {
        report_child_status("connect", status);
        report_child_status("connect", status);
-       /* that's probably the root cause then */
       }
     } else {
       /* child is still running apparently, report the socket problem */
       }
     } else {
       /* child is still running apparently, report the socket problem */
@@ -745,8 +753,10 @@ static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
        syswarn("connect: read from child socket failed");
       else if (e == OOP_EXCEPTION)
        warn("connect: unexpected exception on child socket");
        syswarn("connect: read from child socket failed");
       else if (e == OOP_EXCEPTION)
        warn("connect: unexpected exception on child socket");
-      else
+      else if (!rs)
        warn("connect: unexpected EOF on child socket");
        warn("connect: unexpected EOF on child socket");
+      else
+       fatal("connect: unexpected lack of cmsg from child");
     }
     goto x;
   }
     }
     goto x;
   }
@@ -761,9 +771,10 @@ static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
   CHK(len,   CMSG_LEN(sizeof(conn->fd)));
 #undef CHK
 
   CHK(len,   CMSG_LEN(sizeof(conn->fd)));
 #undef CHK
 
-  if (CMSG_NXTHDR(&msg,h)) { die("connect: child sent many cmsgs"); goto x; }
+  if (CMSG_NXTHDR(&msg,h)) die("connect: child sent many cmsgs");
 
   memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
 
   memcpy(&conn->fd, CMSG_DATA(h), sizeof(conn->fd));
+  loop->on_fd(loop, conn->fd, OOP_EXCEPTION, conn_exception, conn);
 
   int status;
   pid_t got= waitpid(connecting_child, &status, 0);
 
   int status;
   pid_t got= waitpid(connecting_child, &status, 0);
@@ -781,10 +792,14 @@ static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
   }
 
   /* Phew! */
   }
 
   /* Phew! */
+  LIST_INIT(conn->waiting);
+  LIST_INIT(conn->priority);
+  LIST_INIT(conn->sent);
   setnonblock(conn->fd, 1);
   conn->max_queue= conn->stream ? max_queue_per_conn : 1;
   setnonblock(conn->fd, 1);
   conn->max_queue= conn->stream ? max_queue_per_conn : 1;
-  LIST_ADDHEAD(conns, conn);
   notice("C%d connected %s", conn->fd, conn->stream ? "streaming" : "plain");
   notice("C%d connected %s", conn->fd, conn->stream ? "streaming" : "plain");
+  LIST_ADDHEAD(conns, conn);
+
   connect_attempt_discard();
   check_master_queue();
   return 0;
   connect_attempt_discard();
   check_master_queue();
   return 0;
@@ -794,6 +809,18 @@ static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) {
   connect_attempt_discard();
 }
 
   connect_attempt_discard();
 }
 
+static void conn_exception(oop_source *lp, int fd,
+                          oop_event ev, void *conn_v) {
+  Conn *conn= conn_v;
+  char ch;
+  assert(fd == conn->fd);
+  assert(ev == OOP_EXCEPTION);
+  r= read(conn->fd, &ch, 1);
+  if (r<0) connfail(conn,"read failed: %s",strerror(errno));
+  connfail(conn,"exceptional condition on socket (peer sent urgent"
+          " data? read(,,1)=%d)",r);
+}  
+
 static int allow_connect_start(void) {
   return conns.count < max_connections
     && !connecting_child
 static int allow_connect_start(void) {
   return conns.count < max_connections
     && !connecting_child
@@ -1261,6 +1288,8 @@ static void article_done(Conn *conn, Article *art, int whichcount) {
   else if (whichcount == RC_unwanted) update_nocheck(0);
 
   InputFile *ipf= art->ipf;
   else if (whichcount == RC_unwanted) update_nocheck(0);
 
   InputFile *ipf= art->ipf;
+  assert(ipf->fd);
+
   while (art->blanklen) {
     static const char spaces[]=
       "                                                                "
   while (art->blanklen) {
     static const char spaces[]=
       "                                                                "
@@ -1281,11 +1310,10 @@ static void article_done(Conn *conn, Article *art, int whichcount) {
 
   ipf->inprogress--;
   assert(ipf->inprogress >= 0);
 
   ipf->inprogress--;
   assert(ipf->inprogress >= 0);
+  free(art);
 
   if (!ipf->inprogress && ipf != main_input_file)
     queue_check_input_done();
 
   if (!ipf->inprogress && ipf != main_input_file)
     queue_check_input_done();
-
-  free(art);
 }
 
 static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_event ev,
 }
 
 static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_event ev,
@@ -1389,9 +1417,10 @@ static void feedfile_eof(InputFile *ipf) {
   assert(ipf != main_input_file); /* promised by tailing_try_read */
 
   inputfile_tailing_stop(ipf);
   assert(ipf != main_input_file); /* promised by tailing_try_read */
 
   inputfile_tailing_stop(ipf);
-  assert(ipf->fd >= 0);
+  assert(ipf->fd > 0);
+fixme do not close fd do something else because art_done needs it to blank entries;
   if (close(ipf->fd)) sysdie("could not close input file %s", ipf->path);
   if (close(ipf->fd)) sysdie("could not close input file %s", ipf->path);
-  ipf->fd= -1;
+  ipf->fd= 0;
 
   if (ipf == flushing_input_file) {
     assert(sms==sm_SEPARATED || sms==sm_DROPPING);
 
   if (ipf == flushing_input_file) {
     assert(sms==sm_SEPARATED || sms==sm_DROPPING);
@@ -1410,6 +1439,7 @@ static InputFile *open_input_file(const char *path) {
     if (errno==ENOENT) return 0;
     sysfatal("unable to open input file %s", path);
   }
     if (errno==ENOENT) return 0;
     sysfatal("unable to open input file %s", path);
   }
+  assert(fd>0);
 
   InputFile *ipf= xmalloc(sizeof(*ipf) + strlen(path) + 1);
   memset(ipf,0,sizeof(*ipf));
 
   InputFile *ipf= xmalloc(sizeof(*ipf) + strlen(path) + 1);
   memset(ipf,0,sizeof(*ipf));
@@ -1420,14 +1450,16 @@ static InputFile *open_input_file(const char *path) {
   return ipf;
 }
 
   return ipf;
 }
 
-static void close_input_file(InputFile *ipf) {
+static void close_input_file(InputFile *ipf) { /* does not free */
   assert(!ipf->readable_callback); /* must have had ->on_cancel */
   assert(!ipf->filemon); /* must have had inputfile_tailing_stop */
   assert(!ipf->rd); /* must have had inputfile_tailing_stop */
   assert(!ipf->inprogress); /* no dangling pointers pointing here */
 
   assert(!ipf->readable_callback); /* must have had ->on_cancel */
   assert(!ipf->filemon); /* must have had inputfile_tailing_stop */
   assert(!ipf->rd); /* must have had inputfile_tailing_stop */
   assert(!ipf->inprogress); /* no dangling pointers pointing here */
 
-  if (ipf->fd >= 0)
+  if (ipf->fd) {
     if (close(ipf->fd)) sysdie("could not close input file %s", ipf->path);
     if (close(ipf->fd)) sysdie("could not close input file %s", ipf->path);
+    ipf->fd= 0;
+  }
 }
 
 
 }
 
 
@@ -1507,12 +1539,12 @@ static void *feedfile_got_article(oop_source *lp, oop_read *rd,
   ipf->readcount_ok++;
 
   art= xmalloc(sizeof(*art) - 1 + midlen + 1);
   ipf->readcount_ok++;
 
   art= xmalloc(sizeof(*art) - 1 + midlen + 1);
-  art->offset= ipf->offset;
-  art->blanklen= recsz;
-  art->midlen= midlen;
   art->state= art_Unchecked;
   art->state= art_Unchecked;
+  art->midlen= midlen;
   art->ipf= ipf;  ipf->inprogress++;
   art->token= TextToToken(tokentextbuf);
   art->ipf= ipf;  ipf->inprogress++;
   art->token= TextToToken(tokentextbuf);
+  art->offset= ipf->offset;
+  art->blanklen= recsz;
   strcpy(art->messageid, space+1);
   LIST_ADDTAIL(queue, art);
 
   strcpy(art->messageid, space+1);
   LIST_ADDTAIL(queue, art);
 
@@ -1574,10 +1606,10 @@ static ssize_t tailing_try_read(struct oop_readable *rable, void *buffer,
        errno=EAGAIN;
        return -1;
       } else if (ipf==flushing_input_file) {
        errno=EAGAIN;
        return -1;
       } else if (ipf==flushing_input_file) {
-       assert(ipf->fd>=0);
+       assert(ipf->fd);
        assert(sms==sm_SEPARATED || sms==sm_DROPPING);
       } else if (ipf==backlog_input_file) {
        assert(sms==sm_SEPARATED || sms==sm_DROPPING);
       } else if (ipf==backlog_input_file) {
-       assert(ipf->fd>=0);
+       assert(ipf->fd);
       } else {
        abort();
       }
       } else {
        abort();
       }
@@ -1708,7 +1740,7 @@ static const oop_rd_style feedfile_rdstyle= {
 };
 
 static void inputfile_tailing_start(InputFile *ipf) {
 };
 
 static void inputfile_tailing_start(InputFile *ipf) {
-  assert(!ipf->fd);
+  assert(!ipf->rd);
   ipf->readable.on_readable= tailing_on_readable;
   ipf->readable.on_cancel=   tailing_on_cancel;
   ipf->readable.try_read=    tailing_try_read;
   ipf->readable.on_readable= tailing_on_readable;
   ipf->readable.on_cancel=   tailing_on_cancel;
   ipf->readable.try_read=    tailing_try_read;
@@ -1727,7 +1759,7 @@ static void inputfile_tailing_start(InputFile *ipf) {
 }
 
 static void inputfile_tailing_stop(InputFile *ipf) {
 }
 
 static void inputfile_tailing_stop(InputFile *ipf) {
-  assert(ipf->fd);
+  assert(ipf->rd);
   oop_rd_cancel(ipf->rd);
   oop_rd_delete(ipf->rd);
   ipf->rd= 0;
   oop_rd_cancel(ipf->rd);
   oop_rd_delete(ipf->rd);
   ipf->rd= 0;
@@ -1794,7 +1826,7 @@ static void inputfile_tailing_stop(InputFile *ipf) {
      |          V             |     |              V
      |     =============      V     V                    ============
      |      SEPARATED-1       |     |              DROPPING-1
      |          V             |     |              V
      |     =============      V     V                    ============
      |      SEPARATED-1       |     |              DROPPING-1
-     |      flsh->fd>=0       |     |              flsh->fd>=0
+     |      flsh->fd>0        |     |              flsh->fd>0
      |     [Separated]        |     |             [Dropping]
      |      main F idle       |     |              main none
      |      old D tail        |     |              old D tail
      |     [Separated]        |     |             [Dropping]
      |      main F idle       |     |              main none
      |      old D tail        |     |              old D tail
@@ -1804,7 +1836,7 @@ static void inputfile_tailing_stop(InputFile *ipf) {
      |          V             |     |               V
      |     ===============    |     |             ===============
      |      SEPARATED-2       |     |              DROPPING-2
      |          V             |     |               V
      |     ===============    |     |             ===============
      |      SEPARATED-2       |     |              DROPPING-2
-     |      flsh->fd==-1      |     V              flsh->fd==-1
+     |      flsh->fd==0       |     V              flsh->fd==0
      |     [Finishing]        |     |             [Dropping]
      |      main F tail              |     `.             main none
      |      old D closed      |       `.           old D closed
      |     [Finishing]        |     |             [Dropping]
      |      main F tail              |     `.             main none
      |      old D closed      |       `.           old D closed
@@ -1976,7 +2008,7 @@ static void statemc_period_poll(void) {
 static int inputfile_is_done(InputFile *ipf) {
   if (!ipf) return 0;
   if (ipf->inprogress) return 0; /* new article in the meantime */
 static int inputfile_is_done(InputFile *ipf) {
   if (!ipf) return 0;
   if (ipf->inprogress) return 0; /* new article in the meantime */
-  if (ipf->fd >= 0); return 0; /* not had EOF */
+  if (ipf->fd) return 0; /* not had EOF */
   return 1;
 }
 
   return 1;
 }
 
@@ -2078,7 +2110,7 @@ static void statemc_setstate(StateMachineState newsms, int periods,
     if (!main_input_file) xtra= "-ABSENT";
     break;
   case sm_SEPARATED: case sm_DROPPING:
     if (!main_input_file) xtra= "-ABSENT";
     break;
   case sm_SEPARATED: case sm_DROPPING:
-    xtra= flushing_input_file->fd >= 0 ? "-1" : "-2";
+    xtra= flushing_input_file->fd ? "-1" : "-2";
     break;
   default:;
   }
     break;
   default:;
   }
@@ -2388,9 +2420,8 @@ void spawn_inndcomm_flush(const char *why) { /* Moved => Flushing */
 
 static void postfork_inputfile(InputFile *ipf) {
   if (!ipf) return;
 
 static void postfork_inputfile(InputFile *ipf) {
   if (!ipf) return;
-  assert(ipf->fd >= 0);
-  close(ipf->fd);
-  ipf->fd= -1;
+  assert(ipf->fd);
+  perhaps_close(&ipf->fd);
 }
 
 static void postfork_stdio(FILE *f) {
 }
 
 static void postfork_stdio(FILE *f) {
@@ -2690,6 +2721,9 @@ int main(int argc, char **argv) {
 
   notice("starting");
 
 
   notice("starting");
 
+  LIST_INIT(conns);
+  LIST_INIT(queue);
+
   if (!filemon_method_init()) {
     warn("no file monitoring available, polling");
     filepoll_schedule();
   if (!filemon_method_init()) {
     warn("no file monitoring available, polling");
     filepoll_schedule();