X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?p=inn-innduct.git;a=blobdiff_plain;f=backends%2Finnduct.c;h=613a85d6bde8d6344ef35fbcad0677b76d62ea5e;hp=427792089e6ca7dab1c96e51941a6cfab9aa00d4;hb=c6028aa66d8e254a4c52ae376f9f51179a638547;hpb=58574c7dd1bb38e5e4ba4a6c4bd459e9e9a8e625 diff --git a/backends/innduct.c b/backends/innduct.c index 4277920..613a85d 100644 --- a/backends/innduct.c +++ b/backends/innduct.c @@ -1,14 +1,11 @@ /* * todo - * - inotify not working ? - * - some per-conn info thing for control - * - option for realsockdir - * - option for filepoll - * - option for no inotify + * - abolish xk_Malloc + * - rename sm_period_counter as it's just about flushes * - 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 */ /* @@ -210,6 +207,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 -----*/ @@ -272,6 +278,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); @@ -291,6 +298,7 @@ static void statemc_setstate(StateMachineState newsms, int periods, static void statemc_start_flush(const char *why); /* Normal => Flushing */ static void spawn_inndcomm_flush(const char *why); /* Moved => Flushing */ +static int trigger_flush_ok(void); /* => Flushing,FLUSHING, ret 1; or ret 0 */ static void article_done(Conn *conn, Article *art, int whichcount); @@ -306,6 +314,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); @@ -326,7 +337,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; @@ -335,6 +346,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; @@ -365,6 +377,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) \ @@ -479,20 +494,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 InputFile *main_input_file, *flushing_input_file, *backlog_input_file; static int sm_period_counter; +static InputFile *main_input_file, *flushing_input_file, *backlog_input_file; +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; @@ -532,6 +548,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); \ @@ -645,6 +662,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); @@ -656,6 +679,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"); @@ -703,15 +738,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); @@ -724,7 +760,6 @@ static int isewouldblock(int errnoval) { return errnoval==EWOULDBLOCK || errnoval==EAGAIN; } - /*========== command and control connections ==========*/ static int control_master; @@ -764,7 +799,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, @@ -786,17 +820,43 @@ CCMD(help) { fprintf(cc->out, " %s\n", ccmd->cmd); } +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 "sm", func, &sm_period_counter, 1 }, \ + { cmd "conn", func, &until_connect, 0 }, \ + { cmd "blscan", func, &until_backlog_nextscan, 0 }, +POKES("prod ", ccmd_setint_period) +POKES("next ", ccmd_setint) + { "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 } }; @@ -868,8 +928,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; @@ -890,8 +949,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); @@ -951,7 +1009,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), @@ -1078,10 +1136,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 */ @@ -1167,8 +1225,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); @@ -1295,7 +1352,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) @@ -1315,13 +1372,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: @@ -1332,7 +1389,7 @@ static void connect_start(void) { break; default: warn("connect: unexpected response to MODE STREAM: %.50s", - sanitise(buf)); + sanitise(buf,-1)); exitstatus= 2; break; } @@ -1376,7 +1433,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; @@ -1702,7 +1759,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); @@ -1711,15 +1768,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); } @@ -1729,25 +1789,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, code_streaming, musthavesent, 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 */ @@ -1779,6 +1850,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(); @@ -1834,7 +1906,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) @@ -1865,7 +1937,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); @@ -1905,6 +1977,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++; @@ -2044,7 +2117,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 %d wd=%p", iev.wd, ipf); + debug("filemon inotify readable read %p wd=%d", ipf, iev.wd); filemon_callback(ipf); } return OOP_CONTINUE; @@ -2063,6 +2136,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); } @@ -2200,7 +2282,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 @@ -2210,7 +2292,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 @@ -2223,7 +2305,7 @@ static void inputfile_reading_stop(InputFile *ipf) { DROPPED [Dropped] main none - old none + flsh none some backlog ============== | @@ -2339,6 +2421,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"); @@ -2370,24 +2455,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 (!sm_period_counter) return; + sm_period_counter--; + assert(sm_period_counter>=0); + + if (sm_period_counter) 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 */ @@ -2395,20 +2485,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 (+bl=%d,+err=%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) @@ -2416,6 +2512,8 @@ static void notice_processed(InputFile *ipf, const char *what, RESULT_COUNTS(RCI_NOTHING, RCI_TRIPLE_VALS) ); + free(inprog); + #undef CNT } @@ -2428,7 +2526,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)) { @@ -2450,7 +2548,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(); @@ -2651,6 +2749,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); @@ -2697,6 +2796,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; @@ -2839,14 +3004,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); @@ -2876,7 +3045,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; @@ -2897,10 +3066,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 ? "" : ",!rd"); + ipf->fd, + ipf->rd ? "" : ",!rd", + ipf->skippinglong ? "*skiplong" : ""); } static void period(void) { @@ -2933,6 +3104,140 @@ 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", ,sm_period_counter); + 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_Malloc: + diff= xd->info.malloc_tofree - (char*)iv->iov_base; + dinfo= xasprintf("M%5ld", diff); + break; + case xk_Const: + dinfo= xasprintf("Const"); + break; + case xk_Artdata: + dinfo= xasprintf("A%p", xd->info.sm_art); + break; + 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); @@ -3103,8 +3408,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 }, @@ -3112,8 +3419,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 }, @@ -3212,15 +3520,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); @@ -3253,6 +3559,8 @@ int main(int argc, char **argv) { statemc_lock(); + init_signals(); + notice("starting"); if (!become_daemon) @@ -3260,10 +3568,16 @@ int main(int argc, char **argv) { control_init(); - if (!filemon_method_init()) { - warn("filemon: 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);