X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?p=innduct.git;a=blobdiff_plain;f=backends%2Finnduct.c;h=047d99e6f9f3563903ac17e2310fa1470f93066e;hp=c518fd095e9ce0873e7f51bc06a8d67ab38ada6c;hb=a319a3c12dc51eebaf5305db1b6c150af68d5e35;hpb=5c611f214314f05eccb084d53fdefcee9459bc9d diff --git a/backends/innduct.c b/backends/innduct.c index c518fd0..047d99e 100644 --- a/backends/innduct.c +++ b/backends/innduct.c @@ -1,13 +1,28 @@ /* - * todo - * - inotify not working ? - * - some per-conn info thing for control - * - manpage: document control master stuff - * * debugging rune: - * build-lfs/backends/innduct --no-daemon -f `pwd`/fee sit dom + * 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,...]\ @@ -207,6 +222,15 @@ 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 -----*/ @@ -269,6 +293,7 @@ typedef struct InputFile InputFile; typedef struct XmitDetails XmitDetails; typedef struct Filemon_Perfile Filemon_Perfile; typedef enum StateMachineState StateMachineState; +typedef struct ControlCommand ControlCommand; DEFLIST(Conn); DEFLIST(Article); @@ -288,8 +313,9 @@ 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); +static void article_done(Article *art, int whichcount); static void check_assign_articles(void); static void queue_check_input_done(void); @@ -303,6 +329,9 @@ static void period(void); static void open_defer(void); static void close_defer(void); static void search_backlog_file(void); +static void preterminate(void); +static void raise_default(int signo) NORET; +static char *debug_report_ipf(InputFile *ipf); static void inputfile_reading_start(InputFile *ipf); static void inputfile_reading_stop(InputFile *ipf); @@ -346,6 +375,7 @@ static int flushfail_retry_periods=1000; static int backlog_retry_minperiods=50; static int backlog_spontrescan_periods=300; static int spontaneous_flush_periods=100000; +static int max_separated_periods=2000; static int need_activity_periods=1000; static double max_bad_data_ratio= 1; /* conv'd from percentage by main */ @@ -363,6 +393,9 @@ typedef enum { /* in queue in conn->sent */ art_MaxState, } ArtState; +static const char *const artstate_names[]= + { "Unchecked", "Wanted", "Unsolicited", 0 }; + #define RESULT_COUNTS(RCS,RCN) \ RCS(sent) \ RCS(accepted) \ @@ -393,13 +426,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; }; @@ -417,10 +449,13 @@ struct InputFile { Filemon_Perfile *filemon; oop_read *rd; /* non-0: reading; 0: constructing, or had EOF */ - long inprogress; /* no. of articles read but not processed */ off_t offset; int skippinglong; + ArticleList queue; + long inprogress; /* includes queue.count and also articles in conns */ + long autodefer; /* -1 means not doing autodefer */ + int counts[art_MaxState][RCI_max]; int readcount_ok, readcount_blank, readcount_err; char path[]; @@ -476,21 +511,21 @@ struct Conn { /* main initialises */ static oop_source *loop; static ConnList conns; -static ArticleList queue; -static char *path_lock, *path_flushing, *path_defer, *path_control; +static char *path_lock, *path_flushing, *path_defer; +static char *path_control, *path_dump; static char *globpat_backlog; static pid_t self_pid; /* statemc_init initialises */ static StateMachineState sms; -static FILE *defer; +static int until_flush; static InputFile *main_input_file, *flushing_input_file, *backlog_input_file; -static int sm_period_counter; +static FILE *defer; /* initialisation to 0 is good */ static int until_connect, until_backlog_nextscan; static double accept_proportion; -static int nocheck, nocheck_reported; +static int nocheck, nocheck_reported, in_child; /* for simulation, debugging, etc. */ int simulate_flush= -1; @@ -530,6 +565,7 @@ static void logv(int sysloglevel, const char *pfx, int errnoval, #define diewrap(fn, pfx, sysloglevel, err, estatus) \ static void fn(const char *fmt, ...) NORET_PRINTF(1,2); \ static void fn(const char *fmt, ...) { \ + preterminate(); \ VA; \ logv(sysloglevel, pfx, err, fmt, al); \ exit(estatus); \ @@ -643,6 +679,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); @@ -654,6 +696,18 @@ static time_t xtime(void) { return now; } +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) { int r= gettimeofday(tv_r,0); if (r) sysdie("gettimeofday(2) failed"); @@ -701,15 +755,16 @@ static int samefile(const struct stat *a, const struct stat *b) { a->st_dev == b->st_dev); } -static char *sanitise(const char *input) { +static char *sanitise(const char *input, int len) { static char sanibuf[100]; /* returns pointer to this buffer! */ const char *p= input; + const char *endp= len>=0 ? input+len : 0; char *q= sanibuf; *q++= '`'; for (;;) { if (q > sanibuf+sizeof(sanibuf)-8) { strcpy(q,"'.."); break; } - int c= *p++; + int c= (!endp || p=' ' && c<=126 && c!='\\') { *q++= c; continue; } sprintf(q,"\\x%02x",c); @@ -722,7 +777,6 @@ static int isewouldblock(int errnoval) { return errnoval==EWOULDBLOCK || errnoval==EAGAIN; } - /*========== command and control connections ==========*/ static int control_master; @@ -762,7 +816,6 @@ static void control_prompt(ControlConn *cc /* may destroy*/) { control_checkouterr(cc); } -typedef struct ControlCommand ControlCommand; struct ControlCommand { const char *cmd; void (*f)(ControlConn *cc, const ControlCommand *ccmd, @@ -782,25 +835,47 @@ 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(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); + 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 }, - { "pretend flush", ccmd_setintarg, &simulate_flush }, + { "h", ccmd_help }, + { "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 }, { 0 } }; @@ -872,8 +947,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; @@ -894,8 +968,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); @@ -1050,12 +1123,17 @@ static void vconnfail(Conn *conn, const char *fmt, va_list al) { memset(requeue,0,sizeof(requeue)); Article *art; - while ((art= LIST_REMHEAD(conn->priority))) LIST_ADDTAIL(queue, art); - while ((art= LIST_REMHEAD(conn->waiting))) LIST_ADDTAIL(queue, art); + + while ((art= LIST_REMHEAD(conn->priority))) + LIST_ADDTAIL(art->ipf->queue, art); + + while ((art= LIST_REMHEAD(conn->waiting))) + LIST_ADDTAIL(art->ipf->queue, art); + while ((art= LIST_REMHEAD(conn->sent))) { requeue[art->state]++; if (art->state==art_Unsolicited) art->state= art_Unchecked; - LIST_ADDTAIL(queue,art); + LIST_ADDTAIL(art->ipf->queue,art); } int i; @@ -1082,10 +1160,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 */ @@ -1171,8 +1249,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); @@ -1299,7 +1376,7 @@ static void connect_start(void) { } else { buf[l]= 0; fatal("connect: %s: %s", stripped ? "rejected" : "failed", - sanitise(buf)); + sanitise(buf,-1)); } } if (NNTPsendpassword((char*)remote_host, cn_from, cn_to) < 0) @@ -1319,13 +1396,13 @@ static void connect_start(void) { assert(l>=1); if (buf[l-1]!='\n') fatal("connect: response to MODE STREAM is too long: %.100s...", - sanitise(buf)); + sanitise(buf,-1)); l--; if (l>0 && buf[l-1]=='\r') l--; buf[l]= 0; char *ep; int rcode= strtoul(buf,&ep,10); if (ep != &buf[3]) - fatal("connect: bad response to MODE STREAM: %.50s", sanitise(buf)); + fatal("connect: bad response to MODE STREAM: %.50s", sanitise(buf,-1)); switch (rcode) { case 203: @@ -1336,7 +1413,7 @@ static void connect_start(void) { break; default: warn("connect: unexpected response to MODE STREAM: %.50s", - sanitise(buf)); + sanitise(buf,-1)); exitstatus= 2; break; } @@ -1366,9 +1443,23 @@ static void connect_start(void) { /*---------- assigning articles to conns, and transmitting ----------*/ +static Article *dequeue_from(int peek, InputFile *ipf) { + if (!ipf) return 0; + if (peek) return LIST_HEAD(ipf->queue); + else return LIST_REMHEAD(ipf->queue); +} + +static Article *dequeue(int peek) { + Article *art; + art= dequeue_from(peek, flushing_input_file); if (art) return art; + art= dequeue_from(peek, backlog_input_file); if (art) return art; + art= dequeue_from(peek, main_input_file); if (art) return art; + return 0; +} + static void check_assign_articles(void) { for (;;) { - if (!queue.count) + if (!dequeue(1)) break; Conn *walk, *use=0; @@ -1380,7 +1471,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; @@ -1393,7 +1484,7 @@ static void check_assign_articles(void) { if (use) { if (!inqueue) use->since_activity= 0; /* reset idle counter */ while (spare>0) { - Article *art= LIST_REMHEAD(queue); + Article *art= dequeue(0); if (!art) break; LIST_ADDTAIL(use->waiting, art); spare--; @@ -1436,6 +1527,76 @@ static void conn_maybe_write(Conn *conn) { } } +/*---------- expiry and deferral ----------*/ + +static void article_defer(Article *art /* not on a queue */, int whichcount) { + open_defer(); + if (fprintf(defer, "%s %s\n", TokenToText(art->token), art->messageid) <0 + || fflush(defer)) + sysfatal("write to defer file %s",path_defer); + article_done(art, whichcount); +} + +static int article_check_expired(Article *art /* must be queued, not conn */) { + ARTHANDLE *artdata= SMretrieve(art->token, RETR_STAT); + if (artdata) { SMfreearticle(artdata); return 0; } + + LIST_REMOVE(art->ipf->queue, art); + art->missing= 1; + art->ipf->counts[art_Unchecked][RC_missing]++; + article_done(art,-1); + return 1; +} + +static void inputfile_queue_check_expired(InputFile *ipf) { + if (!ipf) return; + + for (;;) { + Article *art= LIST_HEAD(ipf->queue); + int exp= article_check_expired(art); + if (!exp) break; + } +} + +static void article_autodefer(InputFile *ipf, Article *art) { + ipf->autodefer++; + article_defer(art,-1); +} + +static int has_article_in(const ArticleList *al, InputFile *ipf) { + Article *art; + for (art=LIST_HEAD(*al); art; art=LIST_NEXT(art)) + if (art->ipf == ipf) return 1; + return 0; +} + +static void autodefer_input_file(InputFile *ipf) { + ipf->autodefer= 0; + + Article *art; + while ((art= LIST_REMHEAD(ipf->queue))) + article_autodefer(ipf, art); + + if (ipf->inprogress) { + Conn *walk; + FOR_CONN(walk) { + if (has_article_in(&walk->waiting, ipf) || + has_article_in(&walk->priority, ipf) || + has_article_in(&walk->sent, ipf)) + walk->quitting= -1; + } + while (ipf->inprogress) { + FOR_CONN(walk) + if (walk->quitting < 0) goto found; + abort(); /* where are they ?? */ + + found: + connfail(walk, "connection is stuck or crawling," + " and we need to finish flush"); + } + } +} + /*========== article transmission ==========*/ static XmitDetails *xmit_core(Conn *conn, const char *data, int len, @@ -1460,7 +1621,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(); @@ -1534,7 +1694,7 @@ static void conn_make_some_xmits(Conn *conn) { XMIT_LITERAL("\r\n"); xmit_artbody(conn, artdata); } else { - article_done(conn, art, -1); + article_done(art, -1); continue; } } else { @@ -1565,7 +1725,6 @@ static void conn_make_some_xmits(Conn *conn) { } } - /*========== handling responses from peer ==========*/ static const oop_rd_style peer_rd_style= { @@ -1655,7 +1814,7 @@ static void update_nocheck(int accepted) { nocheck= new_nocheck; } -static void article_done(Conn *conn, Article *art, int whichcount) { +static void article_done(Article *art, int whichcount) { if (!art->missing) art->ipf->counts[art->state][whichcount]++; if (whichcount == RC_accepted) update_nocheck(1); @@ -1706,7 +1865,7 @@ static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_rd_event ev, } assert(ev == OOP_RD_OK); - char *sani= sanitise(data); + char *sani= sanitise(data,-1); char *ep; unsigned long code= strtoul(data, &ep, 10); @@ -1744,7 +1903,7 @@ static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_rd_event ev, #define ARTICLE_DEALTWITH(streaming,musthavesent,how) do{ \ code_streaming= (streaming); \ GET_ARTICLE(musthavesent); \ - article_done(conn, art, RC_##how); \ + article_done(art, RC_##how); \ goto dealtwith; \ }while(0) @@ -1789,11 +1948,7 @@ static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_rd_event ev, code_streaming= 1; case 436: /* IHAVE says try later */ GET_ARTICLE(0); - open_defer(); - if (fprintf(defer, "%s %s\n", TokenToText(art->token), art->messageid) <0 - || fflush(defer)) - sysfatal("write to defer file %s",path_defer); - article_done(conn, art, RC_deferred); + article_defer(art, RC_deferred); break; } @@ -1834,6 +1989,8 @@ static InputFile *open_input_file(const char *path) { memset(ipf,0,sizeof(*ipf)); ipf->fd= fd; + ipf->autodefer= -1; + LIST_INIT(ipf->queue); strcpy(ipf->path, path); return ipf; @@ -1853,7 +2010,7 @@ static void close_input_file(InputFile *ipf) { /* does not free */ static void *feedfile_got_bad_data(InputFile *ipf, off_t offset, const char *data, const char *how) { warn("corrupted file: %s, offset %lu: %s: in %s", - ipf->path, (unsigned long)offset, how, sanitise(data)); + ipf->path, (unsigned long)offset, how, sanitise(data,-1)); ipf->readcount_err++; if (ipf->readcount_err > max_bad_data_initial + (ipf->readcount_ok+ipf->readcount_blank) / max_bad_data_ratio) @@ -1884,7 +2041,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); @@ -1924,6 +2081,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)); art->state= art_Unchecked; art->midlen= midlen; art->ipf= ipf; ipf->inprogress++; @@ -1931,7 +2089,12 @@ static void *feedfile_got_article(oop_source *lp, oop_read *rd, art->offset= old_offset; art->blanklen= recsz; strcpy(art->messageid, space+1); - LIST_ADDTAIL(queue, art); + LIST_ADDTAIL(ipf->queue, art); + + if (ipf->autodefer >= 0) + article_autodefer(ipf, art); + else if (ipf==backlog_input_file) + article_check_expired(art); if (sms==sm_NORMAL && ipf==main_input_file && ipf->offset >= target_max_feedfile_size) @@ -2082,6 +2245,15 @@ static int filemon_method_init(void) { return 1; } +static void filemon_method_dump_info(FILE *f) { + int i; + fprintf(f,"inotify"); + DUMPV("%d",,filemon_inotify_fd); + DUMPV("%d",,filemon_inotify_wdmax); + for (i=0; ifilemon); - ipf->filemon= xmalloc(sizeof(*ipf->filemon)); - memset(ipf->filemon, 0, sizeof(*ipf->filemon)); + NEW(ipf->filemon); filemon_method_startfile(ipf, ipf->filemon); } @@ -2219,7 +2391,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 @@ -2229,7 +2401,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 @@ -2242,7 +2414,7 @@ static void inputfile_reading_stop(InputFile *ipf) { DROPPED [Dropped] main none - old none + flsh none some backlog ============== | @@ -2358,7 +2530,11 @@ static void statemc_init(void) { if (file_d) { debug("startup: F!=D => Separated"); startup_set_input_file(file_d); - SMS(SEPARATED, 0, "found both old and current feed files"); + flushing_input_file= main_input_file; + main_input_file= open_input_file(feedfile); + if (!main_input_file) die("feedfile vanished during startup"); + SMS(SEPARATED, max_separated_periods, + "found both old and current feed files"); } else { debug("startup: F exists, D ENOENT => Normal"); InputFile *file_f= open_input_file(feedfile); @@ -2376,7 +2552,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", @@ -2389,24 +2565,39 @@ 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; + + case sm_SEPARATED: + case sm_DROPPING: + warn("took too long to complete old feedfile after flush, autodeferring"); + assert(flushing_input_file); + autodefer_input_file(flushing_input_file); + 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 */ @@ -2414,20 +2605,30 @@ static int inputfile_is_done(InputFile *ipf) { return 1; } -static void notice_processed(InputFile *ipf, const char *what, - const char *spec) { +static void notice_processed(InputFile *ipf, int completed, + const char *what, const char *spec) { + if (!ipf) return; /* allows preterminate to be lazy */ + #define RCI_NOTHING(x) /* nothing */ #define RCI_TRIPLE_FMT(x) " " #x "=" RCI_TRIPLE_FMT_BASE #define RCI_TRIPLE_VALS(x) , RCI_TRIPLE_VALS_BASE(ipf->counts, [RC_##x]) #define CNT(art,rc) (ipf->counts[art_##art][RC_##rc]) - info("processed %s%s read=%d (+bl=%d,+err=%d)" + char *inprog= completed + ? xasprintf("%s","") /* GCC produces a stupid warning for printf("") ! */ + : xasprintf(" inprogress=%ld", ipf->inprogress); + char *autodefer= ipf->autodefer >= 0 + ? xasprintf(" autodeferred=%ld", ipf->autodefer) + : xasprintf("%s",""); + + info("%s %s%s read=%d (+bl=%d,+err=%d)%s%s" " offered=%d (ch=%d,nc=%d) accepted=%d (ch=%d,nc=%d)" RESULT_COUNTS(RCI_NOTHING, RCI_TRIPLE_FMT) , - what, spec, + completed?"completed":"processed", what, spec, ipf->readcount_ok, ipf->readcount_blank, ipf->readcount_err, + inprog, autodefer, CNT(Unchecked,sent) + CNT(Unsolicited,sent) , CNT(Unchecked,sent), CNT(Unsolicited,sent), CNT(Wanted,accepted) + CNT(Unsolicited,accepted) @@ -2435,6 +2636,9 @@ static void notice_processed(InputFile *ipf, const char *what, RESULT_COUNTS(RCI_NOTHING, RCI_TRIPLE_VALS) ); + free(inprog); + free(autodefer); + #undef CNT } @@ -2447,7 +2651,7 @@ static void statemc_check_backlog_done(void) { const char *under= strchr(slash, '_'); const char *rest= under ? under+1 : leaf; if (!strncmp(rest,"backlog",7)) rest += 7; - notice_processed(ipf,"backlog ",rest); + notice_processed(ipf,1,"backlog ",rest); close_input_file(ipf); if (unlink(ipf->path)) { @@ -2469,7 +2673,7 @@ static void statemc_check_flushing_done(void) { assert(sms==sm_SEPARATED || sms==sm_DROPPING); - notice_processed(ipf,"feedfile",""); + notice_processed(ipf,1,"feedfile",""); close_defer(); @@ -2483,7 +2687,7 @@ static void statemc_check_flushing_done(void) { notice("flush complete"); SMS(NORMAL, spontaneous_flush_periods, "flush complete"); } else if (sms==sm_DROPPING) { - SMS(DROPPED, 0, "old flush complete"); + SMS(DROPPED, max_separated_periods, "old flush complete"); search_backlog_file(); notice("feed dropped, but will continue until backlog is finished"); } @@ -2504,7 +2708,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) { @@ -2670,6 +2874,7 @@ static void search_backlog_file(void) { debug("backlog scan: none"); if (sms==sm_DROPPED) { + preterminate(); notice("feed dropped and our work is complete"); int r= unlink(path_control); @@ -2716,6 +2921,72 @@ static void search_backlog_file(void) { return; } +/*---------- shutdown and signal handling ----------*/ + +static void preterminate(void) { + if (in_child) return; + notice_processed(main_input_file,0,"feedfile",""); + notice_processed(flushing_input_file,0,"flushing",""); + if (backlog_input_file) + notice_processed(backlog_input_file,0, "backlog file ", + backlog_input_file->path); +} + +static int signal_self_pipe[2]; +static sig_atomic_t terminate_sig_flag; + +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) { + assert(fd=signal_self_pipe[0]); + char buf[PIPE_BUF]; + int r= read(signal_self_pipe[0], buf, sizeof(buf)); + if (r<0 && !isewouldblock(errno)) sysdie("failed to read signal self pipe"); + if (r==0) die("eof on signal self pipe"); + if (terminate_sig_flag) { + preterminate(); + notice("terminating (%s)", strsignal(terminate_sig_flag)); + raise_default(terminate_sig_flag); + } + return OOP_CONTINUE; +} + +static void sigarrived_handler(int signum) { + static char x; + switch (signum) { + case SIGTERM: + case SIGINT: + if (!terminate_sig_flag) terminate_sig_flag= signum; + break; + default: + abort(); + } + write(signal_self_pipe[1],&x,1); +} + +static void init_signals(void) { + if (signal(SIGPIPE, SIG_IGN) == SIG_ERR) + sysdie("could not ignore SIGPIPE"); + + if (pipe(signal_self_pipe)) sysfatal("create self-pipe for signals"); + + xsetnonblock(signal_self_pipe[0],1); + xsetnonblock(signal_self_pipe[1],1); + + struct sigaction sa; + memset(&sa,0,sizeof(sa)); + sa.sa_handler= sigarrived_handler; + sa.sa_flags= SA_RESTART; + xsigaction(SIGTERM,&sa); + xsigaction(SIGINT,&sa); + + on_fd_read_except(signal_self_pipe[0], sigarrived_event); +} + /*========== flushing the feed ==========*/ static pid_t inndcomm_child; @@ -2750,7 +3021,8 @@ static void *inndcomm_event(oop_source *lp, int fd, oop_event e, void *u) { main_input_file= 0; if (flushing_input_file) { - SMS(DROPPING, 0, "feed dropped by innd, but must finish last flush"); + SMS(DROPPING, max_separated_periods, + "feed dropped by innd, but must finish last flush"); } else { close_defer(); SMS(DROPPED, 0, "feed dropped by innd"); @@ -2768,7 +3040,7 @@ static void *inndcomm_event(oop_source *lp, int fd, oop_event e, void *u) { die("flush succeeded but feedfile %s does not exist!", feedfile); if (flushing_input_file) { - SMS(SEPARATED, spontaneous_flush_periods, "recovery flush complete"); + SMS(SEPARATED, max_separated_periods, "recovery flush complete"); } else { close_defer(); SMS(NORMAL, spontaneous_flush_periods, "flush complete"); @@ -2858,14 +3130,18 @@ static void postfork_stdio(FILE *f, const char *what, const char *what2) { } static void postfork(void) { - if (signal(SIGPIPE, SIG_DFL) == SIG_ERR) - sysdie("(in child) failed to reset SIGPIPE"); + in_child= 1; + + 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); @@ -2895,7 +3171,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; @@ -2916,10 +3192,12 @@ static char *debug_report_ipf(InputFile *ipf) { const char *slash= strrchr(ipf->path,'/'); const char *path= slash ? slash+1 : ipf->path; - return xasprintf("%p/%s:ip=%ld,off=%ld,fd=%d%s", + return xasprintf("%p/%s:queue=%d,ip=%ld,autodef=%ld,off=%ld,fd=%d%s%s", ipf, path, - ipf->inprogress, (long)ipf->offset, - ipf->fd, ipf->rd ? "" : ",!rd"); + ipf->queue.count, ipf->inprogress, ipf->autodefer, + (long)ipf->offset, ipf->fd, + ipf->rd ? "" : ",!rd", + ipf->skippinglong ? "*skiplong" : ""); } static void period(void) { @@ -2928,12 +3206,11 @@ static void period(void) { char *dipf_backlog= debug_report_ipf(backlog_input_file); debug("PERIOD" - " sms=%s[%d] conns=%d queue=%d until_connect=%d" + " sms=%s[%d] conns=%d until_connect=%d" " input_files main:%s flushing:%s backlog:%s" " children connecting=%ld inndcomm=%ld" , - sms_names[sms], sm_period_counter, - conns.count, queue.count, until_connect, + sms_names[sms], until_flush, conns.count, until_connect, dipf_main, dipf_flushing, dipf_backlog, (long)connecting_child, (long)inndcomm_child ); @@ -2944,6 +3221,7 @@ static void period(void) { if (until_connect) until_connect--; + inputfile_queue_check_expired(backlog_input_file); poll_backlog_file(); if (!backlog_input_file) close_defer(); /* want to start on a new backlog */ statemc_period_poll(); @@ -2952,6 +3230,132 @@ static void period(void) { } +/*========== dumping state ==========*/ + +static void dump_article_list(FILE *f, const ControlCommand *c, + const ArticleList *al) { + fprintf(f, " count=%d\n", al->count); + if (!c->xval) return; + + int i; Article *art; + for (i=0, art=LIST_HEAD(*al); art; i++, art=LIST_NEXT(art)) { + fprintf(f," #%05d %-11s", i, artstate_names[art->state]); + DUMPV("%p", art->,ipf); + DUMPV("%d", art->,missing); + DUMPV("%lu", (unsigned long)art->,offset); + DUMPV("%d", art->,blanklen); + DUMPV("%d", art->,midlen); + fprintf(f, " %s %s\n", TokenToText(art->token), art->messageid); + } +} + +static void dump_input_file(FILE *f, const ControlCommand *c, + InputFile *ipf, const char *wh) { + char *dipf= debug_report_ipf(ipf); + fprintf(f,"input %s %s", wh, dipf); + free(dipf); + + if (ipf) { + DUMPV("%d", ipf->,readcount_ok); + DUMPV("%d", ipf->,readcount_blank); + DUMPV("%d", ipf->,readcount_err); + } + fprintf(f,"\n"); + if (ipf) { + ArtState state; const char *const *statename; + for (state=0, statename=artstate_names; *statename; state++,statename++) { +#define RC_DUMP_FMT(x) " " #x "=%d" +#define RC_DUMP_VAL(x) ,ipf->counts[state][RC_##x] + fprintf(f,"input %s counts %-11s" + RESULT_COUNTS(RC_DUMP_FMT,RC_DUMP_FMT) "\n", + wh, *statename + RESULT_COUNTS(RC_DUMP_VAL,RC_DUMP_VAL)); + } + fprintf(f,"input %s queue", wh); + dump_article_list(f,c,&ipf->queue); + } +} + +CCMD(dump) { + int i; + fprintf(cc->out, "dumping state to %s\n", path_dump); + FILE *f= fopen(path_dump, "w"); + if (!f) { fprintf(cc->out, "failed: open: %s\n", strerror(errno)); return; } + + fprintf(f,"general"); + DUMPV("%s", sms_names,[sms]); + DUMPV("%d", ,until_flush); + DUMPV("%ld", (long),self_pid); + DUMPV("%p", , defer); + DUMPV("%d", , until_connect); + DUMPV("%d", , until_backlog_nextscan); + DUMPV("%d", , simulate_flush); + fprintf(f,"\nnocheck"); + DUMPV("%#.10f", , accept_proportion); + DUMPV("%d", , nocheck); + DUMPV("%d", , nocheck_reported); + fprintf(f,"\n"); + + fprintf(f,"special"); + DUMPV("%ld", (long),connecting_child); + DUMPV("%d", , connecting_fdpass_sock); + DUMPV("%d", , control_master); + fprintf(f,"\n"); + + fprintf(f,"filemon "); + filemon_method_dump_info(f); + + dump_input_file(f,c, main_input_file, "main" ); + dump_input_file(f,c, flushing_input_file, "flushing"); + dump_input_file(f,c, backlog_input_file, "backlog" ); + + fprintf(f,"conns count=%d\n", conns.count); + + Conn *conn; + FOR_CONN(conn) { + + 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->,since_activity); + fprintf(f,"\n"); + + fprintf(f,"C%d waiting", conn->fd); dump_article_list(f,c,&conn->waiting); + fprintf(f,"C%d priority",conn->fd); dump_article_list(f,c,&conn->priority); + fprintf(f,"C%d sent", conn->fd); dump_article_list(f,c,&conn->sent); + + fprintf(f,"C%d xmit xmitu=%d\n", conn->fd, conn->xmitu); + for (i=0; ixmitu; i++) { + const struct iovec *iv= &conn->xmit[i]; + const XmitDetails *xd= &conn->xmitd[i]; + char *dinfo; + switch (xd->kind) { + case xk_Const: dinfo= xasprintf("Const"); break; + case xk_Artdata: dinfo= xasprintf("A%p", xd->info.sm_art); break; + default: + abort(); + } + fprintf(f," #%03d %-11s l=%d %s\n", i, dinfo, iv->iov_len, + sanitise(iv->iov_base, iv->iov_len)); + free(dinfo); + } + } + + fprintf(f,"paths"); + DUMPV("%s", , path_lock); + DUMPV("%s", , path_flushing); + DUMPV("%s", , path_defer); + DUMPV("%s", , path_control); + DUMPV("%s", , path_dump); + DUMPV("%s", , globpat_backlog); + fprintf(f,"\n"); + + if (!!ferror(f) + !!fclose(f)) { + fprintf(cc->out, "failed: write: %s\n", strerror(errno)); + return; + } +} + /*========== option parsing ==========*/ static void vbadusage(const char *fmt, va_list al) NORET_PRINTF(1,0); @@ -3145,6 +3549,7 @@ static const Option innduct_options[]= { {0,"earliest-deferred-retry","PERIOD", &backlog_retry_minperiods, op_seconds }, {0,"backlog-rescan-interval","PERIOD",&backlog_spontrescan_periods,op_seconds}, {0,"max-flush-interval", "PERIOD", &spontaneous_flush_periods,op_seconds }, +{0,"flush-finish-timeout", "PERIOD", &max_separated_periods, op_seconds }, {0,"idle-timeout", "PERIOD", &need_activity_periods, op_seconds }, {0,"max-bad-input-data-ratio","PERCENT", &max_bad_data_ratio, op_double }, @@ -3208,6 +3613,7 @@ int main(int argc, char **argv) { convert_to_periods_rndup(&backlog_retry_minperiods); convert_to_periods_rndup(&backlog_spontrescan_periods); convert_to_periods_rndup(&spontaneous_flush_periods); + convert_to_periods_rndup(&max_separated_periods); convert_to_periods_rndup(&need_activity_periods); if (max_bad_data_ratio < 0 || max_bad_data_ratio > 100) @@ -3234,17 +3640,14 @@ int main(int argc, char **argv) { path_flushing= xasprintf("%s_flushing", feedfile); path_defer= xasprintf("%s_defer", feedfile); path_control= xasprintf("%s_control", feedfile); + path_dump= xasprintf("%s_dump", feedfile); globpat_backlog= xasprintf("%s_backlog*", feedfile); oop_source_sys *sysloop= oop_sys_new(); if (!sysloop) sysdie("could not create liboop event loop"); loop= (oop_source*)sysloop; - if (signal(SIGPIPE, SIG_IGN) == SIG_ERR) - sysdie("could not ignore SIGPIPE"); - LIST_INIT(conns); - LIST_INIT(queue); if (become_daemon) { int i; @@ -3275,6 +3678,8 @@ int main(int argc, char **argv) { statemc_lock(); + init_signals(); + notice("starting"); if (!become_daemon)