X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?a=blobdiff_plain;f=statemc.c;h=172bbe8afabf040ac91a9b114267b6b4a920e278;hb=62a5c8052810dc6e5716c1e1e3bf63294a350167;hp=970bb6cb6e4e818f25be8506a627f45a70753914;hpb=f4aee95c41a0d6231d115386b8fbb23f6b8e349a;p=innduct.git diff --git a/statemc.c b/statemc.c index 970bb6c..172bbe8 100644 --- a/statemc.c +++ b/statemc.c @@ -1,3 +1,30 @@ +/* + * innduct + * tailing reliable realtime streaming feeder for inn + * statemc.c - state machine core (see README.states). + * + * Copyright (C) 2010 Ian Jackson + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + * + * (I believe that when you compile and link this as part of the inn2 + * build, with the Makefile runes I have provided, all the libraries + * and files which end up included in innduct are licence-compatible + * with GPLv3. If not then please let me know. -Ian Jackson.) + */ + +#include "innduct.h" /* statemc_init initialises */ @@ -19,7 +46,7 @@ static void startup_set_input_file(InputFile *f) { inputfile_reading_start(f); } -static void statemc_lock(void) { +void statemc_lock(void) { int lockfd; struct stat stab, stabf; @@ -65,7 +92,7 @@ static void statemc_lock(void) { dbg("startup: locked"); } -static void statemc_init(void) { +void statemc_init(void) { struct stat stabdefer; search_backlog_file(); @@ -128,7 +155,7 @@ static void statemc_init(void) { } } -static void statemc_start_flush(const char *why) { /* Normal => Flushing */ +void statemc_start_flush(const char *why) { /* Normal => Flushing */ assert(sms == sm_NORMAL); dbg("starting flush (%s) (%lu >?= %lu) (%d)", @@ -148,7 +175,7 @@ static void statemc_start_flush(const char *why) { /* Normal => Flushing */ spawn_inndcomm_flush(why); /* => Flushing FLUSHING */ } -static int trigger_flush_ok(const char *why) { +int trigger_flush_ok(const char *why) { switch (sms) { case sm_NORMAL: @@ -172,7 +199,7 @@ static int trigger_flush_ok(const char *why) { } } -static void statemc_period_poll(void) { +void statemc_period_poll(void) { if (!until_flush) return; until_flush--; assert(until_flush>=0); @@ -195,9 +222,9 @@ static void notice_processed(InputFile *ipf, int completed, #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 RCI_TRIPLE_VALS(x) ,RCI_TRIPLE_VALS_BASE(ipf->counts.results, [RC_##x]) -#define CNT(art,rc) (ipf->counts[art_##art][RC_##rc]) +#define CNT(art,rc) (ipf->counts.results[art_##art][RC_##rc]) char *inprog= completed ? xasprintf("%s","") /* GCC produces a stupid warning for printf("") ! */ @@ -211,8 +238,9 @@ static void notice_processed(InputFile *ipf, int completed, RESULT_COUNTS(RCI_NOTHING, RCI_TRIPLE_FMT) , completed?"completed":"processed", what, spec, - ipf->readcount_ok, ipf->readcount_blank, ipf->readcount_err, - inprog, autodefer, ipf->count_nooffer_missing, + ipf->counts.events[read_ok], ipf->counts.events[read_blank], + ipf->counts.events[read_err], + inprog, autodefer, ipf->counts.events[nooffer_missing], CNT(Unchecked,sent) + CNT(Unsolicited,sent) , CNT(Unchecked,sent), CNT(Unsolicited,sent), CNT(Wanted,accepted) + CNT(Unsolicited,accepted) @@ -220,13 +248,15 @@ static void notice_processed(InputFile *ipf, int completed, RESULT_COUNTS(RCI_NOTHING, RCI_TRIPLE_VALS) ); + memset(&ipf->counts, 0, sizeof(ipf->counts)); + free(inprog); free(autodefer); #undef CNT } -static void statemc_check_backlog_done(void) { +void statemc_check_backlog_done(void) { InputFile *ipf= backlog_input_file; if (!inputfile_is_done(ipf)) return; @@ -251,7 +281,7 @@ static void statemc_check_backlog_done(void) { return; } -static void statemc_check_flushing_done(void) { +void statemc_check_flushing_done(void) { InputFile *ipf= flushing_input_file; if (!inputfile_is_done(ipf)) return; @@ -286,12 +316,12 @@ static void *statemc_check_input_done(oop_source *lp, struct timeval now, return OOP_CONTINUE; } -static void queue_check_input_done(void) { +void queue_check_input_done(void) { loop->on_time(loop, OOP_TIME_NOW, statemc_check_input_done, 0); } -static void statemc_setstate(StateMachineState newsms, int periods, - const char *forlog, const char *why) { +void statemc_setstate(StateMachineState newsms, int periods, + const char *forlog, const char *why) { sms= newsms; until_flush= periods; @@ -448,23 +478,22 @@ void spawn_inndcomm_flush(const char *why) { /* Moved => Flushing */ /*---------- shutdown and signal handling ----------*/ -static void preterminate(void) { +void preterminate(void) { if (in_child) return; + showstats(); +} + +void showstats(void) { 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); + until_stats_log= stats_log_periods; } static int signal_self_pipe[2]; -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]; @@ -493,7 +522,7 @@ static void sigarrived_handler(int signum) { write(signal_self_pipe[1],&x,1); } -static void init_signals(void) { +void init_signals(void) { if (signal(SIGPIPE, SIG_IGN) == SIG_ERR) syscrash("could not ignore SIGPIPE");