defer.c

   1 /*---------- defer and backlog files ----------*/
   2
   3 static void open_defer(void) {
   4   struct stat stab;
   5
   6   if (defer) return;
   7
   8   defer= fopen(path_defer, "a+");
   9   if (!defer) sysdie("could not open defer file %s", path_defer);
  10
  11   /* truncate away any half-written records */
  12
  13   xfstat_isreg(fileno(defer), &stab, path_defer, "newly opened defer file");
  14
  15   if (stab.st_size > LONG_MAX)
  16     crash("defer file %s size is far too large", path_defer);
  17
  18   if (!stab.st_size)
  19     return;
  20
  21   long orgsize= stab.st_size;
  22   long truncto= stab.st_size;
  23   for (;;) {
  24     if (!truncto) break; /* was only (if anything) one half-truncated record */
  25     if (fseek(defer, truncto-1, SEEK_SET) < 0)
  26       syscrash("seek in defer file %s while truncating partial", path_defer);
  27
  28     int r= getc(defer);
  29     if (r==EOF) {
  30       if (ferror(defer))
  31         syscrash("failed read from defer file %s", path_defer);
  32       else
  33         crash("defer file %s shrank while we were checking it!", path_defer);
  34     }
  35     if (r=='\n') break;
  36     truncto--;
  37   }
  38
  39   if (stab.st_size != truncto) {
  40     warn("truncating half-record at end of defer file %s -"
  41          " shrinking by %ld bytes from %ld to %ld",
  42          path_defer, orgsize - truncto, orgsize, truncto);
  43
  44     if (fflush(defer))
  45       sysdie("could not flush defer file %s", path_defer);
  46     if (ftruncate(fileno(defer), truncto))
  47       syscrash("could not truncate defer file %s", path_defer);
  48
  49   } else {
  50     info("continuing existing defer file %s (%ld bytes)",
  51          path_defer, orgsize);
  52   }
  53   if (fseek(defer, truncto, SEEK_SET))
  54     syscrash("could not seek to new end of defer file %s", path_defer);
  55 }
  56
  57 static void close_defer(void) {
  58   if (!defer)
  59     return;
  60
  61   struct stat stab;
  62   xfstat_isreg(fileno(defer), &stab, path_defer, "defer file");
  63
  64   if (fclose(defer)) sysdie("could not close defer file %s", path_defer);
  65   defer= 0;
  66
  67   time_t now= xtime();
  68
  69   char *backlog= xasprintf("%s_backlog_%lu.%lu", feedfile,
  70                            (unsigned long)now,
  71                            (unsigned long)stab.st_ino);
  72   if (link(path_defer, backlog))
  73     sysdie("could not install defer file %s as backlog file %s",
  74            path_defer, backlog);
  75   if (unlink(path_defer))
  76     syscrash("could not unlink old defer link %s to backlog file %s",
  77              path_defer, backlog);
  78
  79   free(backlog);
  80
  81   if (until_backlog_nextscan < 0 ||
  82       until_backlog_nextscan > backlog_retry_minperiods + 1)
  83     until_backlog_nextscan= backlog_retry_minperiods + 1;
  84 }
  85
  86 void poll_backlog_file(void) {
  87   if (until_backlog_nextscan < 0) return;
  88   if (until_backlog_nextscan-- > 0) return;
  89   search_backlog_file();
  90 }
  91
  92 static void search_backlog_file(void) {
  93   /* returns non-0 iff there are any backlog files */
  94
  95   glob_t gl;
  96   int r;
  97   unsigned ui;
  98   struct stat stab;
  99   const char *oldest_path=0;
 100   time_t oldest_mtime=0, now;
 101
 102   if (backlog_input_file) return;
 103
 104  try_again:
 105
 106   r= glob(globpat_backlog, GLOB_ERR|GLOB_MARK|GLOB_NOSORT, 0, &gl);
 107
 108   switch (r) {
 109   case GLOB_ABORTED:
 110     sysdie("failed to expand backlog pattern %s", globpat_backlog);
 111   case GLOB_NOSPACE:
 112     die("out of memory expanding backlog pattern %s", globpat_backlog);
 113   case 0:
 114     for (ui=0; ui<gl.gl_pathc; ui++) {
 115       const char *path= gl.gl_pathv[ui];
 116
 117       if (strchr(path,'#') || strchr(path,'~')) {
 118         dbg("backlog file search skipping %s", path);
 119         continue;
 120       }
 121       r= stat(path, &stab);
 122       if (r) {
 123         syswarn("failed to stat backlog file %s", path);
 124         continue;
 125       }
 126       if (!S_ISREG(stab.st_mode)) {
 127         warn("backlog file %s is not a plain file (or link to one)", path);
 128         continue;
 129       }
 130       if (!oldest_path || stab.st_mtime < oldest_mtime) {
 131         oldest_path= path;
 132         oldest_mtime= stab.st_mtime;
 133       }
 134     }
 135   case GLOB_NOMATCH: /* fall through */
 136     break;
 137   default:
 138     syscrash("glob expansion of backlog pattern %s gave unexpected"
 139              " nonzero (error?) return value %d", globpat_backlog, r);
 140   }
 141
 142   if (!oldest_path) {
 143     dbg("backlog scan: none");
 144
 145     if (sms==sm_DROPPED) {
 146       preterminate();
 147       notice("feed dropped and our work is complete");
 148
 149       r= unlink(path_cli);
 150       if (r && errno!=ENOENT)
 151         syswarn("failed to unlink cli socket for old feed");
 152
 153       xunlink(path_lock, "lockfile for old feed");
 154       exit(4);
 155     }
 156     until_backlog_nextscan= backlog_spontrescan_periods;
 157     goto xfree;
 158   }
 159
 160   now= xtime();
 161   double age= difftime(now, oldest_mtime);
 162   long age_deficiency= (backlog_retry_minperiods * period_seconds) - age;
 163
 164   if (age_deficiency <= 0) {
 165     dbg("backlog scan: found age=%f deficiency=%ld oldest=%s",
 166           age, age_deficiency, oldest_path);
 167
 168     backlog_input_file= open_input_file(oldest_path);
 169     if (!backlog_input_file) {
 170       warn("backlog file %s vanished as we opened it", oldest_path);
 171       globfree(&gl);
 172       goto try_again;
 173     }
 174     inputfile_reading_start(backlog_input_file);
 175     until_backlog_nextscan= -1;
 176     goto xfree;
 177   }
 178
 179   until_backlog_nextscan= age_deficiency / period_seconds;
 180
 181   if (backlog_spontrescan_periods >= 0 &&
 182       until_backlog_nextscan > backlog_spontrescan_periods)
 183     until_backlog_nextscan= backlog_spontrescan_periods;
 184
 185   dbg("backlog scan: young age=%f deficiency=%ld nextscan=%d oldest=%s",
 186         age, age_deficiency, until_backlog_nextscan, oldest_path);
 187
 188  xfree:
 189   globfree(&gl);
 190   return;
 191 }
 192