[thirdparty/squid.git] / src / helper.cc

#include "squid.h"

#define HELPER_MAX_ARGS 64

static PF helperHandleRead;
static PF helperServerFree;
static void Enqueue(helper * hlp, helper_request *);
static helper_request *Dequeue(helper * hlp);
static helper_server *GetFirstAvailable(helper * hlp);
static void helperDispatch(helper_server * srv, helper_request * r);
static void helperKickQueue(helper * hlp);
static void helperRequestFree(helper_request * r);


void
helperOpenServers(helper * hlp)
{
    char *s;
    char *progname;
    char *shortname;
    char *procname;
    char *args[HELPER_MAX_ARGS];
    char fd_note_buf[FD_DESC_SZ];
    helper_server *srv;
    int nargs = 0;
    int k;
    int x;
    int rfd;
    int wfd;
    wordlist *w;
    if (hlp->cmdline == NULL)
	return;
    progname = hlp->cmdline->key;
    if ((s = strrchr(progname, '/')))
	shortname = xstrdup(s + 1);
    else
	shortname = xstrdup(progname);
    debug(29, 1) ("helperOpenServers: Starting %d '%s' processes\n",
	hlp->n_to_start, shortname);
    procname = xmalloc(strlen(shortname) + 3);
    snprintf(procname, strlen(shortname) + 3, "(%s)", shortname);
    args[nargs++] = procname;
    for (w = hlp->cmdline->next; w && nargs < HELPER_MAX_ARGS; w = w->next)
	args[nargs++] = w->key;
    args[nargs++] = NULL;
    assert(nargs <= HELPER_MAX_ARGS);
    for (k = 0; k < hlp->n_to_start; k++) {
	getCurrentTime();
	rfd = wfd = -1;
	x = ipcCreate(hlp->ipc_type,
	    progname,
	    args,
	    shortname,
	    &rfd,
	    &wfd);
	if (x < 0) {
	    debug(29, 1) ("WARNING: Cannot run '%s' process.\n", progname);
	    continue;
	}
	hlp->n_running++;
	srv = memAllocate(MEM_HELPER_SERVER);
	cbdataAdd(srv, memFree, MEM_HELPER_SERVER);
	srv->flags.alive = 1;
	srv->index = k;
	srv->rfd = rfd;
	srv->wfd = wfd;
	srv->buf = memAllocate(MEM_8K_BUF);
	srv->buf_sz = 8192;
	srv->offset = 0;
	srv->parent = hlp;
	cbdataLock(hlp);	/* lock because of the parent backlink */
	dlinkAddTail(srv, &srv->link, &hlp->servers);
	if (rfd == wfd) {
	    snprintf(fd_note_buf, FD_DESC_SZ, "%s #%d", shortname, k + 1);
	    fd_note(rfd, fd_note_buf);
	} else {
	    snprintf(fd_note_buf, FD_DESC_SZ, "reading %s #%d", shortname, k + 1);
	    fd_note(rfd, fd_note_buf);
	    snprintf(fd_note_buf, FD_DESC_SZ, "writing %s #%d", shortname, k + 1);
	    fd_note(wfd, fd_note_buf);
	}
	commSetNonBlocking(rfd);
	if (wfd != rfd)
	    commSetNonBlocking(wfd);
	comm_add_close_handler(rfd, helperServerFree, srv);
    }
    safe_free(shortname);
    safe_free(procname);
    helperKickQueue(hlp);
}

void
helperSubmit(helper * hlp, const char *buf, HLPCB * callback, void *data)
{
    helper_request *r = memAllocate(MEM_HELPER_REQUEST);
    helper_server *srv;
    if (hlp == NULL) {
	debug(29, 3) ("helperSubmit: hlp == NULL\n");
	callback(data, NULL);
	return;
    }
    r->callback = callback;
    r->data = data;
    r->buf = xstrdup(buf);
    cbdataLock(r->data);
    if ((srv = GetFirstAvailable(hlp)))
	helperDispatch(srv, r);
    else
	Enqueue(hlp, r);
}

void
helperStats(StoreEntry * sentry, helper * hlp)
{
    helper_server *srv;
    dlink_node *link;
    double tt;
    storeAppendPrintf(sentry, "number running: %d of %d\n",
	hlp->n_running, hlp->n_to_start);
    storeAppendPrintf(sentry, "requests sent: %d\n",
	hlp->stats.requests);
    storeAppendPrintf(sentry, "replies received: %d\n",
	hlp->stats.replies);
    storeAppendPrintf(sentry, "queue length: %d\n",
	hlp->stats.queue_size);
    storeAppendPrintf(sentry, "avg service time: %d msec\n",
	hlp->stats.avg_svc_time);
    storeAppendPrintf(sentry, "\n");
    storeAppendPrintf(sentry, "%7s\t%7s\t%11s\t%s\t%7s\t%7s\n",
	"#",
	"FD",
	"# Requests",
	"Flags",
	"Time",
	"Offset");
    for (link = hlp->servers.head; link; link = link->next) {
	srv = link->data;
	tt = 0.001 * tvSubMsec(srv->dispatch_time, current_time);
	storeAppendPrintf(sentry, "%7d\t%7d\t%11d\t%c%c%c%c\t%7.3f\t%7d\n",
	    srv->index + 1,
	    srv->rfd,
	    srv->stats.uses,
	    srv->flags.alive ? 'A' : ' ',
	    srv->flags.busy ? 'B' : ' ',
	    srv->flags.closing ? 'C' : ' ',
	    srv->flags.shutdown ? 'S' : ' ',
	    tt < 0.0 ? 0.0 : tt,
	    (int) srv->offset);
    }
    storeAppendPrintf(sentry, "\nFlags key:\n\n");
    storeAppendPrintf(sentry, "   A = ALIVE\n");
    storeAppendPrintf(sentry, "   B = BUSY\n");
    storeAppendPrintf(sentry, "   C = CLOSING\n");
    storeAppendPrintf(sentry, "   S = SHUTDOWN\n");
}

void
helperShutdown(helper * hlp)
{
    dlink_node *link = hlp->servers.head;
    helper_server *srv;
    while (link) {
	srv = link->data;
	link = link->next;
	if (!srv->flags.alive) {
	    debug(34, 3) ("helperShutdown: %s #%d is NOT ALIVE.\n",
		hlp->id_name, srv->index + 1);
	    continue;
	}
	srv->flags.shutdown = 1;	/* request it to shut itself down */
	if (srv->flags.busy) {
	    debug(34, 3) ("helperShutdown: %s #%d is BUSY.\n",
		hlp->id_name, srv->index + 1);
	    continue;
	}
	if (srv->flags.closing) {
	    debug(34, 3) ("helperShutdown: %s #%d is CLOSING.\n",
		hlp->id_name, srv->index + 1);
	    continue;
	}
	srv->flags.closing = 1;
	comm_close(srv->rfd);
    }
}

helper *
helperCreate(const char *name)
{
    helper *hlp = memAllocate(MEM_HELPER);
    cbdataAdd(hlp, memFree, MEM_HELPER);
    hlp->id_name = name;
    return hlp;
}

void
helperFree(helper * hlp)
{
    /* note, don't free hlp->name, it probably points to static memory */
    if (hlp->queue.head)
	debug(29, 0) ("WARNING: freeing %s helper with %d requests queued\n",
	    hlp->id_name, hlp->stats.queue_size);
    cbdataFree(hlp);
}

/* ====================================================================== */
/* LOCAL FUNCTIONS */
/* ====================================================================== */

static void
helperServerFree(int fd, void *data)
{
    helper_server *srv = data;
    helper *hlp = srv->parent;
    helper_request *r;
    assert(srv->rfd == fd);
    if (srv->buf) {
	memFree(srv->buf, MEM_8K_BUF);
	srv->buf = NULL;
    }
    if ((r = srv->request)) {
	if (cbdataValid(r->data))
	    r->callback(r->data, srv->buf);
	helperRequestFree(r);
	srv->request = NULL;
    }
    if (srv->wfd != srv->rfd)
	comm_close(srv->wfd);
    dlinkDelete(&srv->link, &hlp->servers);
    hlp->n_running--;
    assert(hlp->n_running >= 0);
    if (!srv->flags.shutdown) {
	debug(34, 0) ("WARNING: %s #%d (FD %d) exited\n",
	    hlp->id_name, srv->index + 1, fd);
	if (hlp->n_running < hlp->n_to_start / 2)
	    fatalf("Too few %s processes are running", hlp->id_name);
    }
    cbdataUnlock(srv->parent);
    cbdataFree(srv);
}

static void
helperHandleRead(int fd, void *data)
{
    int len;
    char *t = NULL;
    helper_server *srv = data;
    helper_request *r;
    helper *hlp = srv->parent;
    assert(fd == srv->rfd);
    assert(cbdataValid(data));
    Counter.syscalls.sock.reads++;
    len = read(fd, srv->buf + srv->offset, srv->buf_sz - srv->offset);
    fd_bytes(fd, len, FD_READ);
    debug(29, 5) ("helperHandleRead: %d bytes from %s #%d.\n",
	len, hlp->id_name, srv->index + 1);
    if (len <= 0) {
	if (len < 0)
	    debug(50, 1) ("helperHandleRead: FD %d read: %s\n", fd, xstrerror());
	comm_close(fd);
	return;
    }
    srv->offset += len;
    srv->buf[srv->offset] = '\0';
    r = srv->request;
    if (r == NULL) {
	/* someone spoke without being spoken to */
	debug(29, 1) ("helperHandleRead: unexpected read from %s #%d, %d bytes\n",
	    hlp->id_name, srv->index + 1, len);
	srv->offset = 0;
    } else if ((t = strchr(srv->buf, '\n'))) {
	/* end of reply found */
	debug(29, 3) ("helperHandleRead: end of reply found\n");
	*t = '\0';
	if (cbdataValid(r->data))
	    r->callback(r->data, srv->buf);
	srv->flags.busy = 0;
	srv->offset = 0;
	helperRequestFree(r);
	srv->request = NULL;
	hlp->stats.replies++;
	hlp->stats.avg_svc_time =
	    intAverage(hlp->stats.avg_svc_time,
	    tvSubMsec(srv->dispatch_time, current_time),
	    hlp->stats.replies, REDIRECT_AV_FACTOR);
	if (srv->flags.shutdown)
	    comm_close(srv->wfd);
	else
	    helperKickQueue(hlp);
    } else {
	commSetSelect(srv->rfd, COMM_SELECT_READ, helperHandleRead, srv, 0);
    }
}

static void
Enqueue(helper * hlp, helper_request * r)
{
    dlink_node *link = memAllocate(MEM_DLINK_NODE);
    dlinkAddTail(r, link, &hlp->queue);
    hlp->stats.queue_size++;
    if (hlp->stats.queue_size < hlp->n_running)
	return;
    if (squid_curtime - hlp->last_queue_warn < 600)
	return;
    if (shutting_down || reconfiguring)
	return;
    hlp->last_queue_warn = squid_curtime;
    debug(14, 0) ("WARNING: All %s processes are busy.\n", hlp->id_name);
    debug(14, 0) ("WARNING: %d pending requests queued\n", hlp->stats.queue_size);
    if (hlp->stats.queue_size > hlp->n_running * 2)
	fatalf("Too many queued %s requests", hlp->id_name);
    debug(14, 1) ("Consider increasing the number of %s processes in your config file.\n", hlp->id_name);
}

static helper_request *
Dequeue(helper * hlp)
{
    dlink_node *link;
    helper_request *r = NULL;
    if ((link = hlp->queue.head)) {
	r = link->data;
	dlinkDelete(link, &hlp->queue);
	memFree(link, MEM_DLINK_NODE);
	hlp->stats.queue_size--;
    }
    return r;
}

static helper_server *
GetFirstAvailable(helper * hlp)
{
    dlink_node *n;
    helper_server *srv = NULL;
    if (hlp->n_running == 0)
	return NULL;
    for (n = hlp->servers.head; n != NULL; n = n->next) {
	srv = n->data;
	if (srv->flags.busy)
	    continue;
	if (!srv->flags.alive)
	    continue;
	return srv;
    }
    return NULL;
}

static void
helperDispatch(helper_server * srv, helper_request * r)
{
    helper *hlp = srv->parent;
    if (!cbdataValid(r->data)) {
	debug(29, 1) ("helperDispatch: invalid callback data\n");
	helperRequestFree(r);
	return;
    }
    assert(!srv->flags.busy);
    srv->flags.busy = 1;
    srv->request = r;
    srv->dispatch_time = current_time;
    comm_write(srv->wfd,
	r->buf,
	strlen(r->buf),
	NULL,			/* Handler */
	NULL,			/* Handler-data */
	NULL);			/* free */
    commSetSelect(srv->rfd,
	COMM_SELECT_READ,
	helperHandleRead,
	srv, 0);
    debug(29, 5) ("helperDispatch: Request sent to %s #%d, %d bytes\n",
	hlp->id_name, srv->index + 1, strlen(r->buf));
    srv->stats.uses++;
    hlp->stats.requests++;
}

static void
helperKickQueue(helper * hlp)
{
    helper_request *r;
    helper_server *srv;
    while ((srv = GetFirstAvailable(hlp)) && (r = Dequeue(hlp)))
	helperDispatch(srv, r);
}

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