fix readable logic

diff --git a/backends/innduct.c b/backends/innduct.c
index c884af08a056441f0e654870b3ee5ec840f57f8d..afe94172da08bae01136d9bd335f96ad1b84cd48 100644 (file)
--- 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) {