innfeed/tape.c

   1 /*  $Id: tape.c 6716 2004-05-16 20:26:56Z rra $
   2 **
   3 **  The implementation of the innfeed Tape class.
   4 **
   5 **  Written by James Brister <brister@vix.com>
   6 **
   7 **  The implementation of the Tape class. Tapes are read-only or write-only
   8 **  files that are accessed sequentially. Their basic unit of i/o is an
   9 **  Article. Tapes work out of a single directory and manage all file names
  10 **  themselves.
  11 **
  12 **  Tapes will checkpoint themselves periodically so that when innfeed exits
  13 **  or crashes things can restart close to where they were last. The period
  14 **  checkpointing is handled entirely by the Tape class, but the checkpoint
  15 **  period needs to be set by some external user before the first tape is
  16 **  created.
  17 */
  18
  19 #include "innfeed.h"
  20 #include "config.h"
  21 #include "clibrary.h"
  22
  23 #include <assert.h>
  24 #include <ctype.h>
  25 #include <errno.h>
  26 #include <sys/param.h>
  27 #include <sys/stat.h>
  28 #include <syslog.h>
  29
  30 #if defined (HAVE_DIRENT_H)
  31 #include <dirent.h>
  32 typedef struct dirent DIRENTRY ;
  33 #else
  34 #include <sys/dir.h>
  35 typedef struct direct DIRENTRY ;
  36 #endif
  37
  38 #ifdef TIME_WITH_SYS_TIME
  39 # include <sys/time.h>
  40 # include <time.h>
  41 #else
  42 # ifdef HAVE_SYS_TIME_H
  43 #  include <sys/time.h>
  44 # else
  45 #  include <time.h>
  46 # endif
  47 #endif
  48
  49 #include "inn/innconf.h"
  50 #include "inn/messages.h"
  51 #include "libinn.h"
  52
  53 #include "article.h"
  54 #include "configfile.h"
  55 #include "endpoint.h"
  56 #include "tape.h"
  57
  58 extern char *dflTapeDir;
  59 extern bool genHtml ;
  60
  61 #if 0
  62 /* a structure for temporary storage of articles. */
  63 typedef struct q_e_s
  64 {
  65     Article article ;
  66     struct q_e_s *next ;
  67 } *QueueElem ;
  68 #endif
  69
  70 /* The Tape class type. */
  71 struct tape_s
  72 {
  73     /* the pathname of the file the administrator can drop in by hand. */
  74     char *handFilename ;
  75
  76     /* the pathname of the file the Tape will read from */
  77     char *inputFilename ;
  78
  79     /* the pathname of the file the Tape will write to. */
  80     char *outputFilename ;
  81
  82     /* the pathname of the file used in locking */
  83     char *lockFilename ;
  84
  85     /* the peer we're doing this for. */
  86     char *peerName ;
  87
  88     FILE *inFp ;                /* input FILE */
  89     FILE *outFp ;               /* output FILE */
  90
  91     time_t lastRotated ;        /* time files last got switched */
  92     bool checkNew ;             /* set bool when we need to check for
  93                                    hand-crafted file. */
  94
  95 #if 0
  96     /* the tape holds a small output queue in memory to avoid thrashing. */
  97     QueueElem head ;
  98     QueueElem tail ;
  99     unsigned int qLength ;             /* amount on the queue */
 100 #endif
 101
 102     long outputSize ;           /* the current size of the output file. */
 103     long lossage ;
 104
 105     /* the number of bytes we try to keep the output under. We actually
 106        wait for the outputSize to get 10% greater than this amount before
 107        shrinking the file down again. A value of zero means no limit. */
 108     long outputLowLimit ;
 109     long outputHighLimit ;
 110     double backlogFactor ;
 111
 112     bool scribbled ;            /* have we scribbled a checkpoint value in
 113                                    the file. */
 114     long tellpos ;              /* for input file checkpointing. */
 115     bool changed ;              /* true if tape was read since last
 116                                    checkpoint or start. */
 117
 118     /* true if articles that are output are NOT later input. */
 119     bool noRotate ;
 120
 121     /* true if no articles should ever be spooled */
 122     bool noBacklog ;
 123 };
 124
 125
 126 static void checkpointTape (Tape tape) ;
 127 static void removeTapeGlobally (Tape tape) ;
 128 static void addTapeGlobally (Tape tape) ;
 129 static void prepareFiles (Tape tape) ;
 130 static void tapeCkNewFileCbk (TimeoutId id, void *d) ;
 131 static void tapeCheckpointCallback (TimeoutId id, void *d) ;
 132 #if 0
 133 static void flushTape (Tape tape) ;
 134 #endif
 135 static void tapesSetCheckNew (void) ;
 136 static void initTape (Tape nt) ;
 137 static void tapeCleanup (void) ;
 138
 139
 140
 141 /* pathname of directory we store tape files in. */
 142 static char *tapeDirectory ;
 143
 144 /* the callback ID of the checkpoint timer callback. */
 145 static TimeoutId checkPtId ;
 146 static TimeoutId ckNewFileId ;
 147
 148 /* number of seconds between tape checkpoints. */
 149 static unsigned int tapeCkPtPeriod ;
 150 static unsigned int tapeCkNewFilePeriod ;
 151
 152 static time_t rotatePeriod = TAPE_ROTATE_PERIOD ;
 153
 154 /* global list of tapes so we can checkpoint them periodically */
 155 static Tape *activeTapes ;
 156
 157 /* Size of the activeTapes array */
 158 static size_t activeTapeSize ;
 159
 160 /* index of last element in activeTapes that's being used. */
 161 static size_t activeTapeIdx ;
 162
 163 #if 0
 164 /* default limit of the size of output tapes. */
 165 static long defaultSizeLimit ;
 166 #endif
 167
 168 unsigned int tapeHighwater ;
 169
 170 bool debugShrinking = false ;
 171
 172 extern bool talkToSelf ;        /* main.c */
 173
 174
 175
 176
 177 /* callback when config file is loaded */
 178 int tapeConfigLoadCbk (void *data)
 179 {
 180   int rval = 1 ;
 181   long iv ;
 182   int bv ;
 183   FILE *fp = (FILE *) data ;
 184   char *dir, *p ;
 185
 186   if (getString (topScope,"backlog-directory",&p,NO_INHERIT))
 187     {
 188       dir = buildFilename(innconf->pathspool, p);
 189       free(p);
 190       if (tapeDirectory != NULL && strcmp (tapeDirectory,dir) != 0)
 191         {
 192           warn ("ME config: cannot change backlog-directory of a running"
 193                 " process") ;
 194           free (dir) ;
 195           dir = xstrdup (tapeDirectory) ;
 196         }
 197
 198       if (!isDirectory (dir) && isDirectory (dflTapeDir))
 199         {
 200           logOrPrint (LOG_ERR,fp,
 201                       "ME config: definition of backlog-directory (%s) is a"
 202                       " non-existant directory. Using %s",
 203                       dir,dflTapeDir) ;
 204           free (dir) ;
 205           dir = xstrdup (dflTapeDir) ;
 206         }
 207       else if (!isDirectory (dir))
 208         logAndExit (1,"ME config: no usable value for backlog-directory") ;
 209     }
 210   else if (!isDirectory (dflTapeDir))
 211   {
 212     logAndExit (1,"ME config: no usable value for backlog-directory") ;
 213     return -1; /* not reached */
 214   }
 215   else
 216     dir = xstrdup (dflTapeDir) ;
 217
 218   if (tapeDirectory != NULL)
 219     free (tapeDirectory) ;
 220   tapeDirectory = dir ;
 221
 222
 223
 224   if (getInteger (topScope,"backlog-highwater",&iv,NO_INHERIT))
 225     {
 226       if (iv < 0)
 227         {
 228           rval = 0 ;
 229           logOrPrint (LOG_ERR,fp,
 230                       "ME config: value of %s (%ld) in %s cannot be less"
 231                       " than 0. Using %ld","backlog-highwater",
 232                       iv,"global scope",(long)TAPE_HIGHWATER);
 233           iv = TAPE_HIGHWATER ;
 234         }
 235     }
 236   else
 237     iv = TAPE_HIGHWATER ;
 238   tapeHighwater = (unsigned int) iv ;
 239
 240
 241
 242   if (getInteger (topScope,"backlog-rotate-period",&iv,NO_INHERIT))
 243     {
 244       if (iv < 0)
 245         {
 246           rval = 0 ;
 247           logOrPrint (LOG_ERR,fp,
 248                       "ME config: value of %s (%ld) in %s cannot be less"
 249                       " than 0. Using %ld",
 250                       "backlog-rotate-period",
 251                       iv,"global scope",(long)TAPE_ROTATE_PERIOD);
 252           iv = TAPE_ROTATE_PERIOD ;
 253         }
 254     }
 255   else
 256     iv = TAPE_ROTATE_PERIOD ;
 257   rotatePeriod = (unsigned int) iv ;
 258
 259
 260   if (getInteger (topScope,"backlog-ckpt-period",&iv,NO_INHERIT))
 261     {
 262       if (iv < 0)
 263         {
 264           rval = 0 ;
 265           logOrPrint (LOG_ERR,fp,
 266                       "ME config: value of %s (%ld) in %s cannot be less"
 267                       " than 0. Using %ld",
 268                       "backlog-ckpt-period",iv,
 269                       "global scope",(long)TAPE_CHECKPOINT_PERIOD);
 270           iv = TAPE_CHECKPOINT_PERIOD ;
 271         }
 272     }
 273   else
 274     iv = TAPE_CHECKPOINT_PERIOD ;
 275   tapeCkPtPeriod = (unsigned int) iv ;
 276
 277
 278   if (getInteger (topScope,"backlog-newfile-period",&iv,NO_INHERIT))
 279     {
 280       if (iv < 0)
 281         {
 282           rval = 0 ;
 283           logOrPrint (LOG_ERR,fp,
 284                       "ME config: value of %s (%ld) in %s cannot be less"
 285                       " than 0. Using %ld",
 286                       "backlog-newfile-period",
 287                       iv,"global scope",(long)TAPE_NEWFILE_PERIOD);
 288           iv = TAPE_NEWFILE_PERIOD ;
 289         }
 290     }
 291   else
 292     iv = TAPE_NEWFILE_PERIOD ;
 293   tapeCkNewFilePeriod = (unsigned int) iv ;
 294
 295
 296   if (getBool (topScope,"debug-shrinking",&bv,NO_INHERIT))
 297     debugShrinking = (bv ? true : false) ;
 298
 299   return rval ;
 300 }
 301
 302
 303 /* Create a new Tape object. There are three potential files involved in
 304    I/O. 'peerName' is what  the admin may have dropped in by
 305    hand. 'peerName.input' is the file that was being used as input the last
 306    time things were run. 'peerName.output' is the file that was being used
 307    as output. The file 'peerName' is appended to 'peerName.input' (or
 308    renamed if 'peerName.input' doesn't exist). Then 'peerName.output' is
 309    appeneded (or renamed) to 'peerName.input' */
 310
 311 static bool inited = false ;
 312 Tape newTape (const char *peerName, bool dontRotate)
 313 {
 314   Tape nt = xmalloc (sizeof(struct tape_s)) ;
 315   size_t pLen = strlen (peerName) ;
 316   size_t dLen = strlen (tapeDirectory) ;
 317
 318   if (!inited)
 319     {
 320       inited = true ;
 321       atexit (tapeCleanup) ;
 322     }
 323
 324   ASSERT (nt != NULL) ;
 325
 326   if (endsIn (peerName,INPUT_TAIL))
 327     die ("Sorry, can't have a peer name ending in \"%s\"",INPUT_TAIL) ;
 328
 329   if (endsIn (peerName,OUTPUT_TAIL))
 330     die ("Sorry, can't have a peer name ending in \"%s\"",OUTPUT_TAIL) ;
 331
 332   if (endsIn (peerName,LOCK_TAIL))
 333     die ("Sorry, can't have a peer name ending in \"%s\"",LOCK_TAIL) ;
 334
 335   nt->peerName = xstrdup (peerName) ;
 336
 337   nt->handFilename = xmalloc (pLen + dLen + 2) ;
 338   sprintf (nt->handFilename,"%s/%s",tapeDirectory,peerName) ;
 339
 340   nt->lockFilename = xmalloc (pLen + dLen + strlen(LOCK_TAIL) + 2) ;
 341   sprintf (nt->lockFilename,"%s/%s%s",tapeDirectory,peerName,LOCK_TAIL) ;
 342
 343   nt->inputFilename = xmalloc (pLen + dLen + strlen(INPUT_TAIL) + 2) ;
 344   sprintf (nt->inputFilename,"%s/%s%s",tapeDirectory,peerName,INPUT_TAIL) ;
 345
 346   nt->outputFilename = xmalloc (pLen + dLen + strlen(OUTPUT_TAIL) + 2) ;
 347   sprintf (nt->outputFilename,"%s/%s%s",tapeDirectory,peerName,OUTPUT_TAIL) ;
 348
 349   if ( !lockFile (nt->lockFilename) )
 350     {
 351       warn ("ME lock failed for host: %s", nt->lockFilename) ;
 352
 353       free (nt->handFilename) ;
 354       free (nt->lockFilename) ;
 355       free (nt->inputFilename) ;
 356       free (nt->outputFilename) ;
 357       free (nt) ;
 358
 359       return NULL ;
 360     }
 361
 362   nt->noRotate = false ;        /* for first time prepare */
 363   initTape (nt) ;
 364   nt->noRotate = dontRotate ;
 365
 366   addTapeGlobally (nt) ;
 367
 368   if (checkPtId == 0 && tapeCkPtPeriod > 0)     /* only done once. */
 369     checkPtId = prepareSleep (tapeCheckpointCallback,tapeCkPtPeriod,NULL);
 370
 371   if (ckNewFileId == 0 && tapeCkNewFilePeriod > 0)
 372     ckNewFileId = prepareSleep (tapeCkNewFileCbk,tapeCkNewFilePeriod,NULL) ;
 373
 374   return nt ;
 375 }
 376
 377 static void initTape (Tape nt)
 378 {
 379   value *peerVal = findPeer (nt->peerName) ;
 380   scope *s = (peerVal == NULL ? NULL : peerVal->v.scope_val) ;
 381
 382   nt->inFp = NULL ;
 383   nt->outFp = NULL ;
 384
 385   nt->lastRotated = 0 ;
 386   nt->checkNew = false ;
 387
 388 #if 0
 389   nt->head = NULL ;
 390   nt->tail = NULL ;
 391   nt->qLength = 0 ;
 392 #endif
 393
 394   nt->scribbled = false ;
 395   nt->tellpos = 0 ;
 396
 397   nt->changed = false ;
 398
 399   nt->outputSize = 0 ;
 400   nt->lossage = 0 ;
 401
 402   nt->noBacklog = false ;
 403   nt->backlogFactor = 0.0 ;
 404   nt->outputLowLimit = 0 ;
 405   nt->outputHighLimit = 0 ;
 406
 407   if (!talkToSelf)
 408     {
 409       int bval ;
 410
 411       if (getBool (s, "no-backlog", &bval, INHERIT))
 412         nt->noBacklog = (bval ? true : false);
 413       else
 414         nt->noBacklog = TAPE_DISABLE;
 415
 416       if (getInteger (s,"backlog-limit",&nt->outputLowLimit,INHERIT))
 417         {
 418           if (!getReal (s,"backlog-factor",&nt->backlogFactor,INHERIT))
 419             {
 420               if (!getInteger (s,"backlog-limit-highwater",
 421                                &nt->outputHighLimit,INHERIT))
 422                 {
 423                   warn ("%s no backlog-factor or backlog-high-limit",
 424                         nt->peerName) ;
 425                   nt->outputLowLimit = 0 ;
 426                   nt->outputHighLimit = 0 ;
 427                   nt->backlogFactor = 0.0 ;
 428                 }
 429             }
 430           else
 431             nt->outputHighLimit = (long)(nt->outputLowLimit * nt->backlogFactor);
 432         }
 433       else
 434         warn ("ME config: no definition for required key backlog-limit") ;
 435     }
 436
 437   d_printf (1, "%s spooling: %s\n", nt->peerName,
 438            nt->noBacklog ? "disabled" : "enabled");
 439
 440   d_printf (1,"%s tape backlog limit: [%ld %ld]\n",nt->peerName,
 441            nt->outputLowLimit,
 442            nt->outputHighLimit) ;
 443
 444   prepareFiles (nt) ;
 445 }
 446
 447
 448 void gFlushTapes (void)
 449 {
 450   unsigned int i ;
 451
 452   notice ("ME flushing tapes") ;
 453   for (i = 0 ; i < activeTapeIdx ; i++)
 454     tapeFlush (activeTapes [i]) ;
 455 }
 456
 457
 458
 459 /* close the input and output tapes and reinitialize everything in the
 460   tape. */
 461 void tapeFlush (Tape tape)
 462 {
 463   if (tape->inFp != NULL)
 464     {
 465       checkpointTape (tape) ;
 466       fclose (tape->inFp) ;
 467     }
 468
 469   if (tape->outFp != NULL)
 470     fclose (tape->outFp) ;
 471
 472   initTape (tape) ;
 473 }
 474
 475
 476
 477 void gPrintTapeInfo (FILE *fp, unsigned int indentAmt)
 478 {
 479   char indent [INDENT_BUFFER_SIZE] ;
 480   unsigned int i ;
 481
 482   for (i = 0 ; i < MIN(INDENT_BUFFER_SIZE - 1,indentAmt) ; i++)
 483     indent [i] = ' ' ;
 484   indent [i] = '\0' ;
 485
 486   fprintf (fp,"%sGlobal Tape List : (count %lu) {\n",
 487            indent,(unsigned long) activeTapeIdx) ;
 488
 489   for (i = 0 ; i < activeTapeIdx ; i++)
 490     printTapeInfo (activeTapes [i],fp,indentAmt + INDENT_INCR) ;
 491   fprintf (fp,"%s}\n",indent) ;
 492 }
 493
 494
 495 void tapeLogGlobalStatus (FILE *fp)
 496 {
 497   fprintf (fp,"%sBacklog file global values:%s\n",
 498            genHtml ? "<B>" : "", genHtml ? "</B>" : "") ;
 499   fprintf (fp,"        directory: %s\n",tapeDirectory) ;
 500   fprintf (fp,"    rotate period: %-3ld seconds\n",(long) rotatePeriod) ;
 501   fprintf (fp,"checkpoint period: %-3ld seconds\n",(long) tapeCkPtPeriod) ;
 502   fprintf (fp,"   newfile period: %-3ld seconds\n",(long) tapeCkNewFilePeriod);
 503   fprintf (fp,"\n") ;
 504 }
 505
 506
 507 void tapeLogStatus (Tape tape, FILE *fp)
 508 {
 509   if (tape == NULL)
 510     fprintf (fp,"(no tape)\n") ;
 511   else if (tape->noBacklog)
 512       fprintf (fp,"                 spooling: DISABLED\n");
 513   else if (tape->outputLowLimit == 0)
 514       fprintf (fp,"                 spooling: UNLIMITED\n");
 515   else
 516     {
 517       fprintf (fp,"                 backlog low limit: %ld\n",
 518                tape->outputLowLimit) ;
 519       fprintf (fp,"               backlog upper limit: %ld",
 520                tape->outputHighLimit) ;
 521       if (tape->backlogFactor > 0.0)
 522         fprintf (fp," (factor %1.2f)",tape->backlogFactor) ;
 523       fputc ('\n',fp) ;
 524       fprintf (fp,"                 backlog shrinkage: ") ;
 525       fprintf (fp,"%ld bytes (from current file)\n",tape->lossage) ;
 526     }
 527 }
 528
 529 void printTapeInfo (Tape tape, FILE *fp, unsigned int indentAmt)
 530 {
 531   char indent [INDENT_BUFFER_SIZE] ;
 532   unsigned int i ;
 533 #if 0
 534   QueueElem qe ;
 535 #endif
 536
 537   for (i = 0 ; i < MIN(INDENT_BUFFER_SIZE - 1,indentAmt) ; i++)
 538     indent [i] = ' ' ;
 539   indent [i] = '\0' ;
 540
 541   fprintf (fp,"%sTape : %p {\n",indent,(void *) tape) ;
 542
 543   if (tape == NULL)
 544     {
 545       fprintf (fp,"%s}\n",indent) ;
 546       return ;
 547     }
 548
 549   fprintf (fp,"%s    master-file : %s\n", indent, tape->handFilename) ;
 550   fprintf (fp,"%s    input-file : %s\n", indent, tape->inputFilename) ;
 551   fprintf (fp,"%s    output-file : %s\n",indent, tape->outputFilename) ;
 552   fprintf (fp,"%s    lock-file : %s\n",indent, tape->lockFilename) ;
 553   fprintf (fp,"%s    peerName : %s\n",indent,tape->peerName) ;
 554   fprintf (fp,"%s    input-FILE : %p\n",indent, (void *) tape->inFp) ;
 555   fprintf (fp,"%s    output-FILE : %p\n",indent, (void *) tape->outFp) ;
 556   fprintf (fp,"%s    output-limit : %ld\n",indent, tape->outputLowLimit) ;
 557
 558 #if 0
 559   fprintf (fp,"%s    in-memory article queue (length  %d) {\n",indent,
 560            tape->qLength) ;
 561
 562   for (qe = tape->head ; qe != NULL ; qe = qe->next)
 563     {
 564 #if 0
 565       printArticleInfo (qe->article,fp,indentAmt + INDENT_INCR) ;
 566 #else
 567       fprintf (fp,"%s    %p\n",indent,qe->article) ;
 568 #endif
 569     }
 570
 571   fprintf (fp,"%s    }\n",indent) ;
 572 #endif
 573
 574   fprintf (fp,"%s    tell-position : %ld\n",indent,(long) tape->tellpos) ;
 575   fprintf (fp,"%s    input-FILE-changed : %s\n",indent,
 576            boolToString (tape->changed)) ;
 577
 578   fprintf (fp,"%s    no-rotate : %s\n",indent, boolToString (tape->noRotate));
 579
 580   fprintf (fp,"%s}\n",indent) ;
 581 }
 582
 583
 584
 585
 586 /* delete the tape. Spools the in-memory articles to disk. */
 587 void delTape (Tape tape)
 588 {
 589   struct stat st ;
 590
 591   if (tape == NULL)
 592     return ;
 593
 594 #if 1
 595
 596   if (tape->outFp != NULL && fclose (tape->outFp) != 0)
 597     syswarn ("ME ioerr fclose %s", tape->outputFilename) ;
 598
 599   if (stat(tape->outputFilename, &st) == 0 && st.st_size == 0)
 600     {
 601       d_printf (1,"removing empty output tape: %s\n",tape->outputFilename) ;
 602       unlink (tape->outputFilename) ;
 603     }
 604
 605   tape->outFp = NULL ;
 606   tape->outputSize = 0 ;
 607
 608 #else
 609
 610   tapeClose (tape) ;
 611
 612 #endif
 613
 614   if (tape->inFp != NULL)
 615     {
 616       checkpointTape (tape) ;
 617       fclose (tape->inFp) ;
 618     }
 619
 620   unlockFile (tape->lockFilename) ;
 621
 622   freeCharP (tape->handFilename) ;
 623   freeCharP (tape->inputFilename) ;
 624   freeCharP (tape->outputFilename) ;
 625   freeCharP (tape->lockFilename) ;
 626   freeCharP (tape->peerName) ;
 627
 628   removeTapeGlobally (tape) ;
 629
 630   free (tape) ;
 631 }
 632
 633
 634 void tapeTakeArticle (Tape tape, Article article)
 635 {
 636 #if 0
 637   QueueElem elem ;
 638 #endif
 639   int amt ;
 640   const char *fname, *msgid ;
 641
 642   ASSERT (tape != NULL) ;
 643
 644   /* return immediately if spooling disabled - jgarzik */
 645   if (tape->noBacklog)
 646     {
 647       delArticle (article) ;
 648       return;
 649     }
 650
 651   fname = artFileName (article) ;
 652   msgid = artMsgId (article) ;
 653   amt = fprintf (tape->outFp,"%s %s\n", fname, msgid) ;
 654
 655   /* I'd rather know where I am each time, and I don't trust all
 656     fprintf's to give me character counts. */
 657 #if defined (TRUST_FPRINTF)
 658
 659   tape->outputSize += amt ;
 660
 661 #else
 662 #if defined (NO_TRUST_STRLEN)
 663
 664   tape->outputSize = ftello (tape->outFp) ;
 665
 666 #else
 667
 668   tape->outputSize += strlen(fname) + strlen(msgid) + 2 ; /* " " + "\n" */
 669
 670 #endif
 671 #endif
 672
 673   delArticle (article) ;
 674
 675   if (debugShrinking)
 676     {
 677       struct stat sb ;
 678
 679       fflush (tape->outFp) ;
 680
 681       if (fstat (fileno (tape->outFp),&sb) != 0)
 682         syswarn ("ME oserr fstat %s",tape->outputFilename) ;
 683       else if (sb.st_size != tape->outputSize)
 684         syslog (LOG_ERR,"fstat and ftello do not agree: %ld %ld for %s\n",
 685                 (long)sb.st_size,tape->outputSize,tape->outputFilename) ;
 686     }
 687
 688   if (tape->outputHighLimit > 0 && tape->outputSize >= tape->outputHighLimit)
 689     {
 690       long oldSize = tape->outputSize ;
 691       shrinkfile (tape->outFp,tape->outputLowLimit,tape->outputFilename,"a+");
 692       tape->outputSize = ftello (tape->outFp) ;
 693       tape->lossage += oldSize - tape->outputSize ;
 694     }
 695 }
 696
 697
 698 /* Pick an article off a tape and return it. NULL is returned if there
 699    are no more articles. */
 700 Article getArticle (Tape tape)
 701 {
 702   char line [2048] ;            /* ick. 1024 for filename + 1024 for msgid */
 703   char *p, *q ;
 704   char *msgid, *filename ;
 705   Article art = NULL ;
 706   time_t now = theTime() ;
 707
 708   ASSERT (tape != NULL) ;
 709
 710   if (tape->inFp == NULL && (now - tape->lastRotated) > rotatePeriod)
 711     prepareFiles (tape) ;       /* will flush queue too. */
 712
 713   while (tape->inFp != NULL && art == NULL)
 714     {
 715       tape->changed = true ;
 716
 717       if (fgets (line,sizeof (line), tape->inFp) == NULL)
 718         {
 719           if (ferror (tape->inFp))
 720             syswarn ("ME ioerr on tape file %s", tape->inputFilename) ;
 721           else if ( !feof (tape->inFp) )
 722             syswarn ("ME oserr fgets %s", tape->inputFilename) ;
 723
 724           if (fclose (tape->inFp) != 0)
 725             syswarn ("ME ioerr fclose %s", tape->inputFilename) ;
 726
 727           d_printf (1,"No more articles on tape %s\n",tape->inputFilename) ;
 728
 729           tape->inFp = NULL ;
 730           tape->scribbled = false ;
 731
 732           unlink (tape->inputFilename) ;
 733
 734           if ((now - tape->lastRotated) > rotatePeriod)
 735             prepareFiles (tape) ; /* rotate files to try next. */
 736         }
 737       else
 738         {
 739           msgid = filename = NULL ;
 740
 741           for (p = line ; *p && CTYPE(isspace, *p) ; p++) /* eat whitespace */
 742             /* nada */ ;
 743
 744           if (*p != '\0')
 745             {
 746               q = strchr (p,' ') ;
 747
 748               if (q != NULL)
 749                 {
 750                   filename = p ;
 751                   *q = '\0' ;
 752
 753                   for (q++ ; *q && CTYPE(isspace, *q) ; q++)
 754                     /* nada */ ;
 755
 756                   if (*q != '\0')
 757                     {
 758                       if (((p = strchr (q, ' ')) != NULL) ||
 759                           ((p = strchr (q, '\n')) != NULL))
 760                         *p = '\0' ;
 761
 762                       if (p != NULL)
 763                         msgid = q ;
 764                       else
 765                         filename = NULL ; /* no trailing newline or blank */
 766                     }
 767                   else
 768                     filename = NULL ; /* line had one field and some blanks */
 769                 }
 770               else
 771                 filename = NULL ; /* line only had one field */
 772             }
 773
 774           /* See if message ID looks valid. */
 775           if (msgid) {
 776             for (p = msgid; *p; p++)
 777               ;
 778             if (p > msgid) p--;
 779             if (*msgid != '<' || *p != '>') {
 780               warn ("ME tape invalid messageID in %s: %s",
 781                     tape->inputFilename,msgid);
 782               msgid = NULL;
 783             }
 784           }
 785
 786           if (filename != NULL && msgid != NULL)
 787             art = newArticle (filename, msgid) ;
 788
 789           /* art may be NULL here if the file is no longer valid. */
 790         }
 791     }
 792
 793 #if 0
 794   /* now we either have an article or there is no more on disk */
 795   if (art == NULL)
 796     {
 797       if (tape->inFp != NULL && ((c = fgetc (tape->inFp)) != EOF))
 798         ungetc (c,tape->inFp) ; /* shouldn't happen */
 799       else if (tape->inFp != NULL)
 800         {
 801           /* last article read was the end of the tape. */
 802           if (fclose (tape->inFp) != 0)
 803             syswarn ("ME ioerr fclose %s", tape->inputFilename) ;
 804
 805           tape->inFp = NULL ;
 806           tape->scribbled = false ;
 807
 808           /* toss out the old input file and prepare the new one */
 809           unlink (tape->inputFilename) ;
 810
 811           if (now - tape->lastRotated > rotatePeriod)
 812             prepareFiles (tape) ;
 813         }
 814     }
 815 #endif
 816
 817   if (art == NULL)
 818     d_printf (2,"%s All out of articles in the backlog\n", tape->peerName) ;
 819   else
 820     d_printf (2,"%s Peeled article %s from backlog\n",tape->peerName,
 821              artMsgId (art)) ;
 822
 823   return art ;
 824 }
 825
 826
 827 /****************************************************/
 828 /**                  CLASS FUNCTIONS               **/
 829 /****************************************************/
 830
 831 /* Cause all the Tapes to checkpoint themselves. */
 832 void checkPointTapes (void)
 833 {
 834   unsigned int i ;
 835
 836   for (i = 0 ; i < activeTapeIdx ; i++)
 837     checkpointTape (activeTapes [i]) ;
 838 }
 839
 840
 841 /* make all the tapes set their checkNew flag. */
 842 static void tapesSetCheckNew (void)
 843 {
 844   unsigned int i ;
 845
 846   for (i = 0 ; i < activeTapeIdx ; i++)
 847     activeTapes[i]->checkNew = true ;
 848 }
 849
 850
 851 #if 0
 852 /* Set the pathname of the directory for storing tapes in. */
 853 void setTapeDirectory (const char *newDir)
 854 {
 855   /* the activeTape variable gets set when the first Tape object
 856      is created */
 857   if (activeTapes != NULL)
 858     {
 859       syslog (LOG_CRIT,"Resetting backlog directory") ;
 860       abort() ;
 861     }
 862
 863   if (tapeDirectory != NULL)
 864     freeCharP (tapeDirectory) ;
 865
 866   tapeDirectory = xstrdup (newDir) ;
 867
 868   addPointerFreedOnExit (tapeDirectory) ;
 869 }
 870 #endif
 871
 872 /* Get the pathname of the directory tapes are stored in. */
 873 const char *getTapeDirectory (void)
 874 {
 875   ASSERT (tapeDirectory != NULL) ;
 876
 877 #if 0
 878   if (tapeDirectory == NULL)
 879     {
 880       tapeDirectory = xstrdup (dflTapeDir) ;
 881       addPointerFreedOnExit (tapeDirectory) ;
 882     }
 883 #endif
 884
 885   return tapeDirectory ;
 886 }
 887
 888
 889 #if 0
 890 void setOutputSizeLimit (long val)
 891 {
 892   defaultSizeLimit = val ;
 893 }
 894 #endif
 895
 896
 897
 898
 899
 900
 901 /**********************************************************************/
 902 /*                          PRIVATE FUNCTIONS                         */
 903 /**********************************************************************/
 904
 905
 906 /* Add a new tape to the class-level list of active tapes.  */
 907 static void addTapeGlobally (Tape tape)
 908 {
 909   ASSERT (tape != NULL) ;
 910
 911   if (activeTapeSize == activeTapeIdx)
 912     {
 913       unsigned int i ;
 914
 915       activeTapeSize += 10 ;
 916       if (activeTapes != NULL)
 917         activeTapes = xrealloc (activeTapes, sizeof(Tape) * activeTapeSize) ;
 918       else
 919         activeTapes = xmalloc (sizeof(Tape) * activeTapeSize) ;
 920
 921       for (i = activeTapeIdx ; i < activeTapeSize ; i++)
 922         activeTapes [i] = NULL ;
 923     }
 924   activeTapes [activeTapeIdx++] = tape ;
 925 }
 926
 927
 928 /* Remove a tape for the class-level list of active tapes. */
 929 static void removeTapeGlobally (Tape tape)
 930 {
 931   unsigned int i ;
 932
 933   if (tape == NULL)
 934     return ;
 935
 936   ASSERT (activeTapeIdx > 0) ;
 937
 938   for (i = 0 ; i < activeTapeIdx ; i++)
 939     if (activeTapes [i] == tape)
 940       break ;
 941
 942   ASSERT (i < activeTapeIdx) ;
 943
 944   for ( ; i < (activeTapeIdx - 1) ; i++)
 945     activeTapes [i] = activeTapes [i + 1] ;
 946
 947   activeTapes [--activeTapeIdx] = NULL ;
 948
 949   if (activeTapeIdx == 0)
 950     {
 951       free (activeTapes) ;
 952       activeTapes = NULL ;
 953     }
 954 }
 955
 956
 957 /* Have a tape checkpoint itself so that next process can pick up where
 958    this one left off. */
 959 static void checkpointTape (Tape tape)
 960 {
 961   if (tape->inFp == NULL)       /* no input file being read. */
 962     return ;
 963
 964   if (!tape->changed)   /* haven't read since last checkpoint */
 965     {
 966       d_printf (1,"Not checkpointing unchanged tape: %s\n", tape->peerName) ;
 967       return ;
 968     }
 969
 970   if ((tape->tellpos = ftello (tape->inFp)) < 0)
 971     {
 972       syswarn ("ME oserr ftello %s", tape->inputFilename) ;
 973       return ;
 974     }
 975
 976   /* strlen of "18446744073709551616\n" (2^64) */
 977 #define BITS64 21
 978
 979   /* make sure we're not right at the beginning of the file so we can write. */
 980   if (tape->tellpos > BITS64)
 981     {
 982       rewind (tape->inFp) ;
 983
 984       /* scribble blanks over the first lines characters */
 985       if (!tape->scribbled)
 986         {
 987           int currloc = 0 ;
 988           int c ;
 989
 990           while ((c = fgetc (tape->inFp)) != '\n' || currloc <= BITS64)
 991             if (c == EOF)
 992               return ;
 993             else
 994               currloc++ ;
 995
 996           rewind (tape->inFp) ;
 997
 998           while (currloc-- > 0)
 999             fputc (' ',tape->inFp) ;
1000
1001           rewind (tape->inFp) ;
1002
1003           fflush (tape->inFp) ;
1004           tape->scribbled = true ;
1005         }
1006
1007       fprintf (tape->inFp,"%ld",tape->tellpos) ;
1008
1009       if (fseeko (tape->inFp,tape->tellpos,SEEK_SET) != 0)
1010         syswarn ("ME oserr fseeko(%s,%ld,SEEK_SET)",tape->inputFilename,
1011                  tape->tellpos) ;
1012     }
1013
1014   tape->changed = false ;
1015 }
1016
1017
1018
1019 /* Prepare the tape file(s) for input and output */
1020
1021 /* For a given Tape there are
1022  * three possible files: PEER.input PEER and
1023  * PEER.output. PEER.input and PEER.output are private to
1024  * innfeed. PEER is where a sysadmin can drop a file that (s)he
1025  * wants to inject into the process. If the first line of the input file
1026  * contains only an integer (possibly surrounded by spaces), then this is
1027  * taken to be the position to seek to before reading.
1028  *
1029  * prepareFiles will process them in a manner much like the following shell
1030  * commands:
1031  *
1032  * if [ ! -f PEER.input ]; then
1033  *      if [ -f PEER ]; then mv PEER PEER.input
1034  *      elif [ -f PEER.output ]; then mv PEER.output PEER; fi
1035  * fi
1036  *
1037  * At this point PEER.input is opened for reading if it exists.
1038  *
1039  * The checkpoint file is left as-is unless the PEER.input file
1040  * happens to be newer that the checkpoint file.
1041  */
1042 static void prepareFiles (Tape tape)
1043 {
1044   bool inpExists ;
1045   bool outExists ;
1046   bool newExists ;
1047
1048 #if 0
1049   /* flush any in memory articles to disk */
1050   if (tape->head != NULL && tape->outFp != NULL)
1051     flushTape(tape) ;
1052 #endif
1053
1054   tape->tellpos = 0 ;
1055
1056   /*  First time through, or something external has set checkNew */
1057   if (tape->lastRotated == 0 || tape->checkNew)
1058     {
1059       newExists = fileExistsP (tape->handFilename) ;
1060       if (newExists)
1061         notice ("%s new hand-prepared backlog file", tape->peerName) ;
1062     }
1063   else
1064     newExists = false ;
1065
1066   if (tape->lastRotated == 0)    /* first time here */
1067     {
1068       inpExists = fileExistsP (tape->inputFilename) ;
1069       outExists = fileExistsP (tape->outputFilename) ;
1070     }
1071   else
1072     {
1073       inpExists = (tape->inFp != NULL) ? true : false ; /* can this ever be true?? */
1074       outExists = (tape->outFp != NULL && tape->outputSize > 0) ? true : false ;
1075     }
1076
1077
1078   /* move the hand-dropped file to the input file if needed. */
1079   if (newExists && !inpExists)
1080     {
1081       if (rename (tape->handFilename,tape->inputFilename) != 0)
1082         syswarn ("ME oserr rename %s, %s", tape->handFilename,
1083                  tape->inputFilename);
1084       else
1085         {
1086           notice ("%s grabbing external tape file", tape->peerName) ;
1087           inpExists = true ;
1088         }
1089     }
1090
1091   /* now move the output file to the input file, if needed and only if in
1092      not in NOROTATE mode. */
1093   if (outExists && !inpExists && !tape->noRotate)
1094     {
1095       if (tape->outFp != NULL)
1096         {
1097           fclose (tape->outFp) ;
1098           tape->outFp = NULL ;
1099         }
1100
1101       if (rename (tape->outputFilename,tape->inputFilename) != 0)
1102         syswarn ("ME oserr rename %s, %s", tape->outputFilename,
1103                  tape->inputFilename) ;
1104       else
1105         inpExists = true ;
1106
1107       outExists = false ;
1108     }
1109
1110   /* now open up the input file and seek to the proper position. */
1111   if (inpExists)
1112     {
1113       int c ;
1114       long flength ;
1115
1116       if ((tape->inFp = fopen (tape->inputFilename,"r+")) == NULL)
1117         syswarn ("ME fopen %s", tape->inputFilename) ;
1118       else
1119         {
1120           char buffer [64] ;
1121
1122           if (fgets (buffer,sizeof (buffer) - 1, tape->inFp) == NULL)
1123             {
1124               if (feof (tape->inFp))
1125                 {
1126                   d_printf (1,"Empty input file: %s\n",tape->inputFilename) ;
1127                   unlink (tape->inputFilename) ;
1128                 }
1129               else
1130                 syswarn ("ME oserr fgets %s", tape->inputFilename) ;
1131
1132               fclose (tape->inFp) ;
1133               tape->inFp = NULL ;
1134               tape->scribbled = false ;
1135             }
1136           else
1137             {
1138               unsigned int len = strlen (buffer) ;
1139               long newPos = 0 ;
1140
1141               if (len > 0 && buffer [len - 1] == '\n')
1142                 buffer [--len] = '\0' ;
1143
1144               if (len > 0 && strspn (buffer,"0123456789 \n") == len)
1145                 {
1146                   if (sscanf (buffer,"%ld",&newPos) == 1)
1147                     {
1148                       tape->scribbled = true ;
1149                       tape->tellpos = newPos ;
1150                     }
1151                 }
1152
1153               if ((flength = fileLength (fileno (tape->inFp))) < tape->tellpos)
1154                 {
1155                   warn ("ME tape short: %s %ld %ld", tape->inputFilename,
1156                         flength, tape->tellpos) ;
1157                   tape->tellpos = 0 ;
1158                 }
1159               else if (tape->tellpos == 0)
1160                 rewind (tape->inFp) ;
1161               else if (fseeko (tape->inFp,tape->tellpos - 1,SEEK_SET) != 0)
1162                 syswarn ("ME oserr fseeko(%s,%ld,SEEK_SET)",
1163                          tape->inputFilename,tape->tellpos) ;
1164               else if ((c = fgetc (tape->inFp)) != '\n')
1165                 {
1166                   while (c != EOF && c != '\n')
1167                     c = fgetc (tape->inFp) ;
1168
1169                   if (c == EOF)
1170                     {
1171                       fclose (tape->inFp) ;
1172                       unlink (tape->inputFilename) ;
1173                       tape->inFp = NULL ;
1174                       tape->scribbled = false ;
1175                       prepareFiles (tape) ;
1176                     }
1177                   else
1178                     {
1179                       long oldPos = tape->tellpos ;
1180
1181                       tape->changed = true ;
1182                       checkpointTape (tape) ;
1183
1184                       warn ("ME internal checkpoint line boundary missed:"
1185                             " %s %ld vs. %ld",tape->inputFilename,
1186                               tape->tellpos, oldPos) ;
1187                     }
1188                 }
1189             }
1190         }
1191     }
1192
1193   tape->lastRotated = theTime() ;
1194   tape->checkNew = false ;
1195
1196   /* now open up the output file. */
1197   if (tape->outFp == NULL)
1198     {
1199       if ((tape->outFp = fopen (tape->outputFilename,"a+")) == NULL)
1200         {
1201           syswarn ("ME tape open failed (a+) %s", tape->outputFilename) ;
1202           return ;
1203         }
1204       fseeko (tape->outFp,0,SEEK_END) ;
1205       tape->outputSize = ftello (tape->outFp) ;
1206       tape->lossage = 0 ;
1207     }
1208 }
1209
1210
1211 static void tapeCkNewFileCbk (TimeoutId id, void *d UNUSED)
1212 {
1213   ASSERT (id == ckNewFileId) ;
1214   ASSERT (tapeCkNewFilePeriod > 0) ;
1215
1216   tapesSetCheckNew () ;
1217
1218   ckNewFileId = prepareSleep (tapeCkNewFileCbk,tapeCkNewFilePeriod,NULL) ;
1219 }
1220
1221
1222 /* The timer callback function that will checkpoint all the active tapes. */
1223 static void tapeCheckpointCallback (TimeoutId id, void *d UNUSED)
1224 {
1225   ASSERT (id == checkPtId) ;
1226   ASSERT (tapeCkPtPeriod > 0) ;
1227
1228   d_printf (1,"Checkpointing tapes\n") ;
1229
1230   checkPointTapes () ;
1231
1232   checkPtId = prepareSleep (tapeCheckpointCallback,tapeCkPtPeriod,NULL) ;
1233 }
1234
1235
1236
1237 #if 0
1238 static void flushTape (Tape tape)
1239 {
1240   QueueElem elem ;
1241
1242   /* flush out queue to disk. */
1243   elem = tape->head ;
1244   while (elem != NULL)
1245     {
1246       tape->head = tape->head->next ;
1247       fprintf (tape->outFp,"%s %s\n", artFileName (elem->article),
1248                artMsgId (elem->article)) ;
1249
1250       delArticle (elem->article) ;
1251
1252       free (elem) ;
1253       elem = tape->head ;
1254     }
1255   tape->tail = NULL ;
1256   tape->qLength = 0;
1257
1258
1259   /* I'd rather know where I am each time, and I don't trust all
1260      fprintf's to give me character counts. */
1261   tape->outputSize = ftello (tape->outFp) ;
1262   if (debugShrinking)
1263     {
1264       struct stat buf ;
1265       static bool logged = false ;
1266
1267       fflush (tape->outFp) ;
1268       if (fstat (fileno (tape->outFp),&buf) != 0 && !logged)
1269         {
1270           syslog (LOG_ERR,FSTAT_FAILURE,tape->outputFilename) ;
1271           logged = true ;
1272         }
1273       else if (buf.st_size != tape->outputSize)
1274         {
1275           warn ("ME fstat and ftello do not agree for %s",
1276                 tape->outputFilename) ;
1277           logged = true ;
1278         }
1279     }
1280
1281   if (tape->outputHighLimit > 0 && tape->outputSize > tape->outputHighLimit)
1282     {
1283       shrinkfile (tape->outFp,tape->outputLowLimit,tape->outputFilename,"a+");
1284       tape->outputSize = ftello (tape->outFp) ;
1285     }
1286 }
1287 #endif
1288
1289
1290 static void tapeCleanup (void)
1291 {
1292   free (tapeDirectory) ;
1293   tapeDirectory = NULL ;
1294 }