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


/*
 * $Id: helper.cc,v 1.21 2001/01/05 09:51:38 adrian Exp $
 *
 * DEBUG: section 29    Helper process maintenance
 * AUTHOR: Harvest Derived?
 *
 * SQUID Internet Object Cache  http://squid.nlanr.net/Squid/
 * ----------------------------------------------------------
 *
 *  Squid is the result of efforts by numerous individuals from the
 *  Internet community.  Development is led by Duane Wessels of the
 *  National Laboratory for Applied Network Research and funded by the
 *  National Science Foundation.  Squid is Copyrighted (C) 1998 by
 *  the Regents of the University of California.  Please see the
 *  COPYRIGHT file for full details.  Squid incorporates software
 *  developed and/or copyrighted by other sources.  Please see the
 *  CREDITS file for full details.
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2 of the License, or
 *  (at your option) any later version.
 *  
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *  
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
 *
 */

#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 = CBDATA_ALLOC(helper_server, NULL);
	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\t%7s\n",
	"#",
	"FD",
	"# Requests",
	"Flags",
	"Time",
	"Offset",
	"Request");
    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\t%s\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,
	    srv->request ? log_quote(srv->request->buf) : "(none)");
    }
    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->wfd);
	srv->wfd = -1;
    }
}

helper *
helperCreate(const char *name)
{
    helper *hlp;
    hlp = CBDATA_ALLOC(helper, NULL);
    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 && srv->wfd != -1)
	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));
    statCounter.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);
	    srv->wfd = -1;
	} 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
f740a279	1
f740a279	2	/*
28c60158	3	* $Id: helper.cc,v 1.21 2001/01/05 09:51:38 adrian Exp $
f740a279	4	*
	5	* DEBUG: section 29 Helper process maintenance
	6	* AUTHOR: Harvest Derived?
	7	*
	8	* SQUID Internet Object Cache http://squid.nlanr.net/Squid/
	9	* ----------------------------------------------------------
	10	*
	11	* Squid is the result of efforts by numerous individuals from the
	12	* Internet community. Development is led by Duane Wessels of the
	13	* National Laboratory for Applied Network Research and funded by the
	14	* National Science Foundation. Squid is Copyrighted (C) 1998 by
	15	* the Regents of the University of California. Please see the
	16	* COPYRIGHT file for full details. Squid incorporates software
	17	* developed and/or copyrighted by other sources. Please see the
	18	* CREDITS file for full details.
	19	*
	20	* This program is free software; you can redistribute it and/or modify
	21	* it under the terms of the GNU General Public License as published by
	22	* the Free Software Foundation; either version 2 of the License, or
	23	* (at your option) any later version.
	24	*
	25	* This program is distributed in the hope that it will be useful,
	26	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	27	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	28	* GNU General Public License for more details.
	29	*
	30	* You should have received a copy of the GNU General Public License
	31	* along with this program; if not, write to the Free Software
	32	* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
	33	*
	34	*/
	35
74addf6c	36	#include "squid.h"
	37
	38	#define HELPER_MAX_ARGS 64
	39
	40	static PF helperHandleRead;
1f5f60dd	41	static PF helperServerFree;
74addf6c	42	static void Enqueue(helper * hlp, helper_request *);
	43	static helper_request Dequeue(helper hlp);
	44	static helper_server GetFirstAvailable(helper hlp);
	45	static void helperDispatch(helper_server * srv, helper_request * r);
	46	static void helperKickQueue(helper * hlp);
	47	static void helperRequestFree(helper_request * r);
	48
	49
	50	void
	51	helperOpenServers(helper * hlp)
	52	{
	53	char *s;
	54	char *progname;
	55	char *shortname;
	56	char *procname;
	57	char *args[HELPER_MAX_ARGS];
	58	char fd_note_buf[FD_DESC_SZ];
	59	helper_server *srv;
	60	int nargs = 0;
	61	int k;
	62	int x;
	63	int rfd;
	64	int wfd;
	65	wordlist *w;
	66	if (hlp->cmdline == NULL)
	67	return;
	68	progname = hlp->cmdline->key;
74addf6c	69	if ((s = strrchr(progname, '/')))
	70	shortname = xstrdup(s + 1);
	71	else
	72	shortname = xstrdup(progname);
	73	debug(29, 1) ("helperOpenServers: Starting %d '%s' processes\n",
	74	hlp->n_to_start, shortname);
	75	procname = xmalloc(strlen(shortname) + 3);
	76	snprintf(procname, strlen(shortname) + 3, "(%s)", shortname);
	77	args[nargs++] = procname;
	78	for (w = hlp->cmdline->next; w && nargs < HELPER_MAX_ARGS; w = w->next)
	79	args[nargs++] = w->key;
	80	args[nargs++] = NULL;
	81	assert(nargs <= HELPER_MAX_ARGS);
	82	for (k = 0; k < hlp->n_to_start; k++) {
c68e9c6b	83	getCurrentTime();
74addf6c	84	rfd = wfd = -1;
	85	x = ipcCreate(hlp->ipc_type,
	86	progname,
	87	args,
	88	shortname,
	89	&rfd,
	90	&wfd);
	91	if (x < 0) {
	92	debug(29, 1) ("WARNING: Cannot run '%s' process.\n", progname);
	93	continue;
	94	}
	95	hlp->n_running++;
28c60158	96	srv = CBDATA_ALLOC(helper_server, NULL);
74addf6c	97	srv->flags.alive = 1;
	98	srv->index = k;
	99	srv->rfd = rfd;
	100	srv->wfd = wfd;
	101	srv->buf = memAllocate(MEM_8K_BUF);
	102	srv->buf_sz = 8192;
	103	srv->offset = 0;
	104	srv->parent = hlp;
1f5f60dd	105	cbdataLock(hlp); /* lock because of the parent backlink */
74addf6c	106	dlinkAddTail(srv, &srv->link, &hlp->servers);
	107	if (rfd == wfd) {
	108	snprintf(fd_note_buf, FD_DESC_SZ, "%s #%d", shortname, k + 1);
	109	fd_note(rfd, fd_note_buf);
	110	} else {
	111	snprintf(fd_note_buf, FD_DESC_SZ, "reading %s #%d", shortname, k + 1);
	112	fd_note(rfd, fd_note_buf);
	113	snprintf(fd_note_buf, FD_DESC_SZ, "writing %s #%d", shortname, k + 1);
	114	fd_note(wfd, fd_note_buf);
	115	}
	116	commSetNonBlocking(rfd);
	117	if (wfd != rfd)
	118	commSetNonBlocking(wfd);
1f5f60dd	119	comm_add_close_handler(rfd, helperServerFree, srv);
74addf6c	120	}
	121	safe_free(shortname);
	122	safe_free(procname);
838b993c	123	helperKickQueue(hlp);
74addf6c	124	}
	125
	126	void
	127	helperSubmit(helper * hlp, const char buf, HLPCB callback, void *data)
	128	{
c68e9c6b	129	helper_request *r = memAllocate(MEM_HELPER_REQUEST);
74addf6c	130	helper_server *srv;
5b5f9257	131	if (hlp == NULL) {
7d47d8e6	132	debug(29, 3) ("helperSubmit: hlp == NULL\n");
5b5f9257	133	callback(data, NULL);
	134	return;
	135	}
74addf6c	136	r->callback = callback;
	137	r->data = data;
	138	r->buf = xstrdup(buf);
	139	cbdataLock(r->data);
	140	if ((srv = GetFirstAvailable(hlp)))
	141	helperDispatch(srv, r);
	142	else
	143	Enqueue(hlp, r);
	144	}
	145
	146	void
	147	helperStats(StoreEntry * sentry, helper * hlp)
	148	{
	149	helper_server *srv;
	150	dlink_node *link;
f4ae18d0	151	double tt;
74addf6c	152	storeAppendPrintf(sentry, "number running: %d of %d\n",
	153	hlp->n_running, hlp->n_to_start);
	154	storeAppendPrintf(sentry, "requests sent: %d\n",
	155	hlp->stats.requests);
	156	storeAppendPrintf(sentry, "replies received: %d\n",
	157	hlp->stats.replies);
	158	storeAppendPrintf(sentry, "queue length: %d\n",
	159	hlp->stats.queue_size);
	160	storeAppendPrintf(sentry, "avg service time: %d msec\n",
	161	hlp->stats.avg_svc_time);
	162	storeAppendPrintf(sentry, "\n");
592da4ec	163	storeAppendPrintf(sentry, "%7s\t%7s\t%11s\t%s\t%7s\t%7s\t%7s\n",
74addf6c	164	"#",
	165	"FD",
	166	"# Requests",
	167	"Flags",
	168	"Time",
592da4ec	169	"Offset",
592da4ec	170	"Request");
74addf6c	171	for (link = hlp->servers.head; link; link = link->next) {
74addf6c	172	srv = link->data;
f4ae18d0	173	tt = 0.001 * tvSubMsec(srv->dispatch_time, current_time);
592da4ec	174	storeAppendPrintf(sentry, "%7d\t%7d\t%11d\t%c%c%c%c\t%7.3f\t%7d\t%s\n",
74addf6c	175	srv->index + 1,
	176	srv->rfd,
	177	srv->stats.uses,
	178	srv->flags.alive ? 'A' : ' ',
	179	srv->flags.busy ? 'B' : ' ',
	180	srv->flags.closing ? 'C' : ' ',
	181	srv->flags.shutdown ? 'S' : ' ',
f4ae18d0	182	tt < 0.0 ? 0.0 : tt,
592da4ec	183	(int) srv->offset,
592da4ec	184	srv->request ? log_quote(srv->request->buf) : "(none)");
74addf6c	185	}
	186	storeAppendPrintf(sentry, "\nFlags key:\n\n");
	187	storeAppendPrintf(sentry, " A = ALIVE\n");
	188	storeAppendPrintf(sentry, " B = BUSY\n");
	189	storeAppendPrintf(sentry, " C = CLOSING\n");
	190	storeAppendPrintf(sentry, " S = SHUTDOWN\n");
	191	}
	192
	193	void
	194	helperShutdown(helper * hlp)
	195	{
c68e9c6b	196	dlink_node *link = hlp->servers.head;
74addf6c	197	helper_server *srv;
c68e9c6b	198	while (link) {
74addf6c	199	srv = link->data;
c68e9c6b	200	link = link->next;
74addf6c	201	if (!srv->flags.alive) {
	202	debug(34, 3) ("helperShutdown: %s #%d is NOT ALIVE.\n",
	203	hlp->id_name, srv->index + 1);
	204	continue;
	205	}
1f5f60dd	206	srv->flags.shutdown = 1; /* request it to shut itself down */
74addf6c	207	if (srv->flags.busy) {
	208	debug(34, 3) ("helperShutdown: %s #%d is BUSY.\n",
	209	hlp->id_name, srv->index + 1);
74addf6c	210	continue;
	211	}
	212	if (srv->flags.closing) {
	213	debug(34, 3) ("helperShutdown: %s #%d is CLOSING.\n",
	214	hlp->id_name, srv->index + 1);
	215	continue;
	216	}
74addf6c	217	srv->flags.closing = 1;
3cdb7cd0	218	comm_close(srv->wfd);
3cdb7cd0	219	srv->wfd = -1;
74addf6c	220	}
	221	}
	222
1f5f60dd	223	helper *
	224	helperCreate(const char *name)
	225	{
28c60158	226	helper *hlp;
28c60158	227	hlp = CBDATA_ALLOC(helper, NULL);
1f5f60dd	228	hlp->id_name = name;
	229	return hlp;
	230	}
	231
	232	void
	233	helperFree(helper * hlp)
	234	{
	235	/* note, don't free hlp->name, it probably points to static memory */
fe73896c	236	if (hlp->queue.head)
	237	debug(29, 0) ("WARNING: freeing %s helper with %d requests queued\n",
	238	hlp->id_name, hlp->stats.queue_size);
1f5f60dd	239	cbdataFree(hlp);
	240	}
	241
74addf6c	242	/* ====================================================================== */
	243	/* LOCAL FUNCTIONS */
	244	/* ====================================================================== */
	245
	246	static void
1f5f60dd	247	helperServerFree(int fd, void *data)
74addf6c	248	{
	249	helper_server *srv = data;
	250	helper *hlp = srv->parent;
ac750329	251	helper_request *r;
74addf6c	252	assert(srv->rfd == fd);
74addf6c	253	if (srv->buf) {
db1cd23c	254	memFree(srv->buf, MEM_8K_BUF);
74addf6c	255	srv->buf = NULL;
74addf6c	256	}
ac750329	257	if ((r = srv->request)) {
	258	if (cbdataValid(r->data))
	259	r->callback(r->data, srv->buf);
	260	helperRequestFree(r);
63758217	261	srv->request = NULL;
ac750329	262	}
3cdb7cd0	263	if (srv->wfd != srv->rfd && srv->wfd != -1)
74addf6c	264	comm_close(srv->wfd);
74addf6c	265	dlinkDelete(&srv->link, &hlp->servers);
74addf6c	266	hlp->n_running--;
74addf6c	267	assert(hlp->n_running >= 0);
1f5f60dd	268	if (!srv->flags.shutdown) {
1f5f60dd	269	debug(34, 0) ("WARNING: %s #%d (FD %d) exited\n",
14e87a44	270	hlp->id_name, srv->index + 1, fd);
1f5f60dd	271	if (hlp->n_running < hlp->n_to_start / 2)
14e87a44	272	fatalf("Too few %s processes are running", hlp->id_name);
14e87a44	273	}
1f5f60dd	274	cbdataUnlock(srv->parent);
14e87a44	275	cbdataFree(srv);
74addf6c	276	}
	277
	278	static void
	279	helperHandleRead(int fd, void *data)
	280	{
	281	int len;
	282	char *t = NULL;
	283	helper_server *srv = data;
	284	helper_request *r;
	285	helper *hlp = srv->parent;
	286	assert(fd == srv->rfd);
	287	assert(cbdataValid(data));
83704487	288	statCounter.syscalls.sock.reads++;
74addf6c	289	len = read(fd, srv->buf + srv->offset, srv->buf_sz - srv->offset);
	290	fd_bytes(fd, len, FD_READ);
	291	debug(29, 5) ("helperHandleRead: %d bytes from %s #%d.\n",
	292	len, hlp->id_name, srv->index + 1);
	293	if (len <= 0) {
	294	if (len < 0)
	295	debug(50, 1) ("helperHandleRead: FD %d read: %s\n", fd, xstrerror());
	296	comm_close(fd);
	297	return;
	298	}
	299	srv->offset += len;
	300	srv->buf[srv->offset] = '\0';
	301	r = srv->request;
	302	if (r == NULL) {
	303	/* someone spoke without being spoken to */
	304	debug(29, 1) ("helperHandleRead: unexpected read from %s #%d, %d bytes\n",
	305	hlp->id_name, srv->index + 1, len);
	306	srv->offset = 0;
	307	} else if ((t = strchr(srv->buf, '\n'))) {
	308	/* end of reply found */
	309	debug(29, 3) ("helperHandleRead: end of reply found\n");
	310	*t = '\0';
	311	if (cbdataValid(r->data))
	312	r->callback(r->data, srv->buf);
	313	srv->flags.busy = 0;
	314	srv->offset = 0;
	315	helperRequestFree(r);
63758217	316	srv->request = NULL;
74addf6c	317	hlp->stats.replies++;
	318	hlp->stats.avg_svc_time =
	319	intAverage(hlp->stats.avg_svc_time,
	320	tvSubMsec(srv->dispatch_time, current_time),
	321	hlp->stats.replies, REDIRECT_AV_FACTOR);
3cdb7cd0	322	if (srv->flags.shutdown) {
74addf6c	323	comm_close(srv->wfd);
3cdb7cd0	324	srv->wfd = -1;
3cdb7cd0	325	} else
c68e9c6b	326	helperKickQueue(hlp);
74addf6c	327	} else {
	328	commSetSelect(srv->rfd, COMM_SELECT_READ, helperHandleRead, srv, 0);
	329	}
74addf6c	330	}
	331
	332	static void
	333	Enqueue(helper * hlp, helper_request * r)
	334	{
c68e9c6b	335	dlink_node *link = memAllocate(MEM_DLINK_NODE);
74addf6c	336	dlinkAddTail(r, link, &hlp->queue);
	337	hlp->stats.queue_size++;
	338	if (hlp->stats.queue_size < hlp->n_running)
	339	return;
	340	if (squid_curtime - hlp->last_queue_warn < 600)
	341	return;
fe73896c	342	if (shutting_down \|\| reconfiguring)
fe73896c	343	return;
74addf6c	344	hlp->last_queue_warn = squid_curtime;
	345	debug(14, 0) ("WARNING: All %s processes are busy.\n", hlp->id_name);
	346	debug(14, 0) ("WARNING: %d pending requests queued\n", hlp->stats.queue_size);
	347	if (hlp->stats.queue_size > hlp->n_running * 2)
	348	fatalf("Too many queued %s requests", hlp->id_name);
	349	debug(14, 1) ("Consider increasing the number of %s processes in your config file.\n", hlp->id_name);
	350	}
	351
	352	static helper_request *
	353	Dequeue(helper * hlp)
	354	{
	355	dlink_node *link;
	356	helper_request *r = NULL;
	357	if ((link = hlp->queue.head)) {
	358	r = link->data;
	359	dlinkDelete(link, &hlp->queue);
db1cd23c	360	memFree(link, MEM_DLINK_NODE);
74addf6c	361	hlp->stats.queue_size--;
	362	}
	363	return r;
	364	}
	365
	366	static helper_server *
	367	GetFirstAvailable(helper * hlp)
	368	{
	369	dlink_node *n;
	370	helper_server *srv = NULL;
fe73896c	371	if (hlp->n_running == 0)
fe73896c	372	return NULL;
74addf6c	373	for (n = hlp->servers.head; n != NULL; n = n->next) {
	374	srv = n->data;
	375	if (srv->flags.busy)
	376	continue;
	377	if (!srv->flags.alive)
	378	continue;
	379	return srv;
	380	}
	381	return NULL;
	382	}
	383
	384	static void
	385	helperDispatch(helper_server * srv, helper_request * r)
	386	{
	387	helper *hlp = srv->parent;
	388	if (!cbdataValid(r->data)) {
	389	debug(29, 1) ("helperDispatch: invalid callback data\n");
	390	helperRequestFree(r);
	391	return;
	392	}
	393	assert(!srv->flags.busy);
	394	srv->flags.busy = 1;
	395	srv->request = r;
	396	srv->dispatch_time = current_time;
	397	comm_write(srv->wfd,
	398	r->buf,
	399	strlen(r->buf),
	400	NULL, /* Handler */
	401	NULL, /* Handler-data */
	402	NULL); /* free */
	403	commSetSelect(srv->rfd,
	404	COMM_SELECT_READ,
	405	helperHandleRead,
	406	srv, 0);
	407	debug(29, 5) ("helperDispatch: Request sent to %s #%d, %d bytes\n",
	408	hlp->id_name, srv->index + 1, strlen(r->buf));
	409	srv->stats.uses++;
	410	hlp->stats.requests++;
	411	}
	412
	413	static void
	414	helperKickQueue(helper * hlp)
	415	{
	416	helper_request *r;
	417	helper_server *srv;
	418	while ((srv = GetFirstAvailable(hlp)) && (r = Dequeue(hlp)))
	419	helperDispatch(srv, r);
	420	}
	421
	422	static void
	423	helperRequestFree(helper_request * r)
	424	{
	425	cbdataUnlock(r->data);
	426	xfree(r->buf);
db1cd23c	427	memFree(r, MEM_HELPER_REQUEST);
74addf6c	428	}