src/helper.cc

   1 /*
   2  * $Id$
   3  *
   4  * DEBUG: section 84    Helper process maintenance
   5  * AUTHOR: Harvest Derived?
   6  *
   7  * SQUID Web Proxy Cache          http://www.squid-cache.org/
   8  * ----------------------------------------------------------
   9  *
  10  *  Squid is the result of efforts by numerous individuals from
  11  *  the Internet community; see the CONTRIBUTORS file for full
  12  *  details.   Many organizations have provided support for Squid's
  13  *  development; see the SPONSORS file for full details.  Squid is
  14  *  Copyrighted (C) 2001 by the Regents of the University of
  15  *  California; see the COPYRIGHT file for full details.  Squid
  16  *  incorporates software developed and/or copyrighted by other
  17  *  sources; see the CREDITS file for full details.
  18  *
  19  *  This program is free software; you can redistribute it and/or modify
  20  *  it under the terms of the GNU General Public License as published by
  21  *  the Free Software Foundation; either version 2 of the License, or
  22  *  (at your option) any later version.
  23  *
  24  *  This program is distributed in the hope that it will be useful,
  25  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  26  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  27  *  GNU General Public License for more details.
  28  *
  29  *  You should have received a copy of the GNU General Public License
  30  *  along with this program; if not, write to the Free Software
  31  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
  32  *
  33  */
  34
  35 #include "squid-old.h"
  36 #include "base/AsyncCbdataCalls.h"
  37 #include "comm.h"
  38 #include "comm/Connection.h"
  39 #include "comm/Write.h"
  40 #include "helper.h"
  41 #include "format/Quoting.h"
  42 #include "MemBuf.h"
  43 #include "SquidMath.h"
  44 #include "SquidTime.h"
  45 #include "Store.h"
  46 #include "wordlist.h"
  47
  48 #define HELPER_MAX_ARGS 64
  49
  50 /* size of helper read buffer (maximum?). no reason given for this size */
  51 /* though it has been seen to be too short for some requests */
  52 /* it is dynamic, so increasng should not have side effects */
  53 #define BUF_8KB 8192
  54
  55 static IOCB helperHandleRead;
  56 static IOCB helperStatefulHandleRead;
  57 static void helperServerFree(helper_server *srv);
  58 static void helperStatefulServerFree(helper_stateful_server *srv);
  59 static void Enqueue(helper * hlp, helper_request *);
  60 static helper_request *Dequeue(helper * hlp);
  61 static helper_stateful_request *StatefulDequeue(statefulhelper * hlp);
  62 static helper_server *GetFirstAvailable(helper * hlp);
  63 static helper_stateful_server *StatefulGetFirstAvailable(statefulhelper * hlp);
  64 static void helperDispatch(helper_server * srv, helper_request * r);
  65 static void helperStatefulDispatch(helper_stateful_server * srv, helper_stateful_request * r);
  66 static void helperKickQueue(helper * hlp);
  67 static void helperStatefulKickQueue(statefulhelper * hlp);
  68 static void helperStatefulServerDone(helper_stateful_server * srv);
  69 static void helperRequestFree(helper_request * r);
  70 static void helperStatefulRequestFree(helper_stateful_request * r);
  71 static void StatefulEnqueue(statefulhelper * hlp, helper_stateful_request * r);
  72 static bool helperStartStats(StoreEntry *sentry, void *hlp, const char *label);
  73
  74
  75 CBDATA_CLASS_INIT(helper);
  76 CBDATA_TYPE(helper_server);
  77 CBDATA_CLASS_INIT(statefulhelper);
  78 CBDATA_TYPE(helper_stateful_server);
  79
  80 void
  81 HelperServerBase::closePipesSafely()
  82 {
  83 #if _SQUID_MSWIN_
  84     int no = index + 1;
  85
  86     shutdown(writePipe->fd, SD_BOTH);
  87 #endif
  88
  89     flags.closing = 1;
  90     if (readPipe->fd == writePipe->fd)
  91         readPipe->fd = -1;
  92     else
  93         readPipe->close();
  94     writePipe->close();
  95
  96 #if _SQUID_MSWIN_
  97     if (hIpc) {
  98         if (WaitForSingleObject(hIpc, 5000) != WAIT_OBJECT_0) {
  99             getCurrentTime();
 100             debugs(84, DBG_IMPORTANT, "WARNING: " << hlp->id_name <<
 101                    " #" << no << " (" << hlp->cmdline->key << "," <<
 102                    (long int)pid << ") didn't exit in 5 seconds");
 103         }
 104         CloseHandle(hIpc);
 105     }
 106 #endif
 107 }
 108
 109 void
 110 HelperServerBase::closeWritePipeSafely()
 111 {
 112 #if _SQUID_MSWIN_
 113     int no = index + 1;
 114
 115     shutdown(writePipe->fd, (readPipe->fd == writePipe->fd ? SD_BOTH : SD_SEND));
 116 #endif
 117
 118     flags.closing = 1;
 119     if (readPipe->fd == writePipe->fd)
 120         readPipe->fd = -1;
 121     writePipe->close();
 122
 123 #if _SQUID_MSWIN_
 124     if (hIpc) {
 125         if (WaitForSingleObject(hIpc, 5000) != WAIT_OBJECT_0) {
 126             getCurrentTime();
 127             debugs(84, DBG_IMPORTANT, "WARNING: " << hlp->id_name <<
 128                    " #" << no << " (" << hlp->cmdline->key << "," <<
 129                    (long int)pid << ") didn't exit in 5 seconds");
 130         }
 131         CloseHandle(hIpc);
 132     }
 133 #endif
 134 }
 135
 136 void
 137 helperOpenServers(helper * hlp)
 138 {
 139     char *s;
 140     char *progname;
 141     char *shortname;
 142     char *procname;
 143     const char *args[HELPER_MAX_ARGS+1]; // save space for a NULL terminator
 144     char fd_note_buf[FD_DESC_SZ];
 145     helper_server *srv;
 146     int nargs = 0;
 147     int k;
 148     pid_t pid;
 149     int rfd;
 150     int wfd;
 151     void * hIpc;
 152     wordlist *w;
 153
 154     if (hlp->cmdline == NULL)
 155         return;
 156
 157     progname = hlp->cmdline->key;
 158
 159     if ((s = strrchr(progname, '/')))
 160         shortname = xstrdup(s + 1);
 161     else
 162         shortname = xstrdup(progname);
 163
 164     /* figure out how many new child are actually needed. */
 165     int need_new = hlp->childs.needNew();
 166
 167     debugs(84, 1, "helperOpenServers: Starting " << need_new << "/" << hlp->childs.n_max << " '" << shortname << "' processes");
 168
 169     if (need_new < 1) {
 170         debugs(84, 1, "helperOpenServers: No '" << shortname << "' processes needed.");
 171     }
 172
 173     procname = (char *)xmalloc(strlen(shortname) + 3);
 174
 175     snprintf(procname, strlen(shortname) + 3, "(%s)", shortname);
 176
 177     args[nargs++] = procname;
 178
 179     for (w = hlp->cmdline->next; w && nargs < HELPER_MAX_ARGS; w = w->next)
 180         args[nargs++] = w->key;
 181
 182     args[nargs++] = NULL;
 183
 184     assert(nargs <= HELPER_MAX_ARGS);
 185
 186     for (k = 0; k < need_new; k++) {
 187         getCurrentTime();
 188         rfd = wfd = -1;
 189         pid = ipcCreate(hlp->ipc_type,
 190                         progname,
 191                         args,
 192                         shortname,
 193                         hlp->addr,
 194                         &rfd,
 195                         &wfd,
 196                         &hIpc);
 197
 198         if (pid < 0) {
 199             debugs(84, 1, "WARNING: Cannot run '" << progname << "' process.");
 200             continue;
 201         }
 202
 203         hlp->childs.n_running++;
 204         hlp->childs.n_active++;
 205         CBDATA_INIT_TYPE(helper_server);
 206         srv = cbdataAlloc(helper_server);
 207         srv->hIpc = hIpc;
 208         srv->pid = pid;
 209         srv->index = k;
 210         srv->addr = hlp->addr;
 211         srv->readPipe = new Comm::Connection;
 212         srv->readPipe->fd = rfd;
 213         srv->writePipe = new Comm::Connection;
 214         srv->writePipe->fd = wfd;
 215         srv->rbuf = (char *)memAllocBuf(BUF_8KB, &srv->rbuf_sz);
 216         srv->wqueue = new MemBuf;
 217         srv->roffset = 0;
 218         srv->requests = (helper_request **)xcalloc(hlp->childs.concurrency ? hlp->childs.concurrency : 1, sizeof(*srv->requests));
 219         srv->parent = cbdataReference(hlp);
 220         dlinkAddTail(srv, &srv->link, &hlp->servers);
 221
 222         if (rfd == wfd) {
 223             snprintf(fd_note_buf, FD_DESC_SZ, "%s #%d", shortname, k + 1);
 224             fd_note(rfd, fd_note_buf);
 225         } else {
 226             snprintf(fd_note_buf, FD_DESC_SZ, "reading %s #%d", shortname, k + 1);
 227             fd_note(rfd, fd_note_buf);
 228             snprintf(fd_note_buf, FD_DESC_SZ, "writing %s #%d", shortname, k + 1);
 229             fd_note(wfd, fd_note_buf);
 230         }
 231
 232         commSetNonBlocking(rfd);
 233
 234         if (wfd != rfd)
 235             commSetNonBlocking(wfd);
 236
 237         AsyncCall::Pointer closeCall = asyncCall(5,4, "helperServerFree", cbdataDialer(helperServerFree, srv));
 238         comm_add_close_handler(rfd, closeCall);
 239
 240         AsyncCall::Pointer call = commCbCall(5,4, "helperHandleRead",
 241                                              CommIoCbPtrFun(helperHandleRead, srv));
 242         comm_read(srv->readPipe, srv->rbuf, srv->rbuf_sz - 1, call);
 243     }
 244
 245     hlp->last_restart = squid_curtime;
 246     safe_free(shortname);
 247     safe_free(procname);
 248     helperKickQueue(hlp);
 249 }
 250
 251 /**
 252  * DPW 2007-05-08
 253  *
 254  * helperStatefulOpenServers: create the stateful child helper processes
 255  */
 256 void
 257 helperStatefulOpenServers(statefulhelper * hlp)
 258 {
 259     char *shortname;
 260     const char *args[HELPER_MAX_ARGS+1]; // save space for a NULL terminator
 261     char fd_note_buf[FD_DESC_SZ];
 262     int nargs = 0;
 263
 264     if (hlp->cmdline == NULL)
 265         return;
 266
 267     if (hlp->childs.concurrency)
 268         debugs(84, 0, "ERROR: concurrency= is not yet supported for stateful helpers ('" << hlp->cmdline << "')");
 269
 270     char *progname = hlp->cmdline->key;
 271
 272     char *s;
 273     if ((s = strrchr(progname, '/')))
 274         shortname = xstrdup(s + 1);
 275     else
 276         shortname = xstrdup(progname);
 277
 278     /* figure out haw mant new helpers are needed. */
 279     int need_new = hlp->childs.needNew();
 280
 281     debugs(84, 1, "helperOpenServers: Starting " << need_new << "/" << hlp->childs.n_max << " '" << shortname << "' processes");
 282
 283     if (need_new < 1) {
 284         debugs(84, 1, "helperStatefulOpenServers: No '" << shortname << "' processes needed.");
 285     }
 286
 287     char *procname = (char *)xmalloc(strlen(shortname) + 3);
 288
 289     snprintf(procname, strlen(shortname) + 3, "(%s)", shortname);
 290
 291     args[nargs++] = procname;
 292
 293     for (wordlist *w = hlp->cmdline->next; w && nargs < HELPER_MAX_ARGS; w = w->next)
 294         args[nargs++] = w->key;
 295
 296     args[nargs++] = NULL;
 297
 298     assert(nargs <= HELPER_MAX_ARGS);
 299
 300     for (int k = 0; k < need_new; k++) {
 301         getCurrentTime();
 302         int rfd = -1;
 303         int wfd = -1;
 304         void * hIpc;
 305         pid_t pid = ipcCreate(hlp->ipc_type,
 306                               progname,
 307                               args,
 308                               shortname,
 309                               hlp->addr,
 310                               &rfd,
 311                               &wfd,
 312                               &hIpc);
 313
 314         if (pid < 0) {
 315             debugs(84, 1, "WARNING: Cannot run '" << progname << "' process.");
 316             continue;
 317         }
 318
 319         hlp->childs.n_running++;
 320         hlp->childs.n_active++;
 321         CBDATA_INIT_TYPE(helper_stateful_server);
 322         helper_stateful_server *srv = cbdataAlloc(helper_stateful_server);
 323         srv->hIpc = hIpc;
 324         srv->pid = pid;
 325         srv->flags.reserved = 0;
 326         srv->stats.submits = 0;
 327         srv->stats.releases = 0;
 328         srv->index = k;
 329         srv->addr = hlp->addr;
 330         srv->readPipe = new Comm::Connection;
 331         srv->readPipe->fd = rfd;
 332         srv->writePipe = new Comm::Connection;
 333         srv->writePipe->fd = wfd;
 334         srv->rbuf = (char *)memAllocBuf(BUF_8KB, &srv->rbuf_sz);
 335         srv->roffset = 0;
 336         srv->parent = cbdataReference(hlp);
 337
 338         if (hlp->datapool != NULL)
 339             srv->data = hlp->datapool->alloc();
 340
 341         dlinkAddTail(srv, &srv->link, &hlp->servers);
 342
 343         if (rfd == wfd) {
 344             snprintf(fd_note_buf, FD_DESC_SZ, "%s #%d", shortname, k + 1);
 345             fd_note(rfd, fd_note_buf);
 346         } else {
 347             snprintf(fd_note_buf, FD_DESC_SZ, "reading %s #%d", shortname, k + 1);
 348             fd_note(rfd, fd_note_buf);
 349             snprintf(fd_note_buf, FD_DESC_SZ, "writing %s #%d", shortname, k + 1);
 350             fd_note(wfd, fd_note_buf);
 351         }
 352
 353         commSetNonBlocking(rfd);
 354
 355         if (wfd != rfd)
 356             commSetNonBlocking(wfd);
 357
 358         AsyncCall::Pointer closeCall = asyncCall(5,4, "helperStatefulServerFree", cbdataDialer(helperStatefulServerFree, srv));
 359         comm_add_close_handler(rfd, closeCall);
 360
 361         AsyncCall::Pointer call = commCbCall(5,4, "helperStatefulHandleRead",
 362                                              CommIoCbPtrFun(helperStatefulHandleRead, srv));
 363         comm_read(srv->readPipe, srv->rbuf, srv->rbuf_sz - 1, call);
 364     }
 365
 366     hlp->last_restart = squid_curtime;
 367     safe_free(shortname);
 368     safe_free(procname);
 369     helperStatefulKickQueue(hlp);
 370 }
 371
 372
 373 void
 374 helperSubmit(helper * hlp, const char *buf, HLPCB * callback, void *data)
 375 {
 376     if (hlp == NULL) {
 377         debugs(84, 3, "helperSubmit: hlp == NULL");
 378         callback(data, NULL);
 379         return;
 380     }
 381
 382     helper_request *r = new helper_request;
 383     helper_server *srv;
 384
 385     r->callback = callback;
 386     r->data = cbdataReference(data);
 387     r->buf = xstrdup(buf);
 388
 389     if ((srv = GetFirstAvailable(hlp)))
 390         helperDispatch(srv, r);
 391     else
 392         Enqueue(hlp, r);
 393
 394     debugs(84, 9, "helperSubmit: " << buf);
 395 }
 396
 397 /// lastserver = "server last used as part of a reserved request sequence"
 398 void
 399 helperStatefulSubmit(statefulhelper * hlp, const char *buf, HLPSCB * callback, void *data, helper_stateful_server * lastserver)
 400 {
 401     if (hlp == NULL) {
 402         debugs(84, 3, "helperStatefulSubmit: hlp == NULL");
 403         callback(data, 0, NULL);
 404         return;
 405     }
 406
 407     helper_stateful_request *r = new helper_stateful_request;
 408
 409     r->callback = callback;
 410     r->data = cbdataReference(data);
 411
 412     if (buf != NULL) {
 413         r->buf = xstrdup(buf);
 414         r->placeholder = 0;
 415     } else {
 416         r->buf = NULL;
 417         r->placeholder = 1;
 418     }
 419
 420     if ((buf != NULL) && lastserver) {
 421         debugs(84, 5, "StatefulSubmit with lastserver " << lastserver);
 422         assert(lastserver->flags.reserved);
 423         assert(!(lastserver->request));
 424
 425         debugs(84, 5, "StatefulSubmit dispatching");
 426         helperStatefulDispatch(lastserver, r);
 427     } else {
 428         helper_stateful_server *srv;
 429         if ((srv = StatefulGetFirstAvailable(hlp))) {
 430             helperStatefulDispatch(srv, r);
 431         } else
 432             StatefulEnqueue(hlp, r);
 433     }
 434
 435     debugs(84, 9, "helperStatefulSubmit: placeholder: '" << r->placeholder << "', buf '" << buf << "'.");
 436 }
 437
 438 /**
 439  * DPW 2007-05-08
 440  *
 441  * helperStatefulReleaseServer tells the helper that whoever was
 442  * using it no longer needs its services.
 443  */
 444 void
 445 helperStatefulReleaseServer(helper_stateful_server * srv)
 446 {
 447     debugs(84, 3, HERE << "srv-" << srv->index << " flags.reserved = " << srv->flags.reserved);
 448     if (!srv->flags.reserved)
 449         return;
 450
 451     srv->stats.releases++;
 452
 453     srv->flags.reserved = 0;
 454     if (srv->parent->OnEmptyQueue != NULL && srv->data)
 455         srv->parent->OnEmptyQueue(srv->data);
 456
 457     helperStatefulServerDone(srv);
 458 }
 459
 460 /** return a pointer to the stateful routines data area */
 461 void *
 462 helperStatefulServerGetData(helper_stateful_server * srv)
 463 {
 464     return srv->data;
 465 }
 466
 467 /**
 468  * Dump some stats about the helper states to a StoreEntry
 469  */
 470 void
 471 helperStats(StoreEntry * sentry, helper * hlp, const char *label)
 472 {
 473     if (!helperStartStats(sentry, hlp, label))
 474         return;
 475
 476     storeAppendPrintf(sentry, "program: %s\n",
 477                       hlp->cmdline->key);
 478     storeAppendPrintf(sentry, "number active: %d of %d (%d shutting down)\n",
 479                       hlp->childs.n_active, hlp->childs.n_max, (hlp->childs.n_running - hlp->childs.n_active) );
 480     storeAppendPrintf(sentry, "requests sent: %d\n",
 481                       hlp->stats.requests);
 482     storeAppendPrintf(sentry, "replies received: %d\n",
 483                       hlp->stats.replies);
 484     storeAppendPrintf(sentry, "queue length: %d\n",
 485                       hlp->stats.queue_size);
 486     storeAppendPrintf(sentry, "avg service time: %d msec\n",
 487                       hlp->stats.avg_svc_time);
 488     storeAppendPrintf(sentry, "\n");
 489     storeAppendPrintf(sentry, "%7s\t%7s\t%7s\t%11s\t%s\t%7s\t%7s\t%7s\n",
 490                       "#",
 491                       "FD",
 492                       "PID",
 493                       "# Requests",
 494                       "Flags",
 495                       "Time",
 496                       "Offset",
 497                       "Request");
 498
 499     for (dlink_node *link = hlp->servers.head; link; link = link->next) {
 500         helper_server *srv = (helper_server*)link->data;
 501         double tt = 0.001 * (srv->requests[0] ? tvSubMsec(srv->requests[0]->dispatch_time, current_time) : tvSubMsec(srv->dispatch_time, srv->answer_time));
 502         storeAppendPrintf(sentry, "%7d\t%7d\t%7d\t%11d\t%c%c%c%c\t%7.3f\t%7d\t%s\n",
 503                           srv->index + 1,
 504                           srv->readPipe->fd,
 505                           srv->pid,
 506                           srv->stats.uses,
 507                           srv->stats.pending ? 'B' : ' ',
 508                           srv->flags.writing ? 'W' : ' ',
 509                           srv->flags.closing ? 'C' : ' ',
 510                           srv->flags.shutdown ? 'S' : ' ',
 511                           tt < 0.0 ? 0.0 : tt,
 512                           (int) srv->roffset,
 513                           srv->requests[0] ? Format::QuoteMimeBlob(srv->requests[0]->buf) : "(none)");
 514     }
 515
 516     storeAppendPrintf(sentry, "\nFlags key:\n\n");
 517     storeAppendPrintf(sentry, "   B = BUSY\n");
 518     storeAppendPrintf(sentry, "   W = WRITING\n");
 519     storeAppendPrintf(sentry, "   C = CLOSING\n");
 520     storeAppendPrintf(sentry, "   S = SHUTDOWN PENDING\n");
 521 }
 522
 523 void
 524 helperStatefulStats(StoreEntry * sentry, statefulhelper * hlp, const char *label)
 525 {
 526     if (!helperStartStats(sentry, hlp, label))
 527         return;
 528
 529     storeAppendPrintf(sentry, "program: %s\n",
 530                       hlp->cmdline->key);
 531     storeAppendPrintf(sentry, "number active: %d of %d (%d shutting down)\n",
 532                       hlp->childs.n_active, hlp->childs.n_max, (hlp->childs.n_running - hlp->childs.n_active) );
 533     storeAppendPrintf(sentry, "requests sent: %d\n",
 534                       hlp->stats.requests);
 535     storeAppendPrintf(sentry, "replies received: %d\n",
 536                       hlp->stats.replies);
 537     storeAppendPrintf(sentry, "queue length: %d\n",
 538                       hlp->stats.queue_size);
 539     storeAppendPrintf(sentry, "avg service time: %d msec\n",
 540                       hlp->stats.avg_svc_time);
 541     storeAppendPrintf(sentry, "\n");
 542     storeAppendPrintf(sentry, "%7s\t%7s\t%7s\t%11s\t%6s\t%7s\t%7s\t%7s\n",
 543                       "#",
 544                       "FD",
 545                       "PID",
 546                       "# Requests",
 547                       "Flags",
 548                       "Time",
 549                       "Offset",
 550                       "Request");
 551
 552     for (dlink_node *link = hlp->servers.head; link; link = link->next) {
 553         helper_stateful_server *srv = (helper_stateful_server *)link->data;
 554         double tt = 0.001 * tvSubMsec(srv->dispatch_time, srv->flags.busy ? current_time : srv->answer_time);
 555         storeAppendPrintf(sentry, "%7d\t%7d\t%7d\t%11d\t%c%c%c%c%c\t%7.3f\t%7d\t%s\n",
 556                           srv->index + 1,
 557                           srv->readPipe->fd,
 558                           srv->pid,
 559                           srv->stats.uses,
 560                           srv->flags.busy ? 'B' : ' ',
 561                           srv->flags.closing ? 'C' : ' ',
 562                           srv->flags.reserved ? 'R' : ' ',
 563                           srv->flags.shutdown ? 'S' : ' ',
 564                           srv->request ? (srv->request->placeholder ? 'P' : ' ') : ' ',
 565                                   tt < 0.0 ? 0.0 : tt,
 566                                   (int) srv->roffset,
 567                                   srv->request ? Format::QuoteMimeBlob(srv->request->buf) : "(none)");
 568     }
 569
 570     storeAppendPrintf(sentry, "\nFlags key:\n\n");
 571     storeAppendPrintf(sentry, "   B = BUSY\n");
 572     storeAppendPrintf(sentry, "   C = CLOSING\n");
 573     storeAppendPrintf(sentry, "   R = RESERVED\n");
 574     storeAppendPrintf(sentry, "   S = SHUTDOWN PENDING\n");
 575     storeAppendPrintf(sentry, "   P = PLACEHOLDER\n");
 576 }
 577
 578 void
 579 helperShutdown(helper * hlp)
 580 {
 581     dlink_node *link = hlp->servers.head;
 582
 583     while (link) {
 584         helper_server *srv;
 585         srv = (helper_server *)link->data;
 586         link = link->next;
 587
 588         if (srv->flags.shutdown) {
 589             debugs(84, 3, "helperShutdown: " << hlp->id_name << " #" << srv->index + 1 << " has already SHUT DOWN.");
 590             continue;
 591         }
 592
 593         assert(hlp->childs.n_active > 0);
 594         hlp->childs.n_active--;
 595         srv->flags.shutdown = 1;        /* request it to shut itself down */
 596
 597         if (srv->flags.closing) {
 598             debugs(84, 3, "helperShutdown: " << hlp->id_name << " #" << srv->index + 1 << " is CLOSING.");
 599             continue;
 600         }
 601
 602         if (srv->stats.pending) {
 603             debugs(84, 3, "helperShutdown: " << hlp->id_name << " #" << srv->index + 1 << " is BUSY.");
 604             continue;
 605         }
 606
 607         debugs(84, 3, "helperShutdown: " << hlp->id_name << " #" << srv->index + 1 << " shutting down.");
 608         /* the rest of the details is dealt with in the helperServerFree
 609          * close handler
 610          */
 611         srv->closePipesSafely();
 612     }
 613 }
 614
 615 void
 616 helperStatefulShutdown(statefulhelper * hlp)
 617 {
 618     dlink_node *link = hlp->servers.head;
 619     helper_stateful_server *srv;
 620
 621     while (link) {
 622         srv = (helper_stateful_server *)link->data;
 623         link = link->next;
 624
 625         if (srv->flags.shutdown) {
 626             debugs(84, 3, "helperStatefulShutdown: " << hlp->id_name << " #" << srv->index + 1 << " has already SHUT DOWN.");
 627             continue;
 628         }
 629
 630         assert(hlp->childs.n_active > 0);
 631         hlp->childs.n_active--;
 632         srv->flags.shutdown = 1;        /* request it to shut itself down */
 633
 634         if (srv->flags.busy) {
 635             debugs(84, 3, "helperStatefulShutdown: " << hlp->id_name << " #" << srv->index + 1 << " is BUSY.");
 636             continue;
 637         }
 638
 639         if (srv->flags.closing) {
 640             debugs(84, 3, "helperStatefulShutdown: " << hlp->id_name << " #" << srv->index + 1 << " is CLOSING.");
 641             continue;
 642         }
 643
 644         if (srv->flags.reserved) {
 645             if (shutting_down) {
 646                 debugs(84, 3, "helperStatefulShutdown: " << hlp->id_name << " #" << srv->index + 1 << " is RESERVED. Closing anyway.");
 647             } else {
 648                 debugs(84, 3, "helperStatefulShutdown: " << hlp->id_name << " #" << srv->index + 1 << " is RESERVED. Not Shutting Down Yet.");
 649                 continue;
 650             }
 651         }
 652
 653         debugs(84, 3, "helperStatefulShutdown: " << hlp->id_name << " #" << srv->index + 1 << " shutting down.");
 654
 655         /* the rest of the details is dealt with in the helperStatefulServerFree
 656          * close handler
 657          */
 658         srv->closePipesSafely();
 659     }
 660 }
 661
 662 helper::~helper()
 663 {
 664     /* note, don't free id_name, it probably points to static memory */
 665
 666     if (queue.head)
 667         debugs(84, 0, "WARNING: freeing " << id_name << " helper with " << stats.queue_size << " requests queued");
 668 }
 669
 670 /* ====================================================================== */
 671 /* LOCAL FUNCTIONS */
 672 /* ====================================================================== */
 673
 674 static void
 675 helperServerFree(helper_server *srv)
 676 {
 677     helper *hlp = srv->parent;
 678     helper_request *r;
 679     int i, concurrency = hlp->childs.concurrency;
 680
 681     if (!concurrency)
 682         concurrency = 1;
 683
 684     if (srv->rbuf) {
 685         memFreeBuf(srv->rbuf_sz, srv->rbuf);
 686         srv->rbuf = NULL;
 687     }
 688
 689     srv->wqueue->clean();
 690     delete srv->wqueue;
 691
 692     if (srv->writebuf) {
 693         srv->writebuf->clean();
 694         delete srv->writebuf;
 695         srv->writebuf = NULL;
 696     }
 697
 698     if (Comm::IsConnOpen(srv->writePipe))
 699         srv->closeWritePipeSafely();
 700
 701     dlinkDelete(&srv->link, &hlp->servers);
 702
 703     assert(hlp->childs.n_running > 0);
 704     hlp->childs.n_running--;
 705
 706     if (!srv->flags.shutdown) {
 707         assert(hlp->childs.n_active > 0);
 708         hlp->childs.n_active--;
 709         debugs(84, DBG_CRITICAL, "WARNING: " << hlp->id_name << " #" << srv->index + 1 << " exited");
 710
 711         if (hlp->childs.needNew() > 0) {
 712             debugs(80, 1, "Too few " << hlp->id_name << " processes are running (need " << hlp->childs.needNew() << "/" << hlp->childs.n_max << ")");
 713
 714             if (hlp->childs.n_active < hlp->childs.n_startup && hlp->last_restart > squid_curtime - 30)
 715                 fatalf("The %s helpers are crashing too rapidly, need help!\n", hlp->id_name);
 716
 717             debugs(80, 1, "Starting new helpers");
 718             helperOpenServers(hlp);
 719         }
 720     }
 721
 722     for (i = 0; i < concurrency; i++) {
 723         if ((r = srv->requests[i])) {
 724             void *cbdata;
 725
 726             if (cbdataReferenceValidDone(r->data, &cbdata))
 727                 r->callback(cbdata, NULL);
 728
 729             helperRequestFree(r);
 730
 731             srv->requests[i] = NULL;
 732         }
 733     }
 734     safe_free(srv->requests);
 735
 736     cbdataReferenceDone(srv->parent);
 737     cbdataFree(srv);
 738 }
 739
 740 static void
 741 helperStatefulServerFree(helper_stateful_server *srv)
 742 {
 743     statefulhelper *hlp = srv->parent;
 744     helper_stateful_request *r;
 745
 746     if (srv->rbuf) {
 747         memFreeBuf(srv->rbuf_sz, srv->rbuf);
 748         srv->rbuf = NULL;
 749     }
 750
 751 #if 0
 752     srv->wqueue->clean();
 753
 754     delete srv->wqueue;
 755
 756 #endif
 757
 758     /* TODO: walk the local queue of requests and carry them all out */
 759     if (Comm::IsConnOpen(srv->writePipe))
 760         srv->closeWritePipeSafely();
 761
 762     dlinkDelete(&srv->link, &hlp->servers);
 763
 764     assert(hlp->childs.n_running > 0);
 765     hlp->childs.n_running--;
 766
 767     if (!srv->flags.shutdown) {
 768         assert( hlp->childs.n_active > 0);
 769         hlp->childs.n_active--;
 770         debugs(84, 0, "WARNING: " << hlp->id_name << " #" << srv->index + 1 << " exited");
 771
 772         if (hlp->childs.needNew() > 0) {
 773             debugs(80, 1, "Too few " << hlp->id_name << " processes are running (need " << hlp->childs.needNew() << "/" << hlp->childs.n_max << ")");
 774
 775             if (hlp->childs.n_active < hlp->childs.n_startup && hlp->last_restart > squid_curtime - 30)
 776                 fatalf("The %s helpers are crashing too rapidly, need help!\n", hlp->id_name);
 777
 778             debugs(80, 1, "Starting new helpers");
 779             helperStatefulOpenServers(hlp);
 780         }
 781     }
 782
 783     if ((r = srv->request)) {
 784         void *cbdata;
 785
 786         if (cbdataReferenceValidDone(r->data, &cbdata))
 787             r->callback(cbdata, srv, NULL);
 788
 789         helperStatefulRequestFree(r);
 790
 791         srv->request = NULL;
 792     }
 793
 794     if (srv->data != NULL)
 795         hlp->datapool->freeOne(srv->data);
 796
 797     cbdataReferenceDone(srv->parent);
 798
 799     cbdataFree(srv);
 800 }
 801
 802 /// Calls back with a pointer to the buffer with the helper output
 803 static void helperReturnBuffer(int request_number, helper_server * srv, helper * hlp, char * msg, char * msg_end)
 804 {
 805     helper_request *r = srv->requests[request_number];
 806     if (r) {
 807         HLPCB *callback = r->callback;
 808
 809         srv->requests[request_number] = NULL;
 810
 811         r->callback = NULL;
 812
 813         void *cbdata = NULL;
 814         if (cbdataReferenceValidDone(r->data, &cbdata))
 815             callback(cbdata, msg);
 816
 817         srv->stats.pending--;
 818
 819         hlp->stats.replies++;
 820
 821         srv->answer_time = current_time;
 822
 823         srv->dispatch_time = r->dispatch_time;
 824
 825         hlp->stats.avg_svc_time =
 826             Math::intAverage(hlp->stats.avg_svc_time,
 827                              tvSubMsec(r->dispatch_time, current_time),
 828                              hlp->stats.replies, REDIRECT_AV_FACTOR);
 829
 830         helperRequestFree(r);
 831     } else {
 832         debugs(84, 1, "helperHandleRead: unexpected reply on channel " <<
 833                request_number << " from " << hlp->id_name << " #" << srv->index + 1 <<
 834                " '" << srv->rbuf << "'");
 835     }
 836     srv->roffset -= (msg_end - srv->rbuf);
 837     memmove(srv->rbuf, msg_end, srv->roffset + 1);
 838
 839     if (!srv->flags.shutdown) {
 840         helperKickQueue(hlp);
 841     } else if (!srv->flags.closing && !srv->stats.pending) {
 842         srv->flags.closing=1;
 843         srv->writePipe->close();
 844         return;
 845     }
 846 }
 847
 848 static void
 849 helperHandleRead(const Comm::ConnectionPointer &conn, char *buf, size_t len, comm_err_t flag, int xerrno, void *data)
 850 {
 851     char *t = NULL;
 852     helper_server *srv = (helper_server *)data;
 853     helper *hlp = srv->parent;
 854     assert(cbdataReferenceValid(data));
 855
 856     /* Bail out early on COMM_ERR_CLOSING - close handlers will tidy up for us */
 857
 858     if (flag == COMM_ERR_CLOSING) {
 859         return;
 860     }
 861
 862     assert(conn->fd == srv->readPipe->fd);
 863
 864     debugs(84, 5, "helperHandleRead: " << len << " bytes from " << hlp->id_name << " #" << srv->index + 1);
 865
 866     if (flag != COMM_OK || len == 0) {
 867         srv->closePipesSafely();
 868         return;
 869     }
 870
 871     srv->roffset += len;
 872     srv->rbuf[srv->roffset] = '\0';
 873     debugs(84, 9, "helperHandleRead: '" << srv->rbuf << "'");
 874
 875     if (!srv->stats.pending) {
 876         /* someone spoke without being spoken to */
 877         debugs(84, 1, "helperHandleRead: unexpected read from " <<
 878                hlp->id_name << " #" << srv->index + 1 << ", " << (int)len <<
 879                " bytes '" << srv->rbuf << "'");
 880
 881         srv->roffset = 0;
 882         srv->rbuf[0] = '\0';
 883     }
 884
 885     while ((t = strchr(srv->rbuf, hlp->eom))) {
 886         /* end of reply found */
 887         char *msg = srv->rbuf;
 888         int i = 0;
 889         debugs(84, 3, "helperHandleRead: end of reply found");
 890
 891         if (t > srv->rbuf && t[-1] == '\r' && hlp->eom == '\n')
 892             t[-1] = '\0';
 893
 894         *t++ = '\0';
 895
 896         if (hlp->childs.concurrency) {
 897             i = strtol(msg, &msg, 10);
 898
 899             while (*msg && xisspace(*msg))
 900                 msg++;
 901         }
 902
 903         helperReturnBuffer(i, srv, hlp, msg, t);
 904     }
 905
 906     if (Comm::IsConnOpen(srv->readPipe)) {
 907         AsyncCall::Pointer call = commCbCall(5,4, "helperHandleRead",
 908                                              CommIoCbPtrFun(helperHandleRead, srv));
 909         comm_read(srv->readPipe, srv->rbuf + srv->roffset, srv->rbuf_sz - srv->roffset - 1, call);
 910     }
 911 }
 912
 913 static void
 914 helperStatefulHandleRead(const Comm::ConnectionPointer &conn, char *buf, size_t len, comm_err_t flag, int xerrno, void *data)
 915 {
 916     char *t = NULL;
 917     helper_stateful_server *srv = (helper_stateful_server *)data;
 918     helper_stateful_request *r;
 919     statefulhelper *hlp = srv->parent;
 920     assert(cbdataReferenceValid(data));
 921
 922     /* Bail out early on COMM_ERR_CLOSING - close handlers will tidy up for us */
 923
 924     if (flag == COMM_ERR_CLOSING) {
 925         return;
 926     }
 927
 928     assert(conn->fd == srv->readPipe->fd);
 929
 930     debugs(84, 5, "helperStatefulHandleRead: " << len << " bytes from " <<
 931            hlp->id_name << " #" << srv->index + 1);
 932
 933
 934     if (flag != COMM_OK || len == 0) {
 935         srv->closePipesSafely();
 936         return;
 937     }
 938
 939     srv->roffset += len;
 940     srv->rbuf[srv->roffset] = '\0';
 941     r = srv->request;
 942
 943     if (r == NULL) {
 944         /* someone spoke without being spoken to */
 945         debugs(84, 1, "helperStatefulHandleRead: unexpected read from " <<
 946                hlp->id_name << " #" << srv->index + 1 << ", " << (int)len <<
 947                " bytes '" << srv->rbuf << "'");
 948
 949         srv->roffset = 0;
 950     }
 951
 952     if ((t = strchr(srv->rbuf, hlp->eom))) {
 953         /* end of reply found */
 954         int called = 1;
 955         debugs(84, 3, "helperStatefulHandleRead: end of reply found");
 956
 957         if (t > srv->rbuf && t[-1] == '\r' && hlp->eom == '\n')
 958             t[-1] = '\0';
 959
 960         *t = '\0';
 961
 962         if (r && cbdataReferenceValid(r->data)) {
 963             r->callback(r->data, srv, srv->rbuf);
 964         } else {
 965             debugs(84, 1, "StatefulHandleRead: no callback data registered");
 966             called = 0;
 967         }
 968
 969         srv->flags.busy = 0;
 970         srv->roffset = 0;
 971         helperStatefulRequestFree(r);
 972         srv->request = NULL;
 973         hlp->stats.replies++;
 974         srv->answer_time = current_time;
 975         hlp->stats.avg_svc_time =
 976             Math::intAverage(hlp->stats.avg_svc_time,
 977                              tvSubMsec(srv->dispatch_time, current_time),
 978                              hlp->stats.replies, REDIRECT_AV_FACTOR);
 979
 980         if (called)
 981             helperStatefulServerDone(srv);
 982         else
 983             helperStatefulReleaseServer(srv);
 984     }
 985
 986     if (Comm::IsConnOpen(srv->readPipe)) {
 987         AsyncCall::Pointer call = commCbCall(5,4, "helperStatefulHandleRead",
 988                                              CommIoCbPtrFun(helperStatefulHandleRead, srv));
 989         comm_read(srv->readPipe, srv->rbuf + srv->roffset, srv->rbuf_sz - srv->roffset - 1, call);
 990     }
 991 }
 992
 993 static void
 994 Enqueue(helper * hlp, helper_request * r)
 995 {
 996     dlink_node *link = (dlink_node *)memAllocate(MEM_DLINK_NODE);
 997     dlinkAddTail(r, link, &hlp->queue);
 998     hlp->stats.queue_size++;
 999
1000     /* do this first so idle=N has a chance to grow the child pool before it hits critical. */
1001     if (hlp->childs.needNew() > 0) {
1002         debugs(84, 0, "Starting new " << hlp->id_name << " helpers...");
1003         helperOpenServers(hlp);
1004         return;
1005     }
1006
1007     if (hlp->stats.queue_size < (int)hlp->childs.n_running)
1008         return;
1009
1010     if (squid_curtime - hlp->last_queue_warn < 600)
1011         return;
1012
1013     if (shutting_down || reconfiguring)
1014         return;
1015
1016     hlp->last_queue_warn = squid_curtime;
1017
1018     debugs(84, 0, "WARNING: All " << hlp->childs.n_active << "/" << hlp->childs.n_max << " " << hlp->id_name << " processes are busy.");
1019     debugs(84, 0, "WARNING: " << hlp->stats.queue_size << " pending requests queued");
1020     debugs(84, 0, "WARNING: Consider increasing the number of " << hlp->id_name << " processes in your config file.");
1021
1022     if (hlp->stats.queue_size > (int)hlp->childs.n_running * 2)
1023         fatalf("Too many queued %s requests", hlp->id_name);
1024 }
1025
1026 static void
1027 StatefulEnqueue(statefulhelper * hlp, helper_stateful_request * r)
1028 {
1029     dlink_node *link = (dlink_node *)memAllocate(MEM_DLINK_NODE);
1030     dlinkAddTail(r, link, &hlp->queue);
1031     hlp->stats.queue_size++;
1032
1033     /* do this first so idle=N has a chance to grow the child pool before it hits critical. */
1034     if (hlp->childs.needNew() > 0) {
1035         debugs(84, 0, "Starting new " << hlp->id_name << " helpers...");
1036         helperStatefulOpenServers(hlp);
1037         return;
1038     }
1039
1040     if (hlp->stats.queue_size < (int)hlp->childs.n_running)
1041         return;
1042
1043     if (hlp->stats.queue_size > (int)hlp->childs.n_running * 2)
1044         fatalf("Too many queued %s requests", hlp->id_name);
1045
1046     if (squid_curtime - hlp->last_queue_warn < 600)
1047         return;
1048
1049     if (shutting_down || reconfiguring)
1050         return;
1051
1052     hlp->last_queue_warn = squid_curtime;
1053
1054     debugs(84, 0, "WARNING: All " << hlp->childs.n_active << "/" << hlp->childs.n_max << " " << hlp->id_name << " processes are busy.");
1055     debugs(84, 0, "WARNING: " << hlp->stats.queue_size << " pending requests queued");
1056     debugs(84, 0, "WARNING: Consider increasing the number of " << hlp->id_name << " processes in your config file.");
1057 }
1058
1059 static helper_request *
1060 Dequeue(helper * hlp)
1061 {
1062     dlink_node *link;
1063     helper_request *r = NULL;
1064
1065     if ((link = hlp->queue.head)) {
1066         r = (helper_request *)link->data;
1067         dlinkDelete(link, &hlp->queue);
1068         memFree(link, MEM_DLINK_NODE);
1069         hlp->stats.queue_size--;
1070     }
1071
1072     return r;
1073 }
1074
1075 static helper_stateful_request *
1076 StatefulDequeue(statefulhelper * hlp)
1077 {
1078     dlink_node *link;
1079     helper_stateful_request *r = NULL;
1080
1081     if ((link = hlp->queue.head)) {
1082         r = (helper_stateful_request *)link->data;
1083         dlinkDelete(link, &hlp->queue);
1084         memFree(link, MEM_DLINK_NODE);
1085         hlp->stats.queue_size--;
1086     }
1087
1088     return r;
1089 }
1090
1091 static helper_server *
1092 GetFirstAvailable(helper * hlp)
1093 {
1094     dlink_node *n;
1095     helper_server *srv;
1096     helper_server *selected = NULL;
1097     debugs(84, 5, "GetFirstAvailable: Running servers " << hlp->childs.n_running);
1098
1099     if (hlp->childs.n_running == 0)
1100         return NULL;
1101
1102     /* Find "least" loaded helper (approx) */
1103     for (n = hlp->servers.head; n != NULL; n = n->next) {
1104         srv = (helper_server *)n->data;
1105
1106         if (selected && selected->stats.pending <= srv->stats.pending)
1107             continue;
1108
1109         if (srv->flags.shutdown)
1110             continue;
1111
1112         if (!srv->stats.pending)
1113             return srv;
1114
1115         if (selected) {
1116             selected = srv;
1117             break;
1118         }
1119
1120         selected = srv;
1121     }
1122
1123     /* Check for overload */
1124     if (!selected) {
1125         debugs(84, 5, "GetFirstAvailable: None available.");
1126         return NULL;
1127     }
1128
1129     if (selected->stats.pending >= (hlp->childs.concurrency ? hlp->childs.concurrency : 1)) {
1130         debugs(84, 3, "GetFirstAvailable: Least-loaded helper is overloaded!");
1131         return NULL;
1132     }
1133
1134     debugs(84, 5, "GetFirstAvailable: returning srv-" << selected->index);
1135     return selected;
1136 }
1137
1138 static helper_stateful_server *
1139 StatefulGetFirstAvailable(statefulhelper * hlp)
1140 {
1141     dlink_node *n;
1142     helper_stateful_server *srv = NULL;
1143     debugs(84, 5, "StatefulGetFirstAvailable: Running servers " << hlp->childs.n_running);
1144
1145     if (hlp->childs.n_running == 0)
1146         return NULL;
1147
1148     for (n = hlp->servers.head; n != NULL; n = n->next) {
1149         srv = (helper_stateful_server *)n->data;
1150
1151         if (srv->flags.busy)
1152             continue;
1153
1154         if (srv->flags.reserved)
1155             continue;
1156
1157         if (srv->flags.shutdown)
1158             continue;
1159
1160         if ((hlp->IsAvailable != NULL) && (srv->data != NULL) && !(hlp->IsAvailable(srv->data)))
1161             continue;
1162
1163         debugs(84, 5, "StatefulGetFirstAvailable: returning srv-" << srv->index);
1164         return srv;
1165     }
1166
1167     debugs(84, 5, "StatefulGetFirstAvailable: None available.");
1168     return NULL;
1169 }
1170
1171
1172 static void
1173 helperDispatchWriteDone(const Comm::ConnectionPointer &conn, char *buf, size_t len, comm_err_t flag, int xerrno, void *data)
1174 {
1175     helper_server *srv = (helper_server *)data;
1176
1177     srv->writebuf->clean();
1178     delete srv->writebuf;
1179     srv->writebuf = NULL;
1180     srv->flags.writing = 0;
1181
1182     if (flag != COMM_OK) {
1183         /* Helper server has crashed */
1184         debugs(84, 0, "helperDispatch: Helper " << srv->parent->id_name << " #" << srv->index + 1 << " has crashed");
1185         return;
1186     }
1187
1188     if (!srv->wqueue->isNull()) {
1189         srv->writebuf = srv->wqueue;
1190         srv->wqueue = new MemBuf;
1191         srv->flags.writing = 1;
1192         AsyncCall::Pointer call = commCbCall(5,5, "helperDispatchWriteDone",
1193                                              CommIoCbPtrFun(helperDispatchWriteDone, srv));
1194         Comm::Write(srv->writePipe, srv->writebuf->content(), srv->writebuf->contentSize(), call, NULL);
1195     }
1196 }
1197
1198 static void
1199 helperDispatch(helper_server * srv, helper_request * r)
1200 {
1201     helper *hlp = srv->parent;
1202     helper_request **ptr = NULL;
1203     unsigned int slot;
1204
1205     if (!cbdataReferenceValid(r->data)) {
1206         debugs(84, 1, "helperDispatch: invalid callback data");
1207         helperRequestFree(r);
1208         return;
1209     }
1210
1211     for (slot = 0; slot < (hlp->childs.concurrency ? hlp->childs.concurrency : 1); slot++) {
1212         if (!srv->requests[slot]) {
1213             ptr = &srv->requests[slot];
1214             break;
1215         }
1216     }
1217
1218     assert(ptr);
1219     *ptr = r;
1220     srv->stats.pending += 1;
1221     r->dispatch_time = current_time;
1222
1223     if (srv->wqueue->isNull())
1224         srv->wqueue->init();
1225
1226     if (hlp->childs.concurrency)
1227         srv->wqueue->Printf("%d %s", slot, r->buf);
1228     else
1229         srv->wqueue->append(r->buf, strlen(r->buf));
1230
1231     if (!srv->flags.writing) {
1232         assert(NULL == srv->writebuf);
1233         srv->writebuf = srv->wqueue;
1234         srv->wqueue = new MemBuf;
1235         srv->flags.writing = 1;
1236         AsyncCall::Pointer call = commCbCall(5,5, "helperDispatchWriteDone",
1237                                              CommIoCbPtrFun(helperDispatchWriteDone, srv));
1238         Comm::Write(srv->writePipe, srv->writebuf->content(), srv->writebuf->contentSize(), call, NULL);
1239     }
1240
1241     debugs(84, 5, "helperDispatch: Request sent to " << hlp->id_name << " #" << srv->index + 1 << ", " << strlen(r->buf) << " bytes");
1242
1243     srv->stats.uses++;
1244     hlp->stats.requests++;
1245 }
1246
1247 static void
1248 helperStatefulDispatchWriteDone(const Comm::ConnectionPointer &conn, char *buf, size_t len, comm_err_t flag,
1249                                 int xerrno, void *data)
1250 {
1251     /* nothing! */
1252 }
1253
1254
1255 static void
1256 helperStatefulDispatch(helper_stateful_server * srv, helper_stateful_request * r)
1257 {
1258     statefulhelper *hlp = srv->parent;
1259
1260     if (!cbdataReferenceValid(r->data)) {
1261         debugs(84, 1, "helperStatefulDispatch: invalid callback data");
1262         helperStatefulRequestFree(r);
1263         helperStatefulReleaseServer(srv);
1264         return;
1265     }
1266
1267     debugs(84, 9, "helperStatefulDispatch busying helper " << hlp->id_name << " #" << srv->index + 1);
1268
1269     if (r->placeholder == 1) {
1270         /* a callback is needed before this request can _use_ a helper. */
1271         /* we don't care about releasing this helper. The request NEVER
1272          * gets to the helper. So we throw away the return code */
1273         r->callback(r->data, srv, NULL);
1274         /* throw away the placeholder */
1275         helperStatefulRequestFree(r);
1276         /* and push the queue. Note that the callback may have submitted a new
1277          * request to the helper which is why we test for the request*/
1278
1279         if (srv->request == NULL)
1280             helperStatefulServerDone(srv);
1281
1282         return;
1283     }
1284
1285     srv->flags.busy = 1;
1286     srv->flags.reserved = 1;
1287     srv->request = r;
1288     srv->dispatch_time = current_time;
1289     AsyncCall::Pointer call = commCbCall(5,5, "helperStatefulDispatchWriteDone",
1290                                          CommIoCbPtrFun(helperStatefulDispatchWriteDone, hlp));
1291     Comm::Write(srv->writePipe, r->buf, strlen(r->buf), call, NULL);
1292     debugs(84, 5, "helperStatefulDispatch: Request sent to " <<
1293            hlp->id_name << " #" << srv->index + 1 << ", " <<
1294            (int) strlen(r->buf) << " bytes");
1295
1296     srv->stats.uses++;
1297     hlp->stats.requests++;
1298 }
1299
1300
1301 static void
1302 helperKickQueue(helper * hlp)
1303 {
1304     helper_request *r;
1305     helper_server *srv;
1306
1307     while ((srv = GetFirstAvailable(hlp)) && (r = Dequeue(hlp)))
1308         helperDispatch(srv, r);
1309 }
1310
1311 static void
1312 helperStatefulKickQueue(statefulhelper * hlp)
1313 {
1314     helper_stateful_request *r;
1315     helper_stateful_server *srv;
1316
1317     while ((srv = StatefulGetFirstAvailable(hlp)) && (r = StatefulDequeue(hlp)))
1318         helperStatefulDispatch(srv, r);
1319 }
1320
1321 static void
1322 helperStatefulServerDone(helper_stateful_server * srv)
1323 {
1324     if (!srv->flags.shutdown) {
1325         helperStatefulKickQueue(srv->parent);
1326     } else if (!srv->flags.closing && !srv->flags.reserved && !srv->flags.busy) {
1327         srv->closeWritePipeSafely();
1328         return;
1329     }
1330 }
1331
1332 static void
1333 helperRequestFree(helper_request * r)
1334 {
1335     cbdataReferenceDone(r->data);
1336     xfree(r->buf);
1337     delete r;
1338 }
1339
1340 static void
1341 helperStatefulRequestFree(helper_stateful_request * r)
1342 {
1343     if (r) {
1344         cbdataReferenceDone(r->data);
1345         xfree(r->buf);
1346         delete r;
1347     }
1348 }
1349
1350 // TODO: should helper_ and helper_stateful_ have a common parent?
1351 static bool
1352 helperStartStats(StoreEntry *sentry, void *hlp, const char *label)
1353 {
1354     if (!hlp) {
1355         if (label)
1356             storeAppendPrintf(sentry, "%s: unavailable\n", label);
1357         return false;
1358     }
1359
1360     if (label)
1361         storeAppendPrintf(sentry, "%s:\n", label);
1362
1363     return true;
1364 }