src/helper.cc

   1 #include "squid.h"
   2
   3 #define HELPER_MAX_ARGS 64
   4
   5 static PF helperHandleRead;
   6 static PF helperServerFree;
   7 static void Enqueue(helper * hlp, helper_request *);
   8 static helper_request *Dequeue(helper * hlp);
   9 static helper_server *GetFirstAvailable(helper * hlp);
  10 static void helperDispatch(helper_server * srv, helper_request * r);
  11 static void helperKickQueue(helper * hlp);
  12 static void helperRequestFree(helper_request * r);
  13
  14
  15 void
  16 helperOpenServers(helper * hlp)
  17 {
  18     char *s;
  19     char *progname;
  20     char *shortname;
  21     char *procname;
  22     char *args[HELPER_MAX_ARGS];
  23     char fd_note_buf[FD_DESC_SZ];
  24     helper_server *srv;
  25     int nargs = 0;
  26     int k;
  27     int x;
  28     int rfd;
  29     int wfd;
  30     wordlist *w;
  31     if (hlp->cmdline == NULL)
  32         return;
  33     progname = hlp->cmdline->key;
  34     assert(hlp->servers.head == NULL);
  35     assert(hlp->servers.tail == NULL);
  36     if ((s = strrchr(progname, '/')))
  37         shortname = xstrdup(s + 1);
  38     else
  39         shortname = xstrdup(progname);
  40     debug(29, 1) ("helperOpenServers: Starting %d '%s' processes\n",
  41         hlp->n_to_start, shortname);
  42     procname = xmalloc(strlen(shortname) + 3);
  43     snprintf(procname, strlen(shortname) + 3, "(%s)", shortname);
  44     args[nargs++] = procname;
  45     for (w = hlp->cmdline->next; w && nargs < HELPER_MAX_ARGS; w = w->next)
  46         args[nargs++] = w->key;
  47     args[nargs++] = NULL;
  48     assert(nargs <= HELPER_MAX_ARGS);
  49     for (k = 0; k < hlp->n_to_start; k++) {
  50         rfd = wfd = -1;
  51         x = ipcCreate(hlp->ipc_type,
  52             progname,
  53             args,
  54             shortname,
  55             &rfd,
  56             &wfd);
  57         if (x < 0) {
  58             debug(29, 1) ("WARNING: Cannot run '%s' process.\n", progname);
  59             continue;
  60         }
  61         hlp->n_running++;
  62         srv = xcalloc(1, sizeof(*srv));
  63         cbdataAdd(srv, MEM_NONE);
  64         srv->flags.alive = 1;
  65         srv->index = k;
  66         srv->rfd = rfd;
  67         srv->wfd = wfd;
  68         srv->buf = memAllocate(MEM_8K_BUF);
  69         srv->buf_sz = 8192;
  70         srv->offset = 0;
  71         srv->parent = hlp;
  72         cbdataLock(hlp);        /* lock because of the parent backlink */
  73         dlinkAddTail(srv, &srv->link, &hlp->servers);
  74         if (rfd == wfd) {
  75             snprintf(fd_note_buf, FD_DESC_SZ, "%s #%d", shortname, k + 1);
  76             fd_note(rfd, fd_note_buf);
  77         } else {
  78             snprintf(fd_note_buf, FD_DESC_SZ, "reading %s #%d", shortname, k + 1);
  79             fd_note(rfd, fd_note_buf);
  80             snprintf(fd_note_buf, FD_DESC_SZ, "writing %s #%d", shortname, k + 1);
  81             fd_note(wfd, fd_note_buf);
  82         }
  83         commSetNonBlocking(rfd);
  84         if (wfd != rfd)
  85             commSetNonBlocking(wfd);
  86         comm_add_close_handler(rfd, helperServerFree, srv);
  87     }
  88     safe_free(shortname);
  89     safe_free(procname);
  90 }
  91
  92 void
  93 helperSubmit(helper * hlp, const char *buf, HLPCB * callback, void *data)
  94 {
  95     helper_request *r = xcalloc(1, sizeof(*r));
  96     helper_server *srv;
  97     if (hlp == NULL) {
  98         debug(29,3)("helperSubmit: hlp == NULL\n");
  99         callback(data, NULL);
 100         return;
 101     }
 102     r->callback = callback;
 103     r->data = data;
 104     r->buf = xstrdup(buf);
 105     cbdataLock(r->data);
 106     if ((srv = GetFirstAvailable(hlp)))
 107         helperDispatch(srv, r);
 108     else
 109         Enqueue(hlp, r);
 110 }
 111
 112 void
 113 helperStats(StoreEntry * sentry, helper * hlp)
 114 {
 115     helper_server *srv;
 116     dlink_node *link;
 117     storeAppendPrintf(sentry, "number running: %d of %d\n",
 118         hlp->n_running, hlp->n_to_start);
 119     storeAppendPrintf(sentry, "requests sent: %d\n",
 120         hlp->stats.requests);
 121     storeAppendPrintf(sentry, "replies received: %d\n",
 122         hlp->stats.replies);
 123     storeAppendPrintf(sentry, "queue length: %d\n",
 124         hlp->stats.queue_size);
 125     storeAppendPrintf(sentry, "avg service time: %d msec\n",
 126         hlp->stats.avg_svc_time);
 127     storeAppendPrintf(sentry, "\n");
 128     storeAppendPrintf(sentry, "%7s\t%7s\t%11s\t%s\t%7s\t%7s\n",
 129         "#",
 130         "FD",
 131         "# Requests",
 132         "Flags",
 133         "Time",
 134         "Offset");
 135     for (link = hlp->servers.head; link; link = link->next) {
 136         srv = link->data;
 137         storeAppendPrintf(sentry, "%7d\t%7d\t%11d\t%c%c%c%c\t%7.3f\t%7d\n",
 138             srv->index + 1,
 139             srv->rfd,
 140             srv->stats.uses,
 141             srv->flags.alive ? 'A' : ' ',
 142             srv->flags.busy ? 'B' : ' ',
 143             srv->flags.closing ? 'C' : ' ',
 144             srv->flags.shutdown ? 'S' : ' ',
 145             0.001 * tvSubMsec(srv->dispatch_time, current_time),
 146             (int) srv->offset);
 147     }
 148     storeAppendPrintf(sentry, "\nFlags key:\n\n");
 149     storeAppendPrintf(sentry, "   A = ALIVE\n");
 150     storeAppendPrintf(sentry, "   B = BUSY\n");
 151     storeAppendPrintf(sentry, "   C = CLOSING\n");
 152     storeAppendPrintf(sentry, "   S = SHUTDOWN\n");
 153 }
 154
 155 void
 156 helperShutdown(helper * hlp)
 157 {
 158     dlink_node *link;
 159     helper_server *srv;
 160     for (link = hlp->servers.head; link; link = link->next) {
 161         srv = link->data;
 162         if (!srv->flags.alive) {
 163             debug(34, 3) ("helperShutdown: %s #%d is NOT ALIVE.\n",
 164                 hlp->id_name, srv->index + 1);
 165             continue;
 166         }
 167         srv->flags.shutdown = 1;        /* request it to shut itself down */
 168         if (srv->flags.busy) {
 169             debug(34, 3) ("helperShutdown: %s #%d is BUSY.\n",
 170                 hlp->id_name, srv->index + 1);
 171             continue;
 172         }
 173         if (srv->flags.closing) {
 174             debug(34, 3) ("helperShutdown: %s #%d is CLOSING.\n",
 175                 hlp->id_name, srv->index + 1);
 176             continue;
 177         }
 178         srv->flags.closing = 1;
 179         comm_close(srv->wfd);
 180     }
 181 }
 182
 183 helper *
 184 helperCreate(const char *name)
 185 {
 186     helper *hlp = xcalloc(1, sizeof(*hlp));
 187     cbdataAdd(hlp, MEM_NONE);
 188     hlp->id_name = name;
 189     return hlp;
 190 }
 191
 192 void
 193 helperFree(helper * hlp)
 194 {
 195     /* note, don't free hlp->name, it probably points to static memory */
 196     if (hlp->queue.head)
 197         debug(29, 0) ("WARNING: freeing %s helper with %d requests queued\n",
 198             hlp->id_name, hlp->stats.queue_size);
 199     cbdataFree(hlp);
 200 }
 201
 202 /* ====================================================================== */
 203 /* LOCAL FUNCTIONS */
 204 /* ====================================================================== */
 205
 206 static void
 207 helperServerFree(int fd, void *data)
 208 {
 209     helper_server *srv = data;
 210     helper *hlp = srv->parent;
 211     assert(srv->rfd == fd);
 212     if (srv->buf) {
 213         memFree(MEM_8K_BUF, srv->buf);
 214         srv->buf = NULL;
 215     }
 216     if (srv->wfd != srv->rfd)
 217         comm_close(srv->wfd);
 218     dlinkDelete(&srv->link, &hlp->servers);
 219     hlp->n_running--;
 220     assert(hlp->n_running >= 0);
 221     if (!srv->flags.shutdown) {
 222         debug(34, 0) ("WARNING: %s #%d (FD %d) exited\n",
 223             hlp->id_name, srv->index + 1, fd);
 224         assert(hlp->n_running >= hlp->n_to_start / 2);
 225         if (hlp->n_running < hlp->n_to_start / 2)
 226             fatalf("Too few %s processes are running", hlp->id_name);
 227     }
 228     cbdataUnlock(srv->parent);
 229     cbdataFree(srv);
 230 }
 231
 232 static void
 233 helperHandleRead(int fd, void *data)
 234 {
 235     int len;
 236     char *t = NULL;
 237     helper_server *srv = data;
 238     helper_request *r;
 239     helper *hlp = srv->parent;
 240     assert(fd == srv->rfd);
 241     assert(cbdataValid(data));
 242     Counter.syscalls.sock.reads++;
 243     len = read(fd, srv->buf + srv->offset, srv->buf_sz - srv->offset);
 244     fd_bytes(fd, len, FD_READ);
 245     debug(29, 5) ("helperHandleRead: %d bytes from %s #%d.\n",
 246         len, hlp->id_name, srv->index + 1);
 247     if (len <= 0) {
 248         if (len < 0)
 249             debug(50, 1) ("helperHandleRead: FD %d read: %s\n", fd, xstrerror());
 250         comm_close(fd);
 251         return;
 252     }
 253     srv->offset += len;
 254     srv->buf[srv->offset] = '\0';
 255     r = srv->request;
 256     if (r == NULL) {
 257         /* someone spoke without being spoken to */
 258         debug(29, 1) ("helperHandleRead: unexpected read from %s #%d, %d bytes\n",
 259             hlp->id_name, srv->index + 1, len);
 260         srv->offset = 0;
 261     } else if ((t = strchr(srv->buf, '\n'))) {
 262         /* end of reply found */
 263         debug(29, 3) ("helperHandleRead: end of reply found\n");
 264         *t = '\0';
 265         if (cbdataValid(r->data))
 266             r->callback(r->data, srv->buf);
 267         srv->flags.busy = 0;
 268         srv->offset = 0;
 269         helperRequestFree(r);
 270         hlp->stats.replies++;
 271         hlp->stats.avg_svc_time =
 272             intAverage(hlp->stats.avg_svc_time,
 273             tvSubMsec(srv->dispatch_time, current_time),
 274             hlp->stats.replies, REDIRECT_AV_FACTOR);
 275         if (srv->flags.shutdown)
 276             comm_close(srv->wfd);
 277     } else {
 278         commSetSelect(srv->rfd, COMM_SELECT_READ, helperHandleRead, srv, 0);
 279     }
 280     helperKickQueue(hlp);
 281 }
 282
 283 static void
 284 Enqueue(helper * hlp, helper_request * r)
 285 {
 286     dlink_node *link = xcalloc(1, sizeof(*link));
 287     dlinkAddTail(r, link, &hlp->queue);
 288     hlp->stats.queue_size++;
 289     if (hlp->stats.queue_size < hlp->n_running)
 290         return;
 291     if (squid_curtime - hlp->last_queue_warn < 600)
 292         return;
 293     if (shutting_down || reconfiguring)
 294         return;
 295     hlp->last_queue_warn = squid_curtime;
 296     debug(14, 0) ("WARNING: All %s processes are busy.\n", hlp->id_name);
 297     debug(14, 0) ("WARNING: %d pending requests queued\n", hlp->stats.queue_size);
 298     if (hlp->stats.queue_size > hlp->n_running * 2)
 299         fatalf("Too many queued %s requests", hlp->id_name);
 300     debug(14, 1) ("Consider increasing the number of %s processes in your config file.\n", hlp->id_name);
 301 }
 302
 303 static helper_request *
 304 Dequeue(helper * hlp)
 305 {
 306     dlink_node *link;
 307     helper_request *r = NULL;
 308     if ((link = hlp->queue.head)) {
 309         r = link->data;
 310         dlinkDelete(link, &hlp->queue);
 311         safe_free(link);
 312         hlp->stats.queue_size--;
 313     }
 314     return r;
 315 }
 316
 317 static helper_server *
 318 GetFirstAvailable(helper * hlp)
 319 {
 320     dlink_node *n;
 321     helper_server *srv = NULL;
 322     if (hlp->n_running == 0)
 323         return NULL;
 324     for (n = hlp->servers.head; n != NULL; n = n->next) {
 325         srv = n->data;
 326         if (srv->flags.busy)
 327             continue;
 328         if (!srv->flags.alive)
 329             continue;
 330         return srv;
 331     }
 332     return NULL;
 333 }
 334
 335 static void
 336 helperDispatch(helper_server * srv, helper_request * r)
 337 {
 338     helper *hlp = srv->parent;
 339     if (!cbdataValid(r->data)) {
 340         debug(29, 1) ("helperDispatch: invalid callback data\n");
 341         helperRequestFree(r);
 342         return;
 343     }
 344     assert(!srv->flags.busy);
 345     srv->flags.busy = 1;
 346     srv->request = r;
 347     srv->dispatch_time = current_time;
 348     comm_write(srv->wfd,
 349         r->buf,
 350         strlen(r->buf),
 351         NULL,                   /* Handler */
 352         NULL,                   /* Handler-data */
 353         NULL);                  /* free */
 354     commSetSelect(srv->rfd,
 355         COMM_SELECT_READ,
 356         helperHandleRead,
 357         srv, 0);
 358     debug(29, 5) ("helperDispatch: Request sent to %s #%d, %d bytes\n",
 359         hlp->id_name, srv->index + 1, strlen(r->buf));
 360     srv->stats.uses++;
 361     hlp->stats.requests++;
 362 }
 363
 364 static void
 365 helperKickQueue(helper * hlp)
 366 {
 367     helper_request *r;
 368     helper_server *srv;
 369     while ((srv = GetFirstAvailable(hlp)) && (r = Dequeue(hlp)))
 370         helperDispatch(srv, r);
 371 }
 372
 373 static void
 374 helperRequestFree(helper_request * r)
 375 {
 376     cbdataUnlock(r->data);
 377     xfree(r->buf);
 378     xfree(r);
 379 }