X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?p=inn-innduct.git;a=blobdiff_plain;f=backends%2Finnduct.c;h=18ce508c2ce93672d57e17411ee786c78a276f02;hp=46d88bc65baf2a72eab1f9911f0ea5acc44708b3;hb=d3eae6b85f6f29f04b2e31b78803d9500a8ff0a5;hpb=aeb62f189a5d76b8ca0d23a46f7917679e081ea0;ds=sidebyside diff --git a/backends/innduct.c b/backends/innduct.c index 46d88bc..18ce508 100644 --- a/backends/innduct.c +++ b/backends/innduct.c @@ -1,15 +1,28 @@ /* - * todo - * - actually do something with readable on control master - * - option for realsockdir - * - option for filepoll - * - option for no inotify - * - manpage: document control master stuff - * - manpage: innconf is used for communicating with innd - * - debug this: - * build-lfs/backends/innduct --no-daemon -f `pwd`/fee sit dom + * debugging rune: + * 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,...]\ @@ -190,6 +203,7 @@ perl -ne 'print if m/-8\<-/..m/-\>8-/; print "\f" if m/-\^L-/' backends/innduct. #include #include #include +#include #include #include @@ -208,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 -----*/ @@ -244,7 +267,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) (list_new(&(l).u.li)) +#define LIST_INIT(l) ((l).count=0, list_new(&(l).u.li)) #define LIST_HEAD(l) ((typeof((l).u.for_type))(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)))) @@ -270,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); @@ -289,6 +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 check_assign_articles(void); static void queue_check_input_done(void); @@ -302,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); @@ -322,7 +352,7 @@ static oop_rd_call peer_rd_err, peer_rd_ok; static const char *sitename, *remote_host; static const char *feedfile, *realsockdir="/tmp/innduct.control"; static int quiet_multiple=0; -static int become_daemon=1; +static int become_daemon=1, try_filemon=1; static int try_stream=1; static int port=119; static const char *inndconffile; @@ -331,6 +361,7 @@ 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 filepoll_seconds=5; static int connection_setup_timeout=200; static int inndcomm_flush_timeout=100; @@ -358,18 +389,22 @@ typedef enum { /* in queue in conn->sent */ art_Unchecked, /* not checked, not sent checking */ art_Wanted, /* checked, wanted sent body as requested */ art_Unsolicited, /* - sent body without check */ - art_MaxState + art_MaxState, } ArtState; +static const char *const artstate_names[]= + { "Unchecked", "Wanted", "Unsolicited", 0 }; + #define RESULT_COUNTS(RCS,RCN) \ RCS(sent) \ RCS(accepted) \ RCN(unwanted) \ RCN(rejected) \ RCN(deferred) \ + RCN(missing) \ RCN(connretry) -#define RCI_TRIPLE_FMT_BASE "%d(id%d+bd%d+nc%d)" +#define RCI_TRIPLE_FMT_BASE "%d (id=%d,bod=%d,nc=%d)" #define RCI_TRIPLE_VALS_BASE(counts,x) \ counts[art_Unchecked] x \ + counts[art_Wanted] x \ @@ -390,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; }; @@ -426,7 +460,7 @@ struct InputFile { struct Article { ISNODE(Article); ArtState state; - int midlen; + int midlen, missing; InputFile *ipf; TOKEN token; off_t offset; @@ -474,20 +508,21 @@ struct Conn { 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; @@ -527,6 +562,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); \ @@ -640,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); @@ -651,6 +693,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"); @@ -698,15 +752,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); @@ -719,7 +774,6 @@ static int isewouldblock(int errnoval) { return errnoval==EWOULDBLOCK || errnoval==EAGAIN; } - /*========== command and control connections ==========*/ static int control_master; @@ -759,7 +813,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, @@ -779,19 +832,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 }, + { "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 "flush", func, &until_flush, 1 }, \ + { cmd "conn", func, &until_connect, 0 }, \ + { cmd "blscan", func, &until_backlog_nextscan, 0 }, +POKES("next ", ccmd_setint) +POKES("prod ", ccmd_setint_period) + { "pretend flush", ccmd_setintarg, &simulate_flush }, - { "poke sm", ccmd_setint, &sm_period_counter, 1 }, - { "poke conn", ccmd_setint, &until_connect, 0 }, - { "poke blscan", ccmd_setint, &until_backlog_nextscan, 0 }, { "wedge blscan", ccmd_setint, &until_backlog_nextscan, -1 }, { 0 } }; @@ -863,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; @@ -885,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); @@ -946,7 +1025,7 @@ static void control_init(void) { uid_t self= geteuid(); if (!S_ISDIR(stab.st_mode) || stab.st_uid != self || - stab.st_mode & 0077) { + stab.st_mode & 0007) { warn("no control socket, because real socket directory" " is somehow wrong (ISDIR=%d, uid=%lu (exp.%lu), mode %lo)", !!S_ISDIR(stab.st_mode), @@ -1038,6 +1117,7 @@ static void *conn_exception(oop_source *lp, int fd, static void vconnfail(Conn *conn, const char *fmt, va_list al) { int requeue[art_MaxState]; + memset(requeue,0,sizeof(requeue)); Article *art; while ((art= LIST_REMHEAD(conn->priority))) LIST_ADDTAIL(queue, art); @@ -1072,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 */ @@ -1135,9 +1215,15 @@ static void connect_attempt_discard(void) { } #define PREP_DECL_MSG_CMSG(msg) \ + char msgbyte= 0; \ + struct iovec msgiov; \ + msgiov.iov_base= &msgbyte; \ + msgiov.iov_len= 1; \ struct msghdr msg; \ memset(&msg,0,sizeof(msg)); \ - char msg##cbuf[CMSG_SPACE(sizeof(fd))]; \ + char msg##cbuf[CMSG_SPACE(sizeof(int))]; \ + msg.msg_iov= &msgiov; \ + msg.msg_iovlen= 1; \ msg.msg_control= msg##cbuf; \ msg.msg_controllen= sizeof(msg##cbuf); @@ -1145,7 +1231,9 @@ static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) { Conn *conn= 0; assert(fd == connecting_fdpass_sock); + PREP_DECL_MSG_CMSG(msg); + ssize_t rs= recvmsg(fd, &msg, 0); if (rs<0) { if (isewouldblock(errno)) return OOP_CONTINUE; @@ -1153,8 +1241,10 @@ 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); struct cmsghdr *h= 0; if (rs >= 0) h= CMSG_FIRSTHDR(&msg); @@ -1215,9 +1305,6 @@ 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); conn->max_queue= conn->stream ? max_queue_per_conn : 1; loop->on_fd(loop, conn->fd, OOP_EXCEPTION, conn_exception, conn); @@ -1233,7 +1320,7 @@ static void *connchild_event(oop_source *lp, int fd, oop_event e, void *u) { connect_attempt_discard(); check_assign_articles(); - return 0; + return OOP_CONTINUE; x: conn_dispose(conn); @@ -1251,7 +1338,7 @@ static void connect_start(void) { assert(!connecting_child); assert(!connecting_fdpass_sock); - notice("starting connection attempt"); + info("starting connection attempt"); int socks[2]; int r= socketpair(AF_UNIX, SOCK_STREAM, 0, socks); @@ -1281,13 +1368,13 @@ 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) sysfatal("connect: authentication failed"); if (try_stream) { - if (fputs("MODE STREAM\r\n", cn_to) || + if (fputs("MODE STREAM\r\n", cn_to)==EOF || fflush(cn_to)) sysfatal("connect: could not send MODE STREAM"); buf[sizeof(buf)-1]= 0; @@ -1299,15 +1386,15 @@ static void connect_start(void) { } int l= strlen(buf); assert(l>=1); - if (buf[-1]!='\n') + 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: @@ -1318,7 +1405,7 @@ static void connect_start(void) { break; default: warn("connect: unexpected response to MODE STREAM: %.50s", - sanitise(buf)); + sanitise(buf,-1)); exitstatus= 2; break; } @@ -1334,7 +1421,8 @@ static void connect_start(void) { msg.msg_controllen= cmsg->cmsg_len; r= sendmsg(socks[1], &msg, 0); - if (r) sysdie("sendmsg failed for new connection"); + if (r<0) sysdie("sendmsg failed for new connection"); + if (r!=1) die("sendmsg for new connection gave wrong result %d",r); _exit(exitstatus); } @@ -1361,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; @@ -1375,6 +1463,7 @@ static void check_assign_articles(void) { if (!inqueue) use->since_activity= 0; /* reset idle counter */ while (spare>0) { Article *art= LIST_REMHEAD(queue); + if (!art) break; LIST_ADDTAIL(use->waiting, art); spare--; } @@ -1440,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(); @@ -1499,12 +1587,23 @@ static void conn_make_some_xmits(Conn *conn) { ARTHANDLE *artdata= SMretrieve(art->token, RETR_ALL); + art->state= + art->state == art_Unchecked ? art_Unsolicited : + art->state == art_Wanted ? art_Wanted : + (abort(),-1); + + if (!artdata) art->missing= 1; + art->ipf->counts[art->state][ artdata ? RC_sent : RC_missing ]++; + if (conn->stream) { if (artdata) { XMIT_LITERAL("TAKETHIS "); xmit_noalloc(conn, art->messageid, art->midlen); XMIT_LITERAL("\r\n"); xmit_artbody(conn, artdata); + } else { + article_done(conn, art, -1); + continue; } } else { /* we got 235 from IHAVE */ @@ -1515,20 +1614,15 @@ static void conn_make_some_xmits(Conn *conn) { } } - art->state= - art->state == art_Unchecked ? art_Unsolicited : - art->state == art_Wanted ? art_Wanted : - (abort(),-1); - art->ipf->counts[art->state][RC_sent]++; LIST_ADDTAIL(conn->sent, art); } else { /* check it */ if (conn->stream) - XMIT_LITERAL("IHAVE "); - else XMIT_LITERAL("CHECK "); + else + XMIT_LITERAL("IHAVE "); xmit_noalloc(conn, art->messageid, art->midlen); XMIT_LITERAL("\r\n"); @@ -1630,7 +1724,8 @@ static void update_nocheck(int accepted) { } static void article_done(Conn *conn, Article *art, int whichcount) { - art->ipf->counts[art->state][whichcount]++; + if (!art->missing) art->ipf->counts[art->state][whichcount]++; + if (whichcount == RC_accepted) update_nocheck(1); else if (whichcount == RC_unwanted) update_nocheck(0); @@ -1638,6 +1733,11 @@ static void article_done(Conn *conn, Article *art, int whichcount) { while (art->blanklen) { static const char spaces[]= + " " + " " + " " + " " + " " " " " " " " @@ -1674,7 +1774,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); @@ -1683,15 +1783,18 @@ static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_rd_event ev, return OOP_CONTINUE; } + int conn_busy= + conn->waiting.count || + conn->priority.count || + conn->sent.count || + conn->xmitu; + if (conn->quitting) { if (code!=205 && code!=503) { connfail(conn, "peer gave unexpected response to QUIT: %s", sani); } else { - notice("C%d idle connection closed", conn->fd); - assert(!conn->waiting.count); - assert(!conn->priority.count); - assert(!conn->sent.count); - assert(!conn->xmitu); + notice("C%d idle connection closed by us", conn->fd); + assert(!conn_busy); LIST_REMOVE(conns,conn); conn_dispose(conn); } @@ -1701,25 +1804,36 @@ static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_rd_event ev, conn->since_activity= 0; Article *art; -#define GET_ARTICLE(musthavesent) \ - art= article_reply_check(conn, data, musthavesent, code_streaming, sani); \ - if (art) ; else return OOP_CONTINUE /* reply_check has failed the conn */ +#define GET_ARTICLE(musthavesent) do{ \ + art= article_reply_check(conn, data, code_streaming, musthavesent, sani); \ + if (!art) return OOP_CONTINUE; /* reply_check has failed the conn */ \ + }while(0) -#define ARTICLE_DEALTWITH(streaming,musthavesent,how) \ - code_streaming= (streaming); \ - GET_ARTICLE(musthavesent); \ - article_done(conn, art, RC_##how); break; +#define ARTICLE_DEALTWITH(streaming,musthavesent,how) do{ \ + code_streaming= (streaming); \ + GET_ARTICLE(musthavesent); \ + article_done(conn, art, RC_##how); \ + goto dealtwith; \ + }while(0) -#define PEERBADMSG(m) connfail(conn, m ": %s", sani); return OOP_CONTINUE +#define PEERBADMSG(m) do { \ + connfail(conn, m ": %s", sani); return OOP_CONTINUE; \ + }while(0) int code_streaming= 0; switch (code) { case 400: PEERBADMSG("peer stopped accepting articles"); - case 503: PEERBADMSG("peer timed us out"); default: PEERBADMSG("peer sent unexpected message"); + case 503: + if (conn_busy) PEERBADMSG("peer timed us out"); + notice("C%d idle connection closed by peer", conn->fd); + LIST_REMOVE(conns,conn); + conn_dispose(conn); + return OOP_CONTINUE; + case 435: ARTICLE_DEALTWITH(0,0,unwanted); /* IHAVE says they have it */ case 438: ARTICLE_DEALTWITH(1,0,unwanted); /* CHECK/TAKETHIS: they have it */ @@ -1751,6 +1865,7 @@ static void *peer_rd_ok(oop_source *lp, oop_read *oread, oop_rd_event ev, break; } +dealtwith: conn_maybe_write(conn); check_assign_articles(); @@ -1776,7 +1891,7 @@ static void feedfile_eof(InputFile *ipf) { } static InputFile *open_input_file(const char *path) { - int fd= open(path, O_RDONLY); + int fd= open(path, O_RDWR); if (fd<0) { if (errno==ENOENT) return 0; sysfatal("unable to open input file %s", path); @@ -1806,7 +1921,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) @@ -1837,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); @@ -1877,11 +1992,12 @@ 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++; art->token= TextToToken(tokentextbuf); - art->offset= ipf->offset; + art->offset= old_offset; art->blanklen= recsz; strcpy(art->messageid, space+1); LIST_ADDTAIL(queue, art); @@ -1959,10 +2075,10 @@ static ssize_t tailing_try_read(struct oop_readable *rable, void *buffer, /*---------- filemon implemented with inotify ----------*/ -#if defined(HAVE_INOTIFY) && !defined(HAVE_FILEMON) +#if defined(HAVE_SYS_INOTIFY_H) && !defined(HAVE_FILEMON) #define HAVE_FILEMON -#include +#include static int filemon_inotify_fd; static int filemon_inotify_wdmax; @@ -1978,7 +2094,7 @@ static void filemon_method_startfile(InputFile *ipf, Filemon_Perfile *pf) { if (wd >= filemon_inotify_wdmax) { int newmax= wd+2; - filemon_inotify_wd= xrealloc(filemon_inotify_wd2ipf, + filemon_inotify_wd2ipf= xrealloc(filemon_inotify_wd2ipf, sizeof(*filemon_inotify_wd2ipf) * newmax); memset(filemon_inotify_wd2ipf + filemon_inotify_wdmax, 0, sizeof(*filemon_inotify_wd2ipf) * (newmax - filemon_inotify_wdmax)); @@ -1997,7 +2113,7 @@ static void filemon_method_startfile(InputFile *ipf, Filemon_Perfile *pf) { static void filemon_method_stopfile(InputFile *ipf, Filemon_Perfile *pf) { int wd= pf->wd; debug("filemon inotify stopfile %p wd=%d", ipf, wd); - int r= inotify_rm_watch(filemon_inotify_fd, filemon_inotify_wd); + int r= inotify_rm_watch(filemon_inotify_fd, wd); if (r) sysdie("inotify_rm_watch"); filemon_inotify_wd2ipf[wd]= 0; } @@ -2016,7 +2132,7 @@ static void *filemon_inotify_readable(oop_source *lp, int fd, die("inotify read %d bytes wanted struct of %d", r, (int)sizeof(iev)); } InputFile *ipf= filemon_inotify_wd2ipf[iev.wd]; - debug("filemon inotify readable read %p wd=%p", iev.wd, ipf); + debug("filemon inotify readable read %p wd=%d", ipf, iev.wd); filemon_callback(ipf); } return OOP_CONTINUE; @@ -2025,16 +2141,25 @@ static void *filemon_inotify_readable(oop_source *lp, int fd, static int filemon_method_init(void) { filemon_inotify_fd= inotify_init(); if (filemon_inotify_fd<0) { - syswarn("could not initialise inotify: inotify_init failed"); + syswarn("filemon/inotify: inotify_init failed"); return 0; } - set nonblock; - loop->on_fd(loop, filemon_inotify_fd, OOP_READ, filemon_inotify_readable); + xsetnonblock(filemon_inotify_fd, 1); + loop->on_fd(loop, filemon_inotify_fd, OOP_READ, filemon_inotify_readable, 0); debug("filemon inotify init filemon_inotify_fd=%d", filemon_inotify_fd); 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); } @@ -2169,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 @@ -2179,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 @@ -2192,7 +2320,7 @@ static void inputfile_reading_stop(InputFile *ipf) { DROPPED [Dropped] main none - old none + flsh none some backlog ============== | @@ -2308,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"); @@ -2326,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", @@ -2339,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 */ @@ -2364,20 +2500,26 @@ 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(+%dbl,+%derr)" - " offered=%d(ch%d,nc%d) accepted=%d(ch%d+nc%d)" + char *inprog= completed + ? xasprintf("%s","") /* GCC produces a stupid warning for printf("") ! */ + : xasprintf(" inprogress=%ld", ipf->inprogress); + + info("%s %s%s read=%d (+bl=%d,+err=%d)%s" + " offered=%d (ch=%d,nc=%d) accepted=%d (ch=%d,nc=%d)" RESULT_COUNTS(RCI_NOTHING, RCI_TRIPLE_FMT) , - what, spec, - ipf->readcount_ok, ipf->readcount_blank, ipf->readcount_err, + completed?"completed":"processed", what, spec, + ipf->readcount_ok, ipf->readcount_blank, ipf->readcount_err, inprog, CNT(Unchecked,sent) + CNT(Unsolicited,sent) , CNT(Unchecked,sent), CNT(Unsolicited,sent), CNT(Wanted,accepted) + CNT(Unsolicited,accepted) @@ -2385,6 +2527,8 @@ static void notice_processed(InputFile *ipf, const char *what, RESULT_COUNTS(RCI_NOTHING, RCI_TRIPLE_VALS) ); + free(inprog); + #undef CNT } @@ -2397,7 +2541,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)) { @@ -2419,7 +2563,7 @@ static void statemc_check_flushing_done(void) { assert(sms==sm_SEPARATED || sms==sm_DROPPING); - notice_processed(ipf,"feedfile",0); + notice_processed(ipf,1,"feedfile",""); close_defer(); @@ -2431,7 +2575,7 @@ static void statemc_check_flushing_done(void) { if (sms==sm_SEPARATED) { notice("flush complete"); - SMS(NORMAL, 0, "flush complete"); + SMS(NORMAL, spontaneous_flush_periods, "flush complete"); } else if (sms==sm_DROPPING) { SMS(DROPPED, 0, "old flush complete"); search_backlog_file(); @@ -2454,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) { @@ -2620,6 +2764,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); @@ -2666,6 +2811,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; @@ -2808,14 +3019,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); @@ -2845,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; @@ -2866,10 +3081,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:ip=%ld,off=%ld,fd=%d%s%s", ipf, path, ipf->inprogress, (long)ipf->offset, - ipf->fd, ipf->rd ? "+" : ""); + ipf->fd, + ipf->rd ? "" : ",!rd", + ipf->skippinglong ? "*skiplong" : ""); } static void period(void) { @@ -2879,10 +3096,10 @@ static void period(void) { debug("PERIOD" " sms=%s[%d] conns=%d queue=%d until_connect=%d" - " input_files main:%s old:%s flushing:%s" + " 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 @@ -2902,6 +3119,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, 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)); + } + } +} + +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, main_input_file, "main" ); + dump_input_file(f, flushing_input_file, "flushing"); + dump_input_file(f, 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; + long diff; + 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,"queue"); dump_article_list(f,c,&queue); + + 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); @@ -3072,8 +3415,10 @@ static const Option innduct_options[]= { {'q',"quiet-multiple", 0, &quiet_multiple, op_setint, 1 }, {0,"no-daemon", 0, &become_daemon, op_setint, 0 }, {0,"no-streaming", 0, &try_stream, op_setint, 0 }, +{0,"no-filemon", 0, &try_filemon, op_setint, 0 }, {'C',"inndconf", "F", &inndconffile, op_string }, {'P',"port", "PORT", &port, op_integer }, +{0,"ctrl-sock-dir", 0, &realsockdir, op_string }, {0,"help", 0, 0, help }, {0,"max-connections", "N", &max_connections, op_integer }, @@ -3081,8 +3426,9 @@ static const Option innduct_options[]= { {0,"feedfile-flush-size","BYTES", &target_max_feedfile_size, op_integer }, {0,"period-interval", "TIME", &period_seconds, op_seconds }, -{0,"connection-timeout", "TIME", &connection_setup_timeout, op_seconds }, -{0,"stuck-flush-timeout","TIME", &inndcomm_flush_timeout, op_seconds }, +{0,"connection-timeout", "TIME", &connection_setup_timeout, op_seconds }, +{0,"stuck-flush-timeout", "TIME", &inndcomm_flush_timeout, op_seconds }, +{0,"feedfile-poll", "TIME", &filepoll_seconds, op_seconds }, {0,"no-check-proportion", "PERCENT", &nocheck_thresh, op_double }, {0,"no-check-response-time","ARTICLES", &nocheck_decay, op_double }, @@ -3181,15 +3527,13 @@ 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); @@ -3222,6 +3566,8 @@ int main(int argc, char **argv) { statemc_lock(); + init_signals(); + notice("starting"); if (!become_daemon) @@ -3229,10 +3575,16 @@ int main(int argc, char **argv) { control_init(); - if (!filemon_method_init()) { - warn("no file monitoring available, polling"); - every(5,0,filepoll); + int filemon_ok= 0; + if (!try_filemon) { + notice("filemon: suppressed by command line option, polling"); + } else { + filemon_ok= filemon_method_init(); + if (!filemon_ok) + warn("filemon: no file monitoring available, polling"); } + if (!filemon_ok) + every(filepoll_seconds,0,filepoll); every(period_seconds,1,period);