infile.c

   1 /*
   2  *  innduct
   3  *  tailing reliable realtime streaming feeder for inn
   4  *  infile.c - monitoring and handling of input files
   5  *
   6  *  Copyright (C) 2010 Ian Jackson <ijackson@chiark.greenend.org.uk>
   7  *
   8  *  This program is free software: you can redistribute it and/or modify
   9  *  it under the terms of the GNU General Public License as published by
  10  *  the Free Software Foundation, either version 3 of the License, or
  11  *  (at your option) any later version.
  12  *
  13  *  This program is distributed in the hope that it will be useful,
  14  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  *  GNU General Public License for more details.
  17  *
  18  *  You should have received a copy of the GNU General Public License
  19  *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
  20  *
  21  *  (I believe that when you compile and link this as part of the inn2
  22  *  build, with the Makefile runes I have provided, all the libraries
  23  *  and files which end up included in innduct are licence-compatible
  24  *  with GPLv3.  If not then please let me know.  -Ian Jackson.)
  25  */
  26
  27 #include "innduct.h"
  28
  29 /*========== monitoring of input files ==========*/
  30
  31 static void feedfile_eof(InputFile *ipf) {
  32   assert(ipf != main_input_file); /* promised by tailing_try_read */
  33   inputfile_reading_stop(ipf);
  34
  35   if (ipf == flushing_input_file) {
  36     assert(sms==sm_SEPARATED || sms==sm_DROPPING);
  37     if (main_input_file) inputfile_reading_start(main_input_file);
  38     statemc_check_flushing_done();
  39   } else if (ipf == backlog_input_file) {
  40     statemc_check_backlog_done();
  41   } else {
  42     abort(); /* supposed to wait rather than get EOF on main input file */
  43   }
  44 }
  45
  46 InputFile *open_input_file(const char *path) {
  47   int fd= open(path, O_RDWR);
  48   if (fd<0) {
  49     if (errno==ENOENT) return 0;
  50     sysdie("unable to open input file %s", path);
  51   }
  52   assert(fd>0);
  53
  54   InputFile *ipf= xmalloc(sizeof(*ipf) + strlen(path) + 1);
  55   memset(ipf,0,sizeof(*ipf));
  56
  57   ipf->fd= fd;
  58   ipf->autodefer= -1;
  59   LIST_INIT(ipf->queue);
  60   strcpy(ipf->path, path);
  61
  62   return ipf;
  63 }
  64
  65 void close_input_file(InputFile *ipf) { /* does not free */
  66   assert(!ipf->readable_callback); /* must have had ->on_cancel */
  67   assert(!ipf->filemon); /* must have had inputfile_reading_stop */
  68   assert(!ipf->rd); /* must have had inputfile_reading_stop */
  69   assert(!ipf->inprogress); /* no dangling pointers pointing here */
  70   xclose_perhaps(&ipf->fd, "input file ", ipf->path);
  71 }
  72
  73
  74 /*---------- dealing with articles read in the input file ----------*/
  75
  76 static void *feedfile_got_bad_data(InputFile *ipf, off_t offset,
  77                                    const char *data, const char *how) {
  78   warn("corrupted file: %s, offset %lu: %s: in %s",
  79        ipf->path, (unsigned long)offset, how, sanitise(data,-1));
  80   ipf->counts.events[read_err]++;
  81   if (ipf->counts.events[read_err] > max_bad_data_initial +
  82       (ipf->counts.events[read_ok] + ipf->counts.events[read_blank])
  83                                                   / max_bad_data_ratio)
  84     crash("too much garbage in input file!  (%d errs, %d ok, %d blank)",
  85           ipf->counts.events[read_err], ipf->counts.events[read_ok],
  86           ipf->counts.events[read_blank]);
  87   return OOP_CONTINUE;
  88 }
  89
  90 static void *feedfile_read_err(oop_source *lp, oop_read *rd,
  91                                oop_rd_event ev, const char *errmsg,
  92                                int errnoval, const char *data, size_t recsz,
  93                                void *ipf_v) {
  94   InputFile *ipf= ipf_v;
  95   assert(ev == OOP_RD_SYSTEM);
  96   errno= errnoval;
  97   syscrash("error reading input file: %s, offset %lu",
  98            ipf->path, (unsigned long)ipf->offset);
  99 }
 100
 101 static void *feedfile_got_article(oop_source *lp, oop_read *rd,
 102                                   oop_rd_event ev, const char *errmsg,
 103                                   int errnoval, const char *data, size_t recsz,
 104                                   void *ipf_v) {
 105   InputFile *ipf= ipf_v;
 106   Article *art;
 107   char tokentextbuf[sizeof(TOKEN)*2+3];
 108
 109   if (!data) { feedfile_eof(ipf); return OOP_CONTINUE; }
 110
 111   off_t old_offset= ipf->offset;
 112   ipf->offset += recsz + !!(ev == OOP_RD_OK);
 113
 114 #define X_BAD_DATA(m) return feedfile_got_bad_data(ipf,old_offset,data,m);
 115
 116   if (ev==OOP_RD_PARTREC)
 117     feedfile_got_bad_data(ipf,old_offset,data,"missing final newline");
 118     /* but process it anyway */
 119
 120   if (ipf->skippinglong) {
 121     if (ev==OOP_RD_OK) ipf->skippinglong= 0; /* fine now */
 122     return OOP_CONTINUE;
 123   }
 124   if (ev==OOP_RD_LONG) {
 125     ipf->skippinglong= 1;
 126     X_BAD_DATA("overly long line");
 127   }
 128
 129   if (memchr(data,'\0',recsz)) X_BAD_DATA("nul byte");
 130   if (!recsz) X_BAD_DATA("empty line");
 131
 132   if (data[0]==' ') {
 133     if (strspn(data," ") != recsz) X_BAD_DATA("line partially blanked");
 134     ipf->counts.events[read_blank]++;
 135     return OOP_CONTINUE;
 136   }
 137
 138   char *space= strchr(data,' ');
 139   int tokenlen= space-data;
 140   int midlen= (int)recsz-tokenlen-1;
 141   if (midlen <= 2) X_BAD_DATA("no room for messageid");
 142   if (space[1]!='<' || space[midlen]!='>') X_BAD_DATA("invalid messageid");
 143
 144   if (tokenlen != sizeof(TOKEN)*2+2) X_BAD_DATA("token wrong length");
 145   memcpy(tokentextbuf, data, tokenlen);
 146   tokentextbuf[tokenlen]= 0;
 147   if (!IsToken(tokentextbuf)) X_BAD_DATA("token wrong syntax");
 148
 149   ipf->counts.events[read_ok]++;
 150
 151   art= xmalloc(sizeof(*art) - 1 + midlen + 1);
 152   memset(art,0,sizeof(*art));
 153   art->state= art_Unchecked;
 154   art->midlen= midlen;
 155   art->ipf= ipf;  ipf->inprogress++;
 156   art->token= TextToToken(tokentextbuf);
 157   art->offset= old_offset;
 158   art->blanklen= recsz;
 159   strcpy(art->messageid, space+1);
 160
 161   if (ipf->autodefer >= 0) {
 162     article_autodefer(ipf, art);
 163   } else {
 164     LIST_ADDTAIL(ipf->queue, art);
 165
 166     if (ipf==backlog_input_file)
 167       article_check_expired(art);
 168   }
 169
 170   if (sms==sm_NORMAL && ipf==main_input_file &&
 171       ipf->offset >= target_max_feedfile_size)
 172     statemc_start_flush("feed file size");
 173
 174   check_assign_articles(); /* may destroy conn but that's OK */
 175   check_reading_pause_resume(ipf);
 176   return OOP_CONTINUE;
 177 }
 178
 179 /*========== tailing input file ==========*/
 180
 181 static void tailing_rable_on_time(InputFile *ipf);
 182
 183 static void *tailing_rable_call_time(oop_source *lp, struct timeval tv,
 184                                      void *user) {
 185   /* lifetime of ipf here is OK because destruction will cause
 186    * on_cancel which will cancel this callback */
 187   InputFile *ipf= user;
 188
 189   //dbg("**TRACT** ipf=%p called",ipf);
 190   if (!ipf->fake_readable) return OOP_CONTINUE;
 191
 192   /* we just keep calling readable until our caller (oop_rd)
 193    * has called try_read, and try_read has found EOF so given EAGAIN */
 194   //dbg("**TRACT** ipf=%p reschedule",ipf);
 195   tailing_rable_on_time(ipf);
 196
 197   assert(ipf->readable_callback);
 198   return ipf->readable_callback(loop, &ipf->readable,
 199                                 ipf->readable_callback_user);
 200 }
 201
 202 static void tailing_rable_on_time(InputFile *ipf) {
 203   loop->cancel_time(loop, OOP_TIME_NOW, tailing_rable_call_time, ipf);
 204   loop->on_time(loop, OOP_TIME_NOW, tailing_rable_call_time, ipf);
 205   /* on_time is not idempotent - it counts.   So we need this to make
 206    * sure we only have one outstanding, as otherwise our cancel doesn't work */
 207 }
 208
 209 static void tailing_on_cancel(struct oop_readable *rable) {
 210   InputFile *ipf= (void*)rable;
 211   //dbg("**TOR** ipf=%p on_cancel",ipf);
 212
 213   if (ipf->filemon) filemon_stop(ipf);
 214   //dbg("**TRACT** ipf=%p cancel",ipf);
 215   loop->cancel_time(loop, OOP_TIME_NOW, tailing_rable_call_time, ipf);
 216   ipf->readable_callback= 0;
 217 }
 218
 219 void tailing_make_readable(InputFile *ipf) {
 220   //dbg("**TRACT** ipf=%p makereadable rcb=%p",ipf,
 221   //    (void*)ipf?ipf->readable_callback:0);
 222   if (!ipf || !ipf->readable_callback) /* so callers can be naive */
 223     return;
 224   ipf->fake_readable= 1;
 225   tailing_rable_on_time(ipf);
 226 }
 227
 228 static int tailing_on_readable(struct oop_readable *rable,
 229                                 oop_readable_call *cb, void *user) {
 230   InputFile *ipf= (void*)rable;
 231   //dbg("**TOR** ipf=%p on_readable",ipf);
 232
 233   tailing_on_cancel(rable);
 234   ipf->readable_callback= cb;
 235   ipf->readable_callback_user= user;
 236   filemon_start(ipf);
 237   tailing_make_readable(ipf);
 238   return 0;
 239 }
 240
 241 static ssize_t tailing_try_read(struct oop_readable *rable, void *buffer,
 242                                 size_t length) {
 243   InputFile *ipf= (void*)rable;
 244   for (;;) {
 245     ssize_t r= read(ipf->fd, buffer, length);
 246     if (r==-1) {
 247       if (errno==EINTR) continue;
 248       ipf->fake_readable= 0;
 249       return r;
 250     }
 251     if (!r) {
 252       if (ipf==main_input_file) {
 253         errno=EAGAIN;
 254         ipf->fake_readable= 0;
 255         return -1;
 256       } else if (ipf==flushing_input_file) {
 257         assert(ipf->rd);
 258         assert(sms==sm_SEPARATED || sms==sm_DROPPING);
 259       } else if (ipf==backlog_input_file) {
 260         assert(ipf->rd);
 261       } else {
 262         abort();
 263       }
 264     }
 265     //dbg("**TOR** ipf=%p try_read r=%d",ipf,r);
 266     return r;
 267   }
 268 }
 269
 270 /*---------- interface to start and stop an input file ----------*/
 271
 272 static const oop_rd_style feedfile_rdstyle= {
 273   OOP_RD_DELIM_STRIP, '\n',
 274   OOP_RD_NUL_PERMIT,
 275   OOP_RD_SHORTREC_LONG,
 276 };
 277
 278 void inputfile_reading_resume(InputFile *ipf) {
 279   if (!ipf->rd) return;
 280   if (!ipf->paused) return;
 281
 282   int r= oop_rd_read(ipf->rd, &feedfile_rdstyle, MAX_LINE_FEEDFILE,
 283                      feedfile_got_article,ipf, feedfile_read_err, ipf);
 284   if (r) syscrash("unable start reading feedfile %s",ipf->path);
 285
 286   ipf->paused= 0;
 287 }
 288
 289 void inputfile_reading_pause(InputFile *ipf) {
 290   if (!ipf->rd) return;
 291   if (ipf->paused) return;
 292   oop_rd_cancel(ipf->rd);
 293   ipf->paused= 1;
 294 }
 295
 296 void inputfile_reading_start(InputFile *ipf) {
 297   assert(!ipf->rd);
 298   ipf->readable.on_readable= tailing_on_readable;
 299   ipf->readable.on_cancel=   tailing_on_cancel;
 300   ipf->readable.try_read=    tailing_try_read;
 301   ipf->readable.delete_tidy= 0; /* we never call oop_rd_delete_{tidy,kill} */
 302   ipf->readable.delete_kill= 0;
 303
 304   ipf->readable_callback= 0;
 305   ipf->readable_callback_user= 0;
 306
 307   ipf->rd= oop_rd_new(loop, &ipf->readable, 0,0);
 308   assert(ipf->rd);
 309
 310   ipf->paused= 1;
 311   inputfile_reading_resume(ipf);
 312 }
 313
 314 void inputfile_reading_stop(InputFile *ipf) {
 315   assert(ipf->rd);
 316   inputfile_reading_pause(ipf);
 317   oop_rd_delete(ipf->rd);
 318   ipf->rd= 0;
 319   assert(!ipf->filemon); /* we shouldn't be monitoring it now */
 320 }
 321
 322 void filepoll(void) {
 323   tailing_make_readable(main_input_file);
 324   tailing_make_readable(flushing_input_file);
 325 }
 326
 327 char *dbg_report_ipf(InputFile *ipf) {
 328   if (!ipf) return xasprintf("none");
 329
 330   const char *slash= strrchr(ipf->path,'/');
 331   const char *path= slash ? slash+1 : ipf->path;
 332
 333   return xasprintf("%p/%s:queue=%d,ip=%ld,autodef=%ld,off=%ld,fd=%d%s%s%s",
 334                    ipf, path,
 335                    ipf->queue.count, ipf->inprogress, ipf->autodefer,
 336                    (long)ipf->offset, ipf->fd,
 337                    ipf->rd ? "" : ",!rd",
 338                    ipf->skippinglong ? "*skiplong" : "",
 339                    ipf->rd && ipf->paused ? "*paused" : "");
 340 }