xmit.c

   1 /*
   2  *  innduct
   3  *  tailing reliable realtime streaming feeder for inn
   4  *  xmit.c - transmitting checks and articles, flow control, expiry
   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 const char *const artstate_names[]=
  30   { "Unchecked", "Wanted", "Unsolicited", 0 };
  31
  32 /*---------- assigning articles to conns, and transmitting ----------*/
  33
  34 static Article *dequeue_from(int peek, InputFile *ipf) {
  35   if (!ipf) return 0;
  36   if (peek) return LIST_HEAD(ipf->queue);
  37
  38   Article *art= LIST_REMHEAD(ipf->queue);
  39   if (!art) return 0;
  40   check_reading_pause_resume(ipf);
  41   return art;
  42 }
  43
  44 static Article *dequeue(int peek) {
  45   Article *art;
  46   art= dequeue_from(peek, flushing_input_file);  if (art) return art;
  47   art= dequeue_from(peek, backlog_input_file);   if (art) return art;
  48   art= dequeue_from(peek, main_input_file);      if (art) return art;
  49   return 0;
  50 }
  51
  52 void check_assign_articles(void) {
  53   for (;;) {
  54     if (!dequeue(1))
  55       break;
  56
  57     Conn *walk, *use=0;
  58     int spare=0, inqueue=0;
  59
  60     /* Find a connection to offer this article.  We prefer a busy
  61      * connection to an idle one, provided it's not full.  We take the
  62      * first (oldest) and since that's stable, it will mean we fill up
  63      * connections in order.  That way if we have too many
  64      * connections, the spare ones will go away eventually.
  65      */
  66     FOR_CONN(walk) {
  67       if (walk->quitting) continue;
  68       inqueue= walk->sent.count + walk->priority.count
  69              + walk->waiting.count;
  70       spare= walk->max_queue - inqueue;
  71       assert(inqueue <= max_queue_per_conn);
  72       assert(spare >= 0);
  73       if (inqueue==0) /*idle*/ { if (!use) use= walk; }
  74       else if (spare>0) /*working*/ { use= walk; break; }
  75     }
  76     if (use) {
  77       if (!inqueue) use->since_activity= 0; /* reset idle counter */
  78       while (spare>0) {
  79         Article *art= dequeue(0);
  80         if (!art) break;
  81         LIST_ADDTAIL(use->waiting, art);
  82         lowvol_perperiod[lowvol_circptr]++;
  83         spare--;
  84       }
  85       conn_maybe_write(use);
  86     } else if (allow_connect_start()) {
  87       connect_start();
  88       break;
  89     } else {
  90       break;
  91     }
  92   }
  93 }
  94
  95 static void *conn_writeable(oop_source *l, int fd, oop_event ev, void *u) {
  96   conn_maybe_write(u);
  97   return OOP_CONTINUE;
  98 }
  99
 100 void conn_maybe_write(Conn *conn) {
 101   for (;;) {
 102     conn_make_some_xmits(conn);
 103     if (!conn->xmitu) {
 104       loop->cancel_fd(loop, conn->fd, OOP_WRITE);
 105       conn->oopwriting= 0;
 106       return;
 107     }
 108
 109     void *rp= conn_write_some_xmits(conn);
 110     if (rp==OOP_CONTINUE) {
 111       if (!conn->oopwriting) {
 112         loop->on_fd(loop, conn->fd, OOP_WRITE, conn_writeable, conn);
 113         conn->oopwriting= 1;
 114       }
 115       return;
 116     } else if (rp==OOP_HALT) {
 117       return;
 118     } else if (!rp) {
 119       /* transmitted everything */
 120     } else {
 121       abort();
 122     }
 123   }
 124 }
 125
 126 /*---------- expiry, flow control and deferral ----------*/
 127
 128 /*
 129  * flow control notes
 130  * to ensure articles go away eventually
 131  * separate queue for each input file
 132  *   queue expiry
 133  *     every period, check head of backlog queue for expiry with SMretrieve
 134  *       if too old: discard, and check next article
 135  *     also check every backlog article as we read it
 136  *   flush expiry
 137  *     after too long in SEPARATED/DROPPING ie Separated/Finishing/Dropping
 138  *     one-off: eat queued articles from flushing and write them to defer
 139  *     one-off: connfail all connections which have any articles from flushing
 140  *     newly read articles from flushing go straight to defer
 141  *     this should take care of it and get us out of this state
 142  * to avoid filling up ram needlessly
 143  *   input control
 144  *     limit number of queued articles for each ipf
 145  *     pause/resume inputfile tailing
 146  */
 147
 148 void check_reading_pause_resume(InputFile *ipf) {
 149   if (ipf->queue.count >= max_queue_per_ipf)
 150     inputfile_reading_pause(ipf);
 151   else
 152     inputfile_reading_resume(ipf);
 153 }
 154
 155 void article_defer(Article *art /* not on a queue */, int whichcount) {
 156   open_defer();
 157   if (fprintf(defer, "%s %s\n", TokenToText(art->token), art->messageid) <0
 158       || fflush(defer))
 159     sysdie("write to defer file %s",path_defer);
 160   article_done(art, whichcount);
 161 }
 162
 163 int article_check_expired(Article *art /* must be queued, not conn */) {
 164   ARTHANDLE *artdata= SMretrieve(art->token, RETR_STAT);
 165   if (artdata) { SMfreearticle(artdata); return 0; }
 166
 167   LIST_REMOVE(art->ipf->queue, art);
 168   art->missing= 1;
 169   art->ipf->count_nooffer_missing++;
 170   article_done(art,-1);
 171   return 1;
 172 }
 173
 174 void inputfile_queue_check_expired(InputFile *ipf) {
 175   if (!ipf) return;
 176
 177   for (;;) {
 178     Article *art= LIST_HEAD(ipf->queue);
 179     int expd= article_check_expired(art);
 180     if (!expd) break;
 181   }
 182   check_reading_pause_resume(ipf);
 183 }
 184
 185 void article_autodefer(InputFile *ipf, Article *art) {
 186   ipf->autodefer++;
 187   article_defer(art,-1);
 188 }
 189
 190 static int has_article_in(const ArticleList *al, InputFile *ipf) {
 191   Article *art;
 192   for (art=LIST_HEAD(*al); art; art=LIST_NEXT(art))
 193     if (art->ipf == ipf) return 1;
 194   return 0;
 195 }
 196
 197 static void autodefer_input_file_articles(InputFile *ipf) {
 198   Article *art;
 199   while ((art= LIST_REMHEAD(ipf->queue)))
 200     article_autodefer(ipf, art);
 201 }
 202
 203 void autodefer_input_file(InputFile *ipf) {
 204   static const char *const abandon= "stuck";
 205   ipf->autodefer= 0;
 206
 207   autodefer_input_file_articles(ipf);
 208
 209   if (ipf->inprogress) {
 210     Conn *walk;
 211     FOR_CONN(walk) {
 212       if (has_article_in(&walk->waiting,  ipf) ||
 213           has_article_in(&walk->priority, ipf) ||
 214           has_article_in(&walk->sent,     ipf))
 215         walk->quitting= abandon;
 216     }
 217     while (ipf->inprogress) {
 218       FOR_CONN(walk)
 219         if (walk->quitting == abandon) goto found;
 220       abort(); /* where are they ?? */
 221
 222     found:
 223       connfail(walk, "connection is stuck or crawling,"
 224                " and we need to finish flush");
 225       autodefer_input_file_articles(ipf);
 226     }
 227   }
 228
 229   check_reading_pause_resume(ipf);
 230 }
 231
 232 /*========== article transmission ==========*/
 233
 234 static XmitDetails *xmit_core(Conn *conn, const char *data, int len,
 235                   XmitKind kind) { /* caller must then fill in details */
 236   struct iovec *v= &conn->xmit[conn->xmitu];
 237   XmitDetails *d= &conn->xmitd[conn->xmitu++];
 238   v->iov_base= (char*)data;
 239   v->iov_len= len;
 240   d->kind= kind;
 241   return d;
 242 }
 243
 244 static void xmit_noalloc(Conn *conn, const char *data, int len) {
 245   xmit_core(conn,data,len, xk_Const);
 246 }
 247 #define XMIT_LITERAL(lit) (xmit_noalloc(conn, (lit), sizeof(lit)-1))
 248
 249 static void xmit_artbody(Conn *conn, ARTHANDLE *ah /* consumed */) {
 250   XmitDetails *d= xmit_core(conn, ah->data, ah->len, xk_Artdata);
 251   d->info.sm_art= ah;
 252 }
 253
 254 void xmit_free(XmitDetails *d) {
 255   switch (d->kind) {
 256   case xk_Artdata: SMfreearticle(d->info.sm_art); break;
 257   case xk_Const:                                  break;
 258   default: abort();
 259   }
 260 }
 261
 262 void *conn_write_some_xmits(Conn *conn) {
 263   /* return values:
 264    *      0:            nothing more to write, no need to call us again
 265    *      OOP_CONTINUE: more to write but fd not writeable
 266    *      OOP_HALT:     disaster, have destroyed conn
 267    */
 268   for (;;) {
 269     int count= conn->xmitu;
 270     if (!count) return 0;
 271
 272     if (count > IOV_MAX) count= IOV_MAX;
 273     ssize_t rs= writev(conn->fd, conn->xmit, count);
 274     if (rs < 0) {
 275       if (isewouldblock(errno)) return OOP_CONTINUE;
 276       connfail(conn, "write failed: %s", strerror(errno));
 277       return OOP_HALT;
 278     }
 279     assert(rs > 0);
 280
 281     int done;
 282     for (done=0; rs; ) {
 283       assert(done<conn->xmitu);
 284       struct iovec *vp= &conn->xmit[done];
 285       XmitDetails *dp= &conn->xmitd[done];
 286       assert(vp->iov_len <= SSIZE_MAX);
 287       if ((size_t)rs >= vp->iov_len) {
 288         rs -= vp->iov_len;
 289         xmit_free(dp); /* vp->iov_len -= vp->iov_len, etc. */
 290         done++;
 291       } else {
 292         vp->iov_base= (char*)vp->iov_base + rs;
 293         vp->iov_len -= rs;
 294         break; /* rs -= rs */
 295       }
 296     }
 297     int newu= conn->xmitu - done;
 298     memmove(conn->xmit,  conn->xmit  + done, newu * sizeof(*conn->xmit));
 299     memmove(conn->xmitd, conn->xmitd + done, newu * sizeof(*conn->xmitd));
 300     conn->xmitu= newu;
 301   }
 302 }
 303
 304 void conn_make_some_xmits(Conn *conn) {
 305   for (;;) {
 306     if (conn->xmitu+5 > CONNIOVS)
 307       break;
 308
 309     Article *art= LIST_REMHEAD(conn->priority);
 310     if (!art) art= LIST_REMHEAD(conn->waiting);
 311     if (!art) break;
 312
 313     if (art->state >= art_Wanted || (conn->stream && nocheck)) {
 314       /* actually send it */
 315
 316       ARTHANDLE *artdata= SMretrieve(art->token, RETR_ALL);
 317
 318       art->state=
 319         art->state == art_Unchecked ? art_Unsolicited :
 320         art->state == art_Wanted    ? art_Wanted      :
 321         (abort(),-1);
 322
 323       if (!artdata) art->missing= 1;
 324       art->ipf->counts[art->state][ artdata ? RC_sent : RC_missing ]++;
 325
 326       if (conn->stream) {
 327         if (artdata) {
 328           XMIT_LITERAL("TAKETHIS ");
 329           xmit_noalloc(conn, art->messageid, art->midlen);
 330           XMIT_LITERAL("\r\n");
 331           xmit_artbody(conn, artdata);
 332         } else {
 333           article_done(art, -1);
 334           continue;
 335         }
 336       } else {
 337         /* we got 235 from IHAVE */
 338         if (artdata) {
 339           xmit_artbody(conn, artdata);
 340         } else {
 341           XMIT_LITERAL(".\r\n");
 342         }
 343       }
 344
 345       LIST_ADDTAIL(conn->sent, art);
 346
 347     } else {
 348       /* check it */
 349
 350       if (conn->stream)
 351         XMIT_LITERAL("CHECK ");
 352       else
 353         XMIT_LITERAL("IHAVE ");
 354       xmit_noalloc(conn, art->messageid, art->midlen);
 355       XMIT_LITERAL("\r\n");
 356
 357       assert(art->state == art_Unchecked);
 358       art->ipf->counts[art->state][RC_sent]++;
 359       LIST_ADDTAIL(conn->sent, art);
 360     }
 361   }
 362 }
 363