From a47eafd3860513c1dd61a75abc413e6bd1b5b3f9 Mon Sep 17 00:00:00 2001 From: Ian Jackson Date: Mon, 3 May 2010 15:16:49 +0100 Subject: [PATCH] fix readable logic --- backends/innduct.c | 43 ++++++++++++++++++++++++------------------- 1 file changed, 24 insertions(+), 19 deletions(-) diff --git a/backends/innduct.c b/backends/innduct.c index c884af0..afe9417 100644 --- a/backends/innduct.c +++ b/backends/innduct.c @@ -343,7 +343,7 @@ static void inputfile_reading_resume(InputFile *ipf); static void filemon_start(InputFile *ipf); static void filemon_stop(InputFile *ipf); -static void filemon_callback(InputFile *ipf); +static void tailing_make_readable(InputFile *ipf); static void vconnfail(Conn *conn, const char *fmt, va_list al) PRINTF(2,0); static void connfail(Conn *conn, const char *fmt, ...) PRINTF(2,3); @@ -457,7 +457,7 @@ struct InputFile { oop_read *rd; /* non-0: reading; 0: constructing, or had EOF */ off_t offset; - int skippinglong, paused; + int skippinglong, paused, fake_readable; ArticleList queue; long inprogress; /* includes queue.count and also articles in conns */ @@ -1117,7 +1117,7 @@ static void vconnfail(Conn *conn, const char *fmt, va_list al) { xmit_free(d); char *m= xvasprintf(fmt,al); - warn("C%d connection failed (requeueing " RCI_TRIPLE_FMT_BASE "): %s", + warn("C%d connection failed requeueing " RCI_TRIPLE_FMT_BASE ": %s", conn->fd, RCI_TRIPLE_VALS_BASE(requeue, /*nothing*/), m); free(m); @@ -2125,7 +2125,16 @@ static void *feedfile_got_article(oop_source *lp, oop_read *rd, static void *tailing_rable_call_time(oop_source *loop, struct timeval tv, void *user) { + /* lifetime of ipf here is OK because destruction will cause + * on_cancel which will cancel this callback */ InputFile *ipf= user; + + if (!ipf->fake_readable) return OOP_CONTINUE; + + /* we just keep calling readable until our caller (oop_rd) + * has called try_read, and try_read has found EOF so given EAGAIN */ + loop->on_time(loop, OOP_TIME_NOW, tailing_rable_call_time, ipf); + return ipf->readable_callback(loop, &ipf->readable, ipf->readable_callback_user); } @@ -2138,9 +2147,10 @@ static void tailing_on_cancel(struct oop_readable *rable) { 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 */ +static void tailing_make_readable(InputFile *ipf) { + if (!ipf || !ipf->readable_callback) /* so callers can be naive */ + return; + ipf->fake_readable= 1; loop->on_time(loop, OOP_TIME_NOW, tailing_rable_call_time, ipf); } @@ -2152,8 +2162,7 @@ static int tailing_on_readable(struct oop_readable *rable, ipf->readable_callback= cb; ipf->readable_callback_user= user; filemon_start(ipf); - - tailing_queue_readable(ipf); + tailing_make_readable(ipf); return 0; } @@ -2164,11 +2173,13 @@ static ssize_t tailing_try_read(struct oop_readable *rable, void *buffer, ssize_t r= read(ipf->fd, buffer, length); if (r==-1) { if (errno==EINTR) continue; + ipf->fake_readable= 0; return r; } if (!r) { if (ipf==main_input_file) { errno=EAGAIN; + ipf->fake_readable= 0; return -1; } else if (ipf==flushing_input_file) { assert(ipf->rd); @@ -2179,7 +2190,6 @@ static ssize_t tailing_try_read(struct oop_readable *rable, void *buffer, abort(); } } - tailing_queue_readable(ipf); return r; } } @@ -2244,7 +2254,7 @@ static void *filemon_inotify_readable(oop_source *lp, int fd, } InputFile *ipf= filemon_inotify_wd2ipf[iev.wd]; /*debug("filemon inotify readable read %p wd=%d", ipf, iev.wd);*/ - filemon_callback(ipf); + tailing_make_readable(ipf); } return OOP_CONTINUE; } @@ -2305,11 +2315,6 @@ static void filemon_stop(InputFile *ipf) { ipf->filemon= 0; } -static void filemon_callback(InputFile *ipf) { - if (ipf && ipf->readable_callback) /* so filepoll() can be naive */ - ipf->readable_callback(loop, &ipf->readable, ipf->readable_callback_user); -} - /*---------- interface to start and stop an input file ----------*/ static const oop_rd_style feedfile_rdstyle= { @@ -3048,7 +3053,7 @@ static void *inndcomm_event(oop_source *lp, int fd, oop_event e, void *u) { case INNDCOMMCHILD_ESTATUS_NONESUCH: notice("feed has been dropped by innd, finishing up"); flushing_input_file= main_input_file; - tailing_queue_readable(flushing_input_file); + tailing_make_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 */ @@ -3068,7 +3073,7 @@ static void *inndcomm_event(oop_source *lp, int fd, oop_event e, void *u) { case 0: /* as above */ flushing_input_file= main_input_file; - tailing_queue_readable(flushing_input_file); + tailing_make_readable(flushing_input_file); main_input_file= open_input_file(feedfile); if (!main_input_file) @@ -3217,8 +3222,8 @@ static void every(int interval, int fixed_rate, void (*f)(void)) { } static void filepoll(void) { - filemon_callback(main_input_file); - filemon_callback(flushing_input_file); + tailing_make_readable(main_input_file); + tailing_make_readable(flushing_input_file); } static char *debug_report_ipf(InputFile *ipf) { -- 2.30.2