* If we don't have a backlog file that we're reading, we close the
* defer file that we're writing and make it into a backlog file at
* the first convenient opportunity.
+ * -8<-
OVERALL STATES:
"duct reading" means innduct is reading the file but also
overwriting processed tokens.
+ * ->8- -^L-
*
+ * rune for printing diagrams:
+
+perl -ne 'print if m/-8\<-/..m/-\>8-/; print "\f" if m/-\^L-/' backends/innduct.c |a2ps -R -B -ops
+
*
*/
} InputFile;
#define SMS_LIST(X) \
- X(WAITING) \
X(NORMAL) \
X(FLUSHING) \
- X(FLUSHFAIL) \
+ X(FLUSHFAILED) \
X(SEPARATED) \
- X(DROPPING)
+ X(DROPPING) \
+ X(DROPPED)
typedef enum {
#define SMS_DEF_ENUM(s) sm_##s,
ipf->inprogress--;
assert(ipf->inprogress >= 0);
- if (!ipf->inprogress)
- loop->on_time(loop, OOP_TIME_NOW, statemc_check_input_done, ipf);
+ if (!ipf->inprogress && ipf != main_input_file)
+ queue_check_input_done();
free(art);
}
static void feedfile_eof(InputFile *ipf) {
assert(ipf != main_input_file); /* promised by tailing_try_read */
- inputfile_tailing_stop(ipf);
-
- if (ipf == backlog_input_file) {
- assert(ipf->fd >= 0);
- if (close(ipf->fd)) sysdie("could not close backlog file %s", ipf->path);
- ipf->fd= -1;
- return;
- }
-
- assert(ipf == flushing_input_file);
-
inputfile_tailing_stop(ipf);
assert(ipf->fd >= 0);
if (close(ipf->fd)) sysdie("could not close input file %s", ipf->path);
ipf->fd= -1;
- assert(sms==sm_SEPARATED || sms==sm_DROPPING);
-
- if (main_input_file)
- inputfile_tailing_start(main_input_file);
+ if (ipf == flushing_input_file) {
+ assert(sms==sm_SEPARATED || sms==sm_DROPPING);
+ if (main_input_file) inputfile_tailing_start(main_input_file);
+ statemc_check_flushing_done();
+ } else if (ipf == backlog_input_file) {
+ statemc_check_backlog_done();
+ } else {
+ abort(); /* supposed to wait rather than get EOF on main input file */
+ }
}
static InputFile *open_input_file(const char *path) {
if (ipf->fd >= 0)
if (close(ipf->fd)) sysdie("could not close input file %s", ipf->path);
-
- free(ipf);
}
ipf->offset += recsz + 1;
if (sms==sm_NORMAL && ipf==main_input_file &&
- (ipf->offset >= flush_threshold || !until_spontaneous_flush) {
+ ipf->offset >= flush_threshold)
+ statemc_start_flush("feed file size");
+
+ check_master_queue();
+}
- notice("starting flush (%lu >= %lu)",
- (unsigned long)ipf->offset, (unsigned long)flush_threshold);
+static void statemc_start_flush(const char *why) { /* Normal => Flushing */
+ assert(sms == sm_NORMAL);
- int r= link(feedfile, duct_path);
- if (r) sysdie("link feedfile %s to flushing file %s", feedfile, dut_path);
- /* => Hardlinked */
+ debug("starting flush (%s) (%lu >= %lu) (%d)",
+ why,
+ (unsigned long)ipf->offset, (unsigned long)flush_threshold,
+ sm_period_counter);
- xunlink(feedfile, "old feedfile link");
- /* => Moved */
+ int r= link(feedfile, duct_path);
+ if (r) sysdie("link feedfile %s to flushing file %s", feedfile,
+ path_duct);
+ /* => Hardlinked */
- spawn_inndcomm_flush(); /* => Flushing, sets sms to sm_FLUSHING */
- }
+ xunlink(feedfile, "old feedfile link");
+ /* => Moved */
- check_master_queue();
+ spawn_inndcomm_flush(why); /* => Flushing FLUSHING */
}
-
/*========== tailing input file ==========*/
static void filemon_start(InputFile *ipf) {
ipf->readable_callback= 0;
}
+static void tailing_queue_readable(InputFile *ipf) {
+ /* lifetime of ipf here is OK because destruction will cause
+ * on_cancel which will cancel this callback */
+ loop->on_time(loop, OOP_TIME_NOW, tailing_rable_call_time, ipf);
+}
+
static int tailing_on_readable(struct oop_readable *rable,
oop_readable_call *cb, void *user) {
InputFile *ipf= (void*)rable;
ipf->readable_callback_user= user;
filemon_startfile(ipf);
- loop->on_time(loop, OOP_TIME_NOW, tailing_rable_call_time, ipf);
+ tailing_queue_readable(ipf);
return 0;
}
/* See official state diagram at top of file. We implement
* this as follows:
+ * -8<-
.=======.
||START||
| | |
| | spawn inndcomm flush |
| V |
- | ================ |
- | FLUSHING |
+ | ================== |
+ | FLUSHING[-ABSENT] |
| [Flushing] |
| main D tail/none |
- | ================ |
+ | ================== |
| | |
| | INNDCOMM FLUSH FAILS ^
| |`----------------------->----------. |
| | | |
| | NO SUCH SITE V |
- ^ |`--------------->----. ================ |
- | | \ FLUSHFAIL |
+ ^ |`--------------->----. ==================== |
+ | | \ FLUSHFAILED[-ABSENT] |
| | \ [Moved] |
| | FLUSH OK \ main D tail/none |
- | | open F \ ================ |
+ | | open F \ ==================== |
| | \ | |
| | \ | TIME TO RETRY |
- |<--------'| ,---<---'\ `----------------'
- | D NONE | | D NONE \
- | | | \
- | | | |
- | | V |
- | V | V
- | ============= | ============
- | SEPARATED/ | DROPPING/
- | flsh->fd>=0 | flsh->fd>=0
- | [Separated] | [Dropping]
- | main F idle | main none
- | old D tail | old D tail
- | ============= | ============
- | | | |
- ^ | EOF ON D | | EOF ON D
- | V | V
- | =============== | ===============
- | SEPARATED/ | DROPPING/
- | flsh->fd==-1 V flsh->fd==-1
- | [Finishing] | [Dropping]
- | main F tail `. main none
- | old D closed `. old D closed
- | =============== `. ===============
- | | `. |
- | | ALL D PROCESSED `. | ALL D PROCESSED
- | V install defer as backlog `. V install defer as backlog
- ^ | close D \| close D
- | | unlink D | unlink D
- | | | unlink lock
- | | | exit
- `----------' V
- ==========
- (ESRCH)
- [Droppped]
- ==========
+ | |`------->----. ,---<---'\ `----------------'
+ | | D NONE | | D NONE `----.
+ | V | | V
+ | ============= V V ============
+ | SEPARATED-1 | | DROPPING-1
+ | flsh->fd>=0 | | flsh->fd>=0
+ | [Separated] | | [Dropping]
+ | main F idle | | main none
+ | old D tail | | old D tail
+ | ============= | | ============
+ | | | | install |
+ ^ | EOF ON D | | defer | EOF ON D
+ | V | | V
+ | =============== | | ===============
+ | SEPARATED-2 | | DROPPING-2
+ | flsh->fd==-1 | V flsh->fd==-1
+ | [Finishing] | | [Dropping]
+ | main F tail | `. main none
+ | old D closed | `. old D closed
+ | =============== V `. ===============
+ | | `. |
+ | | ALL D PROCESSED `. | ALL D PROCESSED
+ | V install defer as backlog `. | install defer
+ ^ | close D `. | close D
+ | | unlink D `. | unlink D
+ | | | |
+ | | V V
+ `----------' ==============
+ DROPPED
+ [Dropped]
+ main none
+ old none
+ some backlog
+ ==============
+ |
+ | ALL BACKLOG DONE
+ |
+ | unlink lock
+ | exit
+ V
+ ==========
+ (ESRCH)
+ [Droppped]
+ ==========
+ * ->8-
*/
static void statemc_init(void) {
struct stat stab, stabf;
- int noent;
path_lock= xasprintf("%s_lock", feedfile);
path_flushing= xasprintf("%s_flushing", feedfile);
}
xfstat_isreg(lockfd, &stabf, "lockfile");
- xlstat_isreg(path_lock, &stab, &noent, "lockfile");
+ int lock_noent;
+ xlstat_isreg(path_lock, &stab, &lock_noent, "lockfile");
- if (!noent && samefile(&stab, &stabf))
+ if (!lock_noent && samefile(&stab, &stabf))
break;
if (close(lockfd))
search_backlog_file();
- xlstat_isreg(path_defer, &stab, &noent, "defer file");
- if (noent) {
+ int defer_noent;
+ xlstat_isreg(path_defer, &stab, &defer_noent, "defer file");
+ if (defer_noent) {
debug("startup: ductdefer ENOENT");
} else {
debug("startup: ductdefer nlink=%ld", (long)stab.st_nlink);
}
}
- InputFile *file_d= open_input_file(path_flushing);
-
- if (file_d) {
- struct stat stab_f, stab_d;
+ struct stat stab_f, stab_d;
+ int noent_f;
- xlstat_isreg(feedfile, &stab_f, &noent, "feed file");
- if (noent) {
- debug("startup: D exists, F ENOENT => Moved");
- goto found_moved;
- }
+ InputFile *file_d= open_input_file(path_flushing);
+ if (file_d) xfstat_isreg(file_d->fd, &stab_d, "flushing file");
- debug("startup: F and D both exist");
+ xlstat_isreg(feedfile, &stab_f, &noent_f, "feedfile");
- xfstat_isreg(file_d->fd, &stab_d, "flushing file");
+ if (!noent_f && file_d && samefile(&stab_f, &stab_d)) {
+ debug("startup: F==D => Hardlinked");
+ xunlink(feedfile, "feed file (during startup)"); /* => Moved */
+ noent_f= 1;
+ }
- if (samefile(&stab_d, &stab_f)) {
- debug("startup: F==D => Hardlinked");
- xunlink(path_flushing, "feed file (during startup)");
- found_moved:
- debug(" => Moved");
+ if (noent_f) {
+ debug("startup: F ENOENT => Moved");
+ if (file_d) startup_set_input_file(file_d);
+ spawn_inndcomm_flush("feedfile missing at startup");
+ /* => Flushing, sms:=FLUSHING */
+ } else {
+ if (file_d) {
+ debug("startup: F!=D => Separated");
startup_set_input_file(file_d);
- spawn_inndcomm_flush(); /* => Flushing, sets sms to sm_FLUSHING */
- } else {
- debug("F!=D => Separated");
SMS(SEPARATED, 0, "found both old and current feed files");
- startup_set_input_file(file_d);
+ } else {
+ debug("startup: F exists, D ENOENT => Normal");
+ FILE *file_f= open_input_file(feedfile);
+ if (!file_f) die("feed file vanished during startup");
+ startup_set_input_file(file_f);
+ SMS(NORMAL, flushfail_retry_periods, "normal startup");
}
- } else {
- debug("startup: D ENOENT => Nothing");
- SMS(WAITING, open_wait_periods, "no feed file currently exists");
}
}
-static void statemc_poll(void) {
- if (sms==sm_WAITING) { statemc_waiting_poll(); return; }
-
+static void statemc_period_poll(void) {
if (!sm_period_counter) return;
sm_period_counter--;
assert(sm_period_counter>=0);
if (sm_period_counter) return;
switch (sms) {
- case sm_WAITING:
- fatal("timed out waiting for innd to create feed file %s", feedfile);
- case sm_FLUSHFAIL:
- spawn_inndcomm_flush(void);
+ case sm_NORMAL:
+ statemc_start_flush("periodic"); /* Normal => Flushing; => FLUSHING */
+ break;
+ case sm_FLUSHFAILED:
+ spawn_inndcomm_flush("retry"); /* Moved => Flushing; => FLUSHING */
break;
default:
abort();
}
}
-static void statemc_waiting_poll(void) {
- InputFile *file_f= open_input_file(feedfile);
- if (!file_f) return;
- startup_set_input_file(file_d);
- SMS(NORMAL, 0, "found and opened feed file");
-}
-
static void startup_set_input_file(InputFile *f) {
assert(!main_input_file);
main_input_file= f;
- until_spontaneous_flush= spontaneous_flush_periods;
inputfile_tailing_start(f);
}
-static void *statemc_check_input_done(oop_source *lp,
- struct timeval now, void *ipf_v) {
- InputFile *ipf= ipf_v;
- struct stat stab;
-
- if (ipf->inprogress) return; /* new article in the meantime */
- if (ipf->fd >= 0); return; /* not had EOF */
-
- if (ipf == backlog_input_file) {
- notice_processed(ipf,"backlog file",ipf->path);
- close_input_file(ipf);
- if (unlink(ipf->path)) {
- if (errno != ENOENT)
- sysdie("could not unlink processed backlog file %s", ipf->path);
- warn("backlog file %s vanished while we were reading it"
- " so we couldn't remove it (but it's done now, anyway)",
- ipf->path);
- }
- backlog_input_file= 0;
- search_backlog_file();
- return;
+static int inputfile_is_done(InputFile *ipf) {
+ if (!ipf) return 0;
+ if (ipf->inprogress) return 0; /* new article in the meantime */
+ if (ipf->fd >= 0); return 0; /* not had EOF */
+ return 1;
+}
+
+static void statemc_check_backlog_done(void) {
+ InputFile *ipf= backlog_input_file();
+ if (!inputfile_is_done(ipf)) return;
+
+ notice_processed(ipf,"backlog file",ipf->path);
+ close_input_file(ipf);
+ if (unlink(ipf->path)) {
+ if (errno != ENOENT)
+ sysdie("could not unlink processed backlog file %s", ipf->path);
+ warn("backlog file %s vanished while we were reading it"
+ " so we couldn't remove it (but it's done now, anyway)",
+ ipf->path);
}
+ free(ipf);
+ backlog_input_file= 0;
+ search_backlog_file();
+ return;
+}
+
+static void statemc_check_flushing_done(void) {
+ InputFile *ipf= flushing_input_file;
+ if (!inputfile_is_done(ipf)) return;
- assert(ipf == flushing_input_file);
assert(sms==sm_SEPARATED || sms==sm_DROPPING);
notice_processed(ipf,"feed file",0);
xunlink(path_flushing, "old flushing file");
- if (sms==sm_DROPPING) {
- if (search_backlog_file()) {
- debug("feed dropped but still backlogs to process");
- return;
- }
- notice("feed dropped and our work is complete");
- xunlink(path_lock, "lockfile for old feed");
- exit(0);
- }
-
- open_defer();
-
close_input_file(flushing_input_file);
+ free(flushing_input_file);
flushing_input_file= 0;
- notice("flush complete");
- SMS(NORMAL, 0, "flush complete");
+ if (sms==sm_SEPARATED) {
+ notice("flush complete");
+ SMS(NORMAL, 0, "flush complete");
+ } else if (sms==sm_DROPPING) {
+ SMS(DROPPED, 0, "old flush complete");
+ search_backlog_file();
+ notice("feed dropped, but will continue until backlog is finished");
+ }
+}
+
+static void *statemc_check_input_done(oop_source *lp, struct timeval now,
+ void *u) {
+ assert(!inputfile_is_done(main_input_file));
+ statemc_check_flushing_done();
+ statemc_check_backlog_done();
+ return OOP_CONTINUE;
+}
+
+static 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) {
sms= newsms;
sm_period_counter= periods;
+
+ const char *xtra= "";
+ switch (sms) {
+ case sm_FLUSHING: sm_FLUSHFAILED:
+ if (!main_input_file) xtra= "-ABSENT";
+ break;
+ case sm_SEPARATED: case sm_DROPPING:
+ xtra= flushing_input_file->fd >= 0 ? "-1" : "-2";
+ break;
+ default:;
+ }
+
if (periods) {
- info("%s[%d] %s",periods,forlog,why);
+ info("%s%s[%d] %s",forlog,xtra,periods,why);
} else {
- info("%s %s",forlog,why);
+ info("%s%s %s",forlog,xtra,why);
}
}
search_backlog_file();
}
-static int search_backlog_file(void) {
+static void search_backlog_file(void) {
/* returns non-0 iff there are any backlog files */
glob_t gl;
if (!oldest_path) {
debug("backlog scan: none");
+
+ if (sms==sm_DROPPED) {
+ notice("feed dropped and our work is complete");
+ xunlink(path_lock, "lockfile for old feed");
+ exit(0);
+ }
until_backlog_nextscan= backlog_spontaneous_rescan_periods;
return 0;
}
case INNDCOMMCHILD_ESTATUS_NONESUCH:
warn("feed has been dropped by innd, finishing up");
flushing_input_file= main_input_file;
+ tailing_queue_readable(flushing_input_file);
+ /* we probably previously returned EAGAIN from our fake read method
+ * when in fact we were at EOF, so signal another readable event
+ * so we actually see the EOF */
+
main_input_file= 0;
- SMS(DROPPING, 0, "dropped by innd");
+
+ if (flushing_input_file) {
+ SMS(DROPPING, 0, "feed dropped by innd, but must finish last flush");
+ } else {
+ close_defer();
+ SMS(DROPPED, 0, "feed dropped by innd");
+ search_backlog_file();
+ }
return OOP_CONTINUE;
case 0:
+ /* as above */
flushing_input_file= main_input_file;
+ tailing_queue_readable(flushing_input_file);
+
main_input_file= open_input_file(feedfile);
if (!main_input_file)
die("flush succeeded but feedfile %s does not exist!", feedfile);
- until_spontaneous_flush= spontaneous_flush_periods;
- SMS(SEPARATED, 0, "feed file missing");
+
+ if (flushing_input_file) {
+ SMS(SEPARATED, spontaneous_flush_periods, "recovery flush complete");
+ } else {
+ close_defer();
+ SMS(NORMAL, spontaneous_flush_periods, "flush complete");
+ }
return OOP_CONTINUE;
default:
}
failed:
- SMS(FLUSHFAIL, flushfail_retry_periods, "flush failed, will retry");
+ SMS(FLUSHFAILED, flushfail_retry_periods, "flush failed, will retry");
}
static void inndcommfail(const char *what) {
exit(INNDCOMMCHILD_ESTATUS_FAIL);
}
-void spawn_inndcomm_flush(void) {
+void spawn_inndcomm_flush(const char *why) { /* Moved => Flushing */
int pipefds[2];
- assert(sms==sm_NORMAL || sms==sm_FLUSHFAIL);
+ notice("flushing %s",why);
+
+ assert(sms==sm_NORMAL || sms==sm_FLUSHFAILED);
assert(!inndcomm_child);
if (pipe(pipefds)) sysdie("create pipe for inndcomm child sentinel");
int sentinel_fd= pipefds[0];
on_fd_read_except(sentinel_fd, inndcomm_event);
- SMS(FLUSHING, 0, "flush is in progress");
+ SMS(FLUSHING, 0, why);
}
/*========== main program ==========*/
EVERY(period, {PERIOD_SECONDS,0}, {
debug("PERIOD"
- " sms=%s queue=%d sm_period_counter=%d"
- " connect_delay=%d until_spontaneous_flush=%d"
+ " sms=%s[%d] queue=%d connect_delay=%d"
" input_files" DEBUGF_IPF(main) DEBUGF_IPF(old) DEBUGF_FMT(flushing)
" conns idle=%d working=%d full=%d"
" children connecting=%ld inndcomm_child"
,
- sms_names[sms], queue.count, sm_period_counter,
- connect_delay, until_spontaneous_flush,
+ sms_names[sms], sm_period_counter, queue.count, connect_delay,
DEBUG_IPF(main), DEBUG_IPF(flushing), DEBUG_IPF(flushing),
idle.count, working.count, full.count,
(long)connecting_child, (long)inndcomm_child
);
+
if (connect_delay) connect_delay--;
- if (until_spontaneous_flush) until_spontaneous_flush--;
+
poll_backlog_file();
if (!backlog_input_file) close_defer(); /* want to start on a new backlog */
- statemc_poll();
+ statemc_period_poll();
check_master_queue();
});
{ 0, "max-queue-size", &max_queue_per_conn op_integer },
{ 0, "reconnect-interval", &reconnect_delay_periods, op_periods_rndup },
{ 0, "flush-retry-interval", &flushfail_retry_periods, op_periods_rndup },
-{ 0, "feedfile-open-timeout", &open_wait_periods, op_periods_rndup },
{ 0, "connection-timeout", &connection_timeout, op_seconds },
{ 0, "inndcomm-timeout", &inndcomm_flush_timeout, op_seconds },
};
~ian / innduct.git / blobdiff