X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?a=blobdiff_plain;f=backends%2Finnduct.c;h=2d17982fc20223b098cbb9e886924ac4f89d9f3c;hb=82c25a0a6fb2b67c5cedfa7bf6a7c20973f9de92;hp=a1c749196099fd480acaf76c822234139a72d4bd;hpb=4ee24c6c7fea8188f9e838d683a1924e694ecc27;p=innduct.git diff --git a/backends/innduct.c b/backends/innduct.c index a1c7491..2d17982 100644 --- a/backends/innduct.c +++ b/backends/innduct.c @@ -1,10 +1,4 @@ /* - * todo - * - reset signals TERM and INT (and HUP) in children - * - * - manpage: document control master stuff - * - admin-initiated flush - * * debugging rune: * build-lfs/backends/innduct --connection-timeout=30 --no-daemon -C ../inn.conf -f `pwd`/fee sit localhost */ @@ -208,6 +202,7 @@ 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))) @@ -298,6 +293,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); @@ -314,7 +310,7 @@ 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); +static void raise_default(int signo) NORET; static char *debug_report_ipf(InputFile *ipf); static void inputfile_reading_start(InputFile *ipf); @@ -409,13 +405,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; }; @@ -500,7 +495,7 @@ static pid_t self_pid; /* statemc_init initialises */ static StateMachineState sms; -static int sm_period_counter; +static int until_flush; static InputFile *main_input_file, *flushing_input_file, *backlog_input_file; static FILE *defer; @@ -678,9 +673,16 @@ static time_t xtime(void) { return now; } -static void xsigaction(int s, const struct sigaction *sa) { - int r= sigaction(s,sa,0); - if (r) sysdie("sigaction failed for \"%s\"", strsignal(s)); +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) { @@ -810,13 +812,14 @@ 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(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(); } -CCMD(dump); +CCMD(flush) { + int ok= trigger_flush_ok(); + if (!ok) fprintf(cc->out,"already flushing (state is %s)\n", sms_names[sms]); +} CCMD(stop) { preterminate(); @@ -825,19 +828,29 @@ CCMD(stop) { 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 }, + { "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 }, @@ -1496,7 +1509,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(); @@ -1960,7 +1972,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)); + memset(art,0,sizeof(*art)); art->state= art_Unchecked; art->midlen= midlen; art->ipf= ipf; ipf->inprogress++; @@ -2265,7 +2277,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 @@ -2275,7 +2287,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 @@ -2288,7 +2300,7 @@ static void inputfile_reading_stop(InputFile *ipf) { DROPPED [Dropped] main none - old none + flsh none some backlog ============== | @@ -2404,6 +2416,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"); @@ -2422,7 +2437,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", @@ -2435,24 +2450,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 */ @@ -2558,7 +2578,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) { @@ -2776,7 +2796,7 @@ static void search_backlog_file(void) { static void preterminate(void) { if (in_child) return; notice_processed(main_input_file,0,"feedfile",""); - notice_processed(flushing_input_file,0,"flushing file",""); + notice_processed(flushing_input_file,0,"flushing",""); if (backlog_input_file) notice_processed(backlog_input_file,0, "backlog file ", backlog_input_file->path); @@ -2786,11 +2806,9 @@ static int signal_self_pipe[2]; static sig_atomic_t terminate_sig_flag; static void raise_default(int signo) { - struct sigaction sa; - memset(&sa,0,sizeof(sa)); - sa.sa_handler= SIG_DFL; - xsigaction(signo,&sa); + xsigsetdefault(signo); raise(signo); + abort(); } static void *sigarrived_event(oop_source *lp, int fd, oop_event e, void *u) { @@ -2803,7 +2821,6 @@ static void *sigarrived_event(oop_source *lp, int fd, oop_event e, void *u) { preterminate(); notice("terminating (%s)", strsignal(terminate_sig_flag)); raise_default(terminate_sig_flag); - abort(); } return OOP_CONTINUE; } @@ -2811,7 +2828,8 @@ static void *sigarrived_event(oop_source *lp, int fd, oop_event e, void *u) { static void sigarrived_handler(int signum) { static char x; switch (signum) { - case SIGINT: case SIGTERM: + case SIGTERM: + case SIGINT: if (!terminate_sig_flag) terminate_sig_flag= signum; break; default: @@ -2983,8 +3001,10 @@ static void postfork_stdio(FILE *f, const char *what, const char *what2) { static void postfork(void) { in_child= 1; - if (signal(SIGPIPE, SIG_DFL) == SIG_ERR) - sysdie("(in child) failed to reset SIGPIPE"); + xsigsetdefault(SIGTERM); + xsigsetdefault(SIGINT); + xsigsetdefault(SIGPIPE); + if (terminate_sig_flag) raise(terminate_sig_flag); postfork_inputfile(main_input_file); postfork_inputfile(flushing_input_file); @@ -3059,7 +3079,7 @@ static void period(void) { " 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 @@ -3130,7 +3150,7 @@ CCMD(dump) { fprintf(f,"general"); DUMPV("%s", sms_names,[sms]); - DUMPV("%d", ,sm_period_counter); + DUMPV("%d", ,until_flush); DUMPV("%ld", (long),self_pid); DUMPV("%p", , defer); DUMPV("%d", , until_connect); @@ -3177,16 +3197,8 @@ CCMD(dump) { 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; + case xk_Const: dinfo= xasprintf("Const"); break; + case xk_Artdata: dinfo= xasprintf("A%p", xd->info.sm_art); break; default: abort(); }