From: Ian Jackson Date: Tue, 27 Apr 2010 13:04:53 +0000 (+0100) Subject: check fd watches; other bugfixes X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?p=inn-innduct.git;a=commitdiff_plain;h=ffe397bdaa6e5f125df3b5abfd71dc402008be9b;ds=sidebyside check fd watches; other bugfixes --- diff --git a/backends/innduct.c b/backends/innduct.c index 1328d37..43ae915 100644 --- a/backends/innduct.c +++ b/backends/innduct.c @@ -1,8 +1,8 @@ /* * TODO - * - check all structs initialised - * - check all fd watches properly undone * - check all init functions called + * - check ipf->inprogress managed properly + * - xperhaps_close * - 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_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)))) @@ -391,7 +392,7 @@ struct InputFile { oop_readable_call *readable_callback; void *readable_callback_user; - int fd; + int fd; /* may be 0, meaning closed */ Filemon_Perfile *filemon; oop_read *rd; @@ -436,7 +437,8 @@ static const char *sms_names[]= { 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 */ @@ -679,6 +681,10 @@ static int isewouldblock(int errnoval) { 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; @@ -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 (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 (!WIFEXITED(status)) { + } else { report_child_status("connect", status); - /* that's probably the root cause then */ } } 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"); - else + else if (!rs) warn("connect: unexpected EOF on child socket"); + else + fatal("connect: unexpected lack of cmsg from child"); } 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 - 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)); + loop->on_fd(loop, conn->fd, OOP_EXCEPTION, conn_exception, conn); 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! */ + 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; - LIST_ADDHEAD(conns, conn); notice("C%d connected %s", conn->fd, conn->stream ? "streaming" : "plain"); + LIST_ADDHEAD(conns, conn); + 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(); } +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 @@ -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; + assert(ipf->fd); + 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); + free(art); 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, @@ -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->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); - ipf->fd= -1; + ipf->fd= 0; 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); } + assert(fd>0); 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; } -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 */ - if (ipf->fd >= 0) + if (ipf->fd) { 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); - art->offset= ipf->offset; - art->blanklen= recsz; - art->midlen= midlen; art->state= art_Unchecked; + art->midlen= midlen; 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); @@ -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) { - assert(ipf->fd>=0); + assert(ipf->fd); assert(sms==sm_SEPARATED || sms==sm_DROPPING); } else if (ipf==backlog_input_file) { - assert(ipf->fd>=0); + assert(ipf->fd); } else { abort(); } @@ -1708,7 +1740,7 @@ static const oop_rd_style feedfile_rdstyle= { }; 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; @@ -1727,7 +1759,7 @@ static void inputfile_tailing_start(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; @@ -1794,7 +1826,7 @@ static void inputfile_tailing_stop(InputFile *ipf) { | 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 @@ -1804,7 +1836,7 @@ static void inputfile_tailing_stop(InputFile *ipf) { | 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 @@ -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 */ - if (ipf->fd >= 0); return 0; /* not had EOF */ + if (ipf->fd) return 0; /* not had EOF */ 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: - xtra= flushing_input_file->fd >= 0 ? "-1" : "-2"; + xtra= flushing_input_file->fd ? "-1" : "-2"; break; default:; } @@ -2388,9 +2420,8 @@ void spawn_inndcomm_flush(const char *why) { /* Moved => Flushing */ 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) { @@ -2690,6 +2721,9 @@ int main(int argc, char **argv) { notice("starting"); + LIST_INIT(conns); + LIST_INIT(queue); + if (!filemon_method_init()) { warn("no file monitoring available, polling"); filepoll_schedule();