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


/*
 * $Id: store_client.cc,v 1.49 1998/12/05 00:54:44 wessels Exp $
 *
 * DEBUG: section 20    Storage Manager Client-Side Interface
 * AUTHOR: Duane Wessels
 *
 * 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
 *  Duane Wessels and the University of California San Diego.  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"

/*
 * NOTE: 'Header' refers to the swapfile metadata header.
 *       'Body' refers to the swapfile body, which is the full
 *        HTTP reply (including HTTP headers and body).
 */
static DRCB storeClientReadBody;
static DRCB storeClientReadHeader;
static SIH storeClientFileOpened;
static void storeClientCopy2(StoreEntry * e, store_client * sc);
static void storeClientFileRead(store_client * sc);
static EVH storeClientCopyEvent;
static store_client_t storeClientType(StoreEntry *);

/* check if there is any client waiting for this object at all */
/* return 1 if there is at least one client */
int
storeClientWaiting(const StoreEntry * e)
{
    MemObject *mem = e->mem_obj;
    store_client *sc;
    for (sc = mem->clients; sc; sc = sc->next) {
	if (sc->callback_data != NULL)
	    return 1;
    }
    return 0;
}

store_client *
storeClientListSearch(const MemObject * mem, void *data)
{
    store_client *sc;
    for (sc = mem->clients; sc; sc = sc->next) {
	if (sc->callback_data == data)
	    break;
    }
    return sc;
}

static store_client_t
storeClientType(StoreEntry * e)
{
    MemObject *mem = e->mem_obj;
    if (mem->inmem_lo)
	return STORE_DISK_CLIENT;
    if (e->store_status == STORE_ABORTED) {
	/* I don't think we should be adding clients to aborted entries */
	debug(20, 1) ("storeClientType: adding to STORE_ABORTED entry\n");
	return STORE_MEM_CLIENT;
    }
    if (e->store_status == STORE_OK) {
	if (mem->inmem_lo == 0 && mem->inmem_hi > 0)
	    return STORE_MEM_CLIENT;
	else
	    return STORE_DISK_CLIENT;
    }
    /* here and past, entry is STORE_PENDING */
    /*
     * If this is the first client, let it be the mem client
     */
    else if (mem->nclients == 1)
	return STORE_MEM_CLIENT;
    /*
     * otherwise, make subsequent clients read from disk so they
     * can not delay the first, and vice-versa.
     */
    else
	return STORE_DISK_CLIENT;
}

/* add client with fd to client list */
void
storeClientListAdd(StoreEntry * e, void *data)
{
    MemObject *mem = e->mem_obj;
    store_client **T;
    store_client *sc;
    assert(mem);
    if (storeClientListSearch(mem, data) != NULL)
	return;
    mem->nclients++;
    sc = memAllocate(MEM_STORE_CLIENT);
    cbdataAdd(sc, memFree, MEM_STORE_CLIENT);	/* sc is callback_data for file_read */
    sc->callback_data = data;
    sc->seen_offset = 0;
    sc->copy_offset = 0;
    sc->swapin_fd = -1;
    sc->flags.disk_io_pending = 0;
    sc->entry = e;
    sc->type = storeClientType(e);
    if (sc->type == STORE_DISK_CLIENT)
	/* assert we'll be able to get the data we want */
	/* maybe we should open swapin_fd here */
	assert(e->swap_file_number > -1 || storeSwapOutAble(e));
    for (T = &mem->clients; *T; T = &(*T)->next);
    *T = sc;
}

static void
storeClientCopyEvent(void *data)
{
    store_client *sc = data;
    debug(20, 3) ("storeClientCopyEvent: Running\n");
    sc->flags.copy_event_pending = 0;
    if (!sc->callback)
	return;
    storeClientCopy2(sc->entry, sc);
}

/* copy bytes requested by the client */
void
storeClientCopy(StoreEntry * e,
    off_t seen_offset,
    off_t copy_offset,
    size_t size,
    char *buf,
    STCB * callback,
    void *data)
{
    store_client *sc;
    assert(e->store_status != STORE_ABORTED);
    debug(20, 3) ("storeClientCopy: %s, seen %d, want %d, size %d, cb %p, cbdata %p\n",
	storeKeyText(e->key),
	(int) seen_offset,
	(int) copy_offset,
	(int) size,
	callback,
	data);
    sc = storeClientListSearch(e->mem_obj, data);
    assert(sc != NULL);
    assert(sc->callback == NULL);
    sc->copy_offset = copy_offset;
    sc->seen_offset = seen_offset;
    sc->callback = callback;
    sc->copy_buf = buf;
    sc->copy_size = size;
    sc->copy_offset = copy_offset;
    storeClientCopy2(e, sc);
}

/*
 * This function is used below to decide if we have any more data to
 * send to the client.  If the store_status is STORE_PENDING, then we
 * do have more data to send.  If its STORE_OK or STORE_ABORTED, then
 * we continue checking.  If the object length is negative, then we
 * don't know the real length and must open the swap file to find out.
 * If the length is >= 0, then we compare it to the requested copy
 * offset.
 */
static int
storeClientNoMoreToSend(StoreEntry * e, store_client * sc)
{
    ssize_t len;
    if (e->store_status == STORE_PENDING)
	return 0;
    if ((len = objectLen(e)) < 0)
	return 0;
    if (sc->copy_offset < len)
	return 0;
    return 1;
}

static void
storeClientCopy2(StoreEntry * e, store_client * sc)
{
    STCB *callback = sc->callback;
    MemObject *mem = e->mem_obj;
    size_t sz;
    if (sc->flags.copy_event_pending)
	return;
    if (EBIT_TEST(e->flags, ENTRY_FWD_HDR_WAIT)) {
	debug(20, 5) ("storeClientCopy2: returning because ENTRY_FWD_HDR_WAIT set\n");
	return;
    }
    if (sc->flags.store_copying) {
	sc->flags.copy_event_pending = 1;
	debug(20, 3) ("storeClientCopy2: Queueing storeClientCopyEvent()\n");
	eventAdd("storeClientCopyEvent", storeClientCopyEvent, sc, 0.0, 0);
	return;
    }
    cbdataLock(sc);		/* ick, prevent sc from getting freed */
    sc->flags.store_copying = 1;
    debug(20, 3) ("storeClientCopy2: %s\n", storeKeyText(e->key));
    assert(callback != NULL);
    /*
     * We used to check for STORE_ABORTED here.  But there were some
     * problems.  For example, we might have a slow client (or two) and
     * the server-side is reading far ahead and swapping to disk.  Even
     * if the server-side aborts, we want to give the client(s)
     * everything we got before the abort condition occurred.
     */
    if (storeClientNoMoreToSend(e, sc)) {
	/* There is no more to send! */
#if USE_ASYNC_IO
	if (sc->flags.disk_io_pending) {
	    if (sc->swapin_fd >= 0)
		aioCancel(sc->swapin_fd, NULL);
	    else
		aioCancel(-1, sc);
	}
#endif
	sc->flags.disk_io_pending = 0;
	sc->callback = NULL;
	callback(sc->callback_data, sc->copy_buf, 0);
    } else if (e->store_status == STORE_PENDING && sc->seen_offset >= mem->inmem_hi) {
	/* client has already seen this, wait for more */
	debug(20, 3) ("storeClientCopy2: Waiting for more\n");
    } else if (sc->copy_offset >= mem->inmem_lo && sc->copy_offset < mem->inmem_hi) {
	/* What the client wants is in memory */
	debug(20, 3) ("storeClientCopy2: Copying from memory\n");
	sz = stmemCopy(&mem->data_hdr, sc->copy_offset, sc->copy_buf, sc->copy_size);
#if USE_ASYNC_IO
	if (sc->flags.disk_io_pending) {
	    if (sc->swapin_fd >= 0)
		aioCancel(sc->swapin_fd, NULL);
	    else
		aioCancel(-1, sc);
	}
#endif
	sc->flags.disk_io_pending = 0;
	sc->callback = NULL;
	callback(sc->callback_data, sc->copy_buf, sz);
    } else if (sc->swapin_fd < 0) {
	debug(20, 3) ("storeClientCopy2: Need to open swap in file\n");
	assert(sc->type == STORE_DISK_CLIENT);
	/* gotta open the swapin file */
	/* assert(sc->copy_offset == 0); */
	if (!sc->flags.disk_io_pending) {
	    sc->flags.disk_io_pending = 1;
	    storeSwapInStart(e, storeClientFileOpened, sc);
	} else {
	    debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
	}
    } else {
	debug(20, 3) ("storeClientCopy: reading from disk FD %d\n",
	    sc->swapin_fd);
	assert(sc->type == STORE_DISK_CLIENT);
	if (!sc->flags.disk_io_pending) {
	    storeClientFileRead(sc);
	} else {
	    debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
	}
    }
    sc->flags.store_copying = 0;
    cbdataUnlock(sc);		/* ick, allow sc to be freed */
}

static void
storeClientFileOpened(int fd, void *data)
{
    store_client *sc = data;
    STCB *callback = sc->callback;
    if (fd < 0) {
	debug(20, 3) ("storeClientFileOpened: failed\n");
	sc->flags.disk_io_pending = 0;
	sc->callback = NULL;
	callback(sc->callback_data, sc->copy_buf, -1);
	return;
    }
    sc->swapin_fd = fd;
    storeClientFileRead(sc);
}

static void
storeClientFileRead(store_client * sc)
{
    MemObject *mem = sc->entry->mem_obj;
    assert(sc->callback != NULL);
#ifdef OPTIMISTIC_IO
    sc->flags.disk_io_pending = 1;
#endif
    if (mem->swap_hdr_sz == 0) {
	file_read(sc->swapin_fd,
	    sc->copy_buf,
	    sc->copy_size,
	    0,
	    storeClientReadHeader,
	    sc);
    } else {
	if (sc->entry->swap_status == SWAPOUT_WRITING)
	    assert(mem->swapout.done_offset > sc->copy_offset + mem->swap_hdr_sz);
	file_read(sc->swapin_fd,
	    sc->copy_buf,
	    sc->copy_size,
	    sc->copy_offset + mem->swap_hdr_sz,
	    storeClientReadBody,
	    sc);
    }
#ifndef OPTIMISTIC_IO
    sc->flags.disk_io_pending = 1;
#endif
}

static void
storeClientReadBody(int fd, const char *buf, int len, int flagnotused, void *data)
{
    store_client *sc = data;
    MemObject *mem = sc->entry->mem_obj;
    STCB *callback = sc->callback;
    assert(sc->flags.disk_io_pending);
    sc->flags.disk_io_pending = 0;
    assert(sc->callback != NULL);
    debug(20, 3) ("storeClientReadBody: FD %d, len %d\n", fd, len);
    if (sc->copy_offset == 0 && len > 0 && mem->reply->sline.status == 0)
	httpReplyParse(mem->reply, sc->copy_buf);
    sc->callback = NULL;
    callback(sc->callback_data, sc->copy_buf, len);
}

static void
storeClientReadHeader(int fd, const char *buf, int len, int flagnotused, void *data)
{
    store_client *sc = data;
    StoreEntry *e = sc->entry;
    MemObject *mem = e->mem_obj;
    STCB *callback = sc->callback;
    int swap_hdr_sz = 0;
    size_t body_sz;
    size_t copy_sz;
    tlv *tlv_list;
    assert(sc->flags.disk_io_pending);
    sc->flags.disk_io_pending = 0;
    assert(sc->callback != NULL);
    debug(20, 3) ("storeClientReadHeader: FD %d, len %d\n", fd, len);
    if (len < 0) {
	debug(20, 3) ("storeClientReadHeader: FD %d: %s\n", fd, xstrerror());
	sc->callback = NULL;
	callback(sc->callback_data, sc->copy_buf, len);
	return;
    }
    tlv_list = storeSwapMetaUnpack(buf, &swap_hdr_sz);
    if (tlv_list == NULL) {
	debug(20, 1) ("storeClientReadHeader: failed to unpack meta data\n");
	sc->callback = NULL;
	callback(sc->callback_data, sc->copy_buf, -1);
	return;
    }
    /*
     * XXX Here we should check the meta data and make sure we got
     * the right object.
     */
    storeSwapTLVFree(tlv_list);
    mem->swap_hdr_sz = swap_hdr_sz;
    mem->object_sz = e->swap_file_sz - swap_hdr_sz;
    /*
     * If our last read got some data the client wants, then give
     * it to them, otherwise schedule another read.
     */
    body_sz = len - swap_hdr_sz;
    if (sc->copy_offset < body_sz) {
	/*
	 * we have (part of) what they want
	 */
	copy_sz = XMIN(sc->copy_size, body_sz);
	debug(20, 3) ("storeClientReadHeader: copying %d bytes of body\n",
	    copy_sz);
	xmemmove(sc->copy_buf, sc->copy_buf + swap_hdr_sz, copy_sz);
	if (sc->copy_offset == 0 && len > 0 && mem->reply->sline.status == 0)
	    httpReplyParse(mem->reply, sc->copy_buf);
	sc->callback = NULL;
	callback(sc->callback_data, sc->copy_buf, copy_sz);
	return;
    }
    /*
     * we don't have what the client wants, but at least we now
     * know the swap header size.
     */
    storeClientFileRead(sc);
}

int
storeClientCopyPending(StoreEntry * e, void *data)
{
    /* return 1 if there is a callback registered for this client */
    store_client *sc = storeClientListSearch(e->mem_obj, data);
    if (sc == NULL)
	return 0;
    if (sc->callback == NULL)
	return 0;
    return 1;
}

int
storeUnregister(StoreEntry * e, void *data)
{
    MemObject *mem = e->mem_obj;
    store_client *sc;
    store_client **S;
    STCB *callback;
    if (mem == NULL)
	return 0;
    debug(20, 3) ("storeUnregister: called for '%s'\n", storeKeyText(e->key));
    for (S = &mem->clients; (sc = *S) != NULL; S = &(*S)->next) {
	if (sc->callback_data == data)
	    break;
    }
    if (sc == NULL)
	return 0;
    if (sc == mem->clients) {
	/*
	 * If we are unregistering the _first_ client for this
	 * entry, then we have to reset the client FD to -1.
	 */
	mem->fd = -1;
    }
    *S = sc->next;
    mem->nclients--;
    sc->flags.disk_io_pending = 0;
    if (e->store_status == STORE_OK && e->swap_status != SWAPOUT_DONE)
	storeCheckSwapOut(e);
    if (sc->swapin_fd > -1) {
	commSetSelect(sc->swapin_fd, COMM_SELECT_READ, NULL, NULL, 0);
	file_close(sc->swapin_fd);
	/* XXX this probably leaks file_read handler structures */
    }
#if USE_ASYNC_IO
    else
	aioCancel(-1, sc);
#endif
    if ((callback = sc->callback) != NULL) {
	/* callback with ssize = -1 to indicate unexpected termination */
	debug(20, 3) ("storeUnregister: store_client for %s has a callback\n",
	    mem->url);
	sc->callback = NULL;
	callback(sc->callback_data, sc->copy_buf, -1);
    }
    cbdataFree(sc);
    return 1;
}

off_t
storeLowestMemReaderOffset(const StoreEntry * entry)
{
    const MemObject *mem = entry->mem_obj;
    off_t lowest = mem->inmem_hi;
    store_client *sc;
    store_client *nx = NULL;
    for (sc = mem->clients; sc; sc = nx) {
	nx = sc->next;
	if (sc->callback_data == NULL)	/* open slot */
	    continue;
	if (sc->type != STORE_MEM_CLIENT)
	    continue;
	if (sc->copy_offset < lowest)
	    lowest = sc->copy_offset;
    }
    return lowest;
}

/* Call handlers waiting for  data to be appended to E. */
void
InvokeHandlers(StoreEntry * e)
{
    int i = 0;
    MemObject *mem = e->mem_obj;
    store_client *sc;
    store_client *nx = NULL;
    assert(mem->clients != NULL || mem->nclients == 0);
    debug(20, 3) ("InvokeHandlers: %s\n", storeKeyText(e->key));
    /* walk the entire list looking for valid callbacks */
    for (sc = mem->clients; sc; sc = nx) {
	nx = sc->next;
	debug(20, 3) ("InvokeHandlers: checking client #%d\n", i++);
	if (sc->callback_data == NULL)
	    continue;
	if (sc->callback == NULL)
	    continue;
	storeClientCopy2(e, sc);
    }
}

int
storePendingNClients(const StoreEntry * e)
{
    MemObject *mem = e->mem_obj;
    int npend = NULL == mem ? 0 : mem->nclients;
    debug(20, 3) ("storePendingNClients: returning %d\n", npend);
    return npend;
}
Commit	Line	Data
9cef6668	1
9cef6668	2	/*
db1cd23c	3	* $Id: store_client.cc,v 1.49 1998/12/05 00:54:44 wessels Exp $
9cef6668	4	*
	5	* DEBUG: section 20 Storage Manager Client-Side Interface
	6	* AUTHOR: Duane Wessels
	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	* Duane Wessels and the University of California San Diego. Please
	16	* see the COPYRIGHT file for full details. Squid incorporates
	17	* software developed and/or copyrighted by other sources. Please see
	18	* the 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
f09f5b26	36	#include "squid.h"
f09f5b26	37
e3ef2b09	38	/*
	39	* NOTE: 'Header' refers to the swapfile metadata header.
	40	* 'Body' refers to the swapfile body, which is the full
	41	* HTTP reply (including HTTP headers and body).
	42	*/
	43	static DRCB storeClientReadBody;
	44	static DRCB storeClientReadHeader;
	45	static SIH storeClientFileOpened;
f09f5b26	46	static void storeClientCopy2(StoreEntry * e, store_client * sc);
e3ef2b09	47	static void storeClientFileRead(store_client * sc);
d6f51e3c	48	static EVH storeClientCopyEvent;
7405a782	49	static store_client_t storeClientType(StoreEntry *);
f09f5b26	50
	51	/* check if there is any client waiting for this object at all */
	52	/* return 1 if there is at least one client */
	53	int
	54	storeClientWaiting(const StoreEntry * e)
	55	{
	56	MemObject *mem = e->mem_obj;
	57	store_client *sc;
	58	for (sc = mem->clients; sc; sc = sc->next) {
	59	if (sc->callback_data != NULL)
	60	return 1;
	61	}
	62	return 0;
	63	}
	64
	65	store_client *
	66	storeClientListSearch(const MemObject * mem, void *data)
	67	{
	68	store_client *sc;
	69	for (sc = mem->clients; sc; sc = sc->next) {
	70	if (sc->callback_data == data)
	71	break;
	72	}
	73	return sc;
	74	}
	75
7405a782	76	static store_client_t
135171fe	77	storeClientType(StoreEntry * e)
7405a782	78	{
7405a782	79	MemObject *mem = e->mem_obj;
7405a782	80	if (mem->inmem_lo)
7405a782	81	return STORE_DISK_CLIENT;
6a888d16	82	if (e->store_status == STORE_ABORTED) {
6a888d16	83	/* I don't think we should be adding clients to aborted entries */
135171fe	84	debug(20, 1) ("storeClientType: adding to STORE_ABORTED entry\n");
7405a782	85	return STORE_MEM_CLIENT;
6a888d16	86	}
	87	if (e->store_status == STORE_OK) {
	88	if (mem->inmem_lo == 0 && mem->inmem_hi > 0)
	89	return STORE_MEM_CLIENT;
	90	else
	91	return STORE_DISK_CLIENT;
	92	}
	93	/* here and past, entry is STORE_PENDING */
7405a782	94	/*
	95	* If this is the first client, let it be the mem client
	96	*/
6a888d16	97	else if (mem->nclients == 1)
6a888d16	98	return STORE_MEM_CLIENT;
7405a782	99	/*
	100	* otherwise, make subsequent clients read from disk so they
	101	* can not delay the first, and vice-versa.
	102	*/
	103	else
6a888d16	104	return STORE_DISK_CLIENT;
7405a782	105	}
7405a782	106
f09f5b26	107	/* add client with fd to client list */
	108	void
	109	storeClientListAdd(StoreEntry * e, void *data)
	110	{
	111	MemObject *mem = e->mem_obj;
	112	store_client **T;
	113	store_client *sc;
	114	assert(mem);
	115	if (storeClientListSearch(mem, data) != NULL)
	116	return;
	117	mem->nclients++;
7021844c	118	sc = memAllocate(MEM_STORE_CLIENT);
db1cd23c	119	cbdataAdd(sc, memFree, MEM_STORE_CLIENT); /* sc is callback_data for file_read */
f09f5b26	120	sc->callback_data = data;
	121	sc->seen_offset = 0;
	122	sc->copy_offset = 0;
	123	sc->swapin_fd = -1;
f115fadd	124	sc->flags.disk_io_pending = 0;
07304bf9	125	sc->entry = e;
7405a782	126	sc->type = storeClientType(e);
7405a782	127	if (sc->type == STORE_DISK_CLIENT)
698a99c2	128	/* assert we'll be able to get the data we want */
698a99c2	129	/* maybe we should open swapin_fd here */
5f25e839	130	assert(e->swap_file_number > -1 \|\| storeSwapOutAble(e));
f09f5b26	131	for (T = &mem->clients; T; T = &(T)->next);
	132	*T = sc;
	133	}
	134
f115fadd	135	static void
	136	storeClientCopyEvent(void *data)
	137	{
	138	store_client *sc = data;
fe4a33ac	139	debug(20, 3) ("storeClientCopyEvent: Running\n");
67fd69de	140	sc->flags.copy_event_pending = 0;
a2899918	141	if (!sc->callback)
a2899918	142	return;
f115fadd	143	storeClientCopy2(sc->entry, sc);
f115fadd	144	}
f115fadd	145
f09f5b26	146	/* copy bytes requested by the client */
	147	void
	148	storeClientCopy(StoreEntry * e,
	149	off_t seen_offset,
	150	off_t copy_offset,
	151	size_t size,
	152	char *buf,
	153	STCB * callback,
	154	void *data)
	155	{
	156	store_client *sc;
f09f5b26	157	assert(e->store_status != STORE_ABORTED);
f09f5b26	158	debug(20, 3) ("storeClientCopy: %s, seen %d, want %d, size %d, cb %p, cbdata %p\n",
	159	storeKeyText(e->key),
	160	(int) seen_offset,
	161	(int) copy_offset,
	162	(int) size,
	163	callback,
	164	data);
	165	sc = storeClientListSearch(e->mem_obj, data);
	166	assert(sc != NULL);
	167	assert(sc->callback == NULL);
	168	sc->copy_offset = copy_offset;
	169	sc->seen_offset = seen_offset;
	170	sc->callback = callback;
	171	sc->copy_buf = buf;
	172	sc->copy_size = size;
	173	sc->copy_offset = copy_offset;
67fd69de	174	storeClientCopy2(e, sc);
f09f5b26	175	}
f09f5b26	176
07304bf9	177	/*
07304bf9	178	* This function is used below to decide if we have any more data to
0bb129ee	179	* send to the client. If the store_status is STORE_PENDING, then we
	180	* do have more data to send. If its STORE_OK or STORE_ABORTED, then
	181	* we continue checking. If the object length is negative, then we
	182	* don't know the real length and must open the swap file to find out.
	183	* If the length is >= 0, then we compare it to the requested copy
	184	* offset.
07304bf9	185	*/
	186	static int
	187	storeClientNoMoreToSend(StoreEntry * e, store_client * sc)
	188	{
	189	ssize_t len;
0bb129ee	190	if (e->store_status == STORE_PENDING)
07304bf9	191	return 0;
	192	if ((len = objectLen(e)) < 0)
	193	return 0;
	194	if (sc->copy_offset < len)
	195	return 0;
	196	return 1;
	197	}
	198
f09f5b26	199	static void
	200	storeClientCopy2(StoreEntry * e, store_client * sc)
	201	{
	202	STCB *callback = sc->callback;
	203	MemObject *mem = e->mem_obj;
	204	size_t sz;
67fd69de	205	if (sc->flags.copy_event_pending)
67fd69de	206	return;
db1cd23c	207	if (EBIT_TEST(e->flags, ENTRY_FWD_HDR_WAIT)) {
db1cd23c	208	debug(20, 5) ("storeClientCopy2: returning because ENTRY_FWD_HDR_WAIT set\n");
7e3e1d01	209	return;
db1cd23c	210	}
67fd69de	211	if (sc->flags.store_copying) {
67fd69de	212	sc->flags.copy_event_pending = 1;
3d36b8ab	213	debug(20, 3) ("storeClientCopy2: Queueing storeClientCopyEvent()\n");
c43f5247	214	eventAdd("storeClientCopyEvent", storeClientCopyEvent, sc, 0.0, 0);
67fd69de	215	return;
67fd69de	216	}
a455c097	217	cbdataLock(sc); /* ick, prevent sc from getting freed */
67fd69de	218	sc->flags.store_copying = 1;
f09f5b26	219	debug(20, 3) ("storeClientCopy2: %s\n", storeKeyText(e->key));
f09f5b26	220	assert(callback != NULL);
0bb129ee	221	/*
	222	* We used to check for STORE_ABORTED here. But there were some
	223	* problems. For example, we might have a slow client (or two) and
	224	* the server-side is reading far ahead and swapping to disk. Even
	225	* if the server-side aborts, we want to give the client(s)
	226	* everything we got before the abort condition occurred.
	227	*/
	228	if (storeClientNoMoreToSend(e, sc)) {
f09f5b26	229	/* There is no more to send! */
f09f5b26	230	#if USE_ASYNC_IO
27002b34	231	if (sc->flags.disk_io_pending) {
f09f5b26	232	if (sc->swapin_fd >= 0)
	233	aioCancel(sc->swapin_fd, NULL);
	234	else
	235	aioCancel(-1, sc);
	236	}
	237	#endif
f115fadd	238	sc->flags.disk_io_pending = 0;
f09f5b26	239	sc->callback = NULL;
f09f5b26	240	callback(sc->callback_data, sc->copy_buf, 0);
2a503f87	241	} else if (e->store_status == STORE_PENDING && sc->seen_offset >= mem->inmem_hi) {
f09f5b26	242	/* client has already seen this, wait for more */
f09f5b26	243	debug(20, 3) ("storeClientCopy2: Waiting for more\n");
3450c822	244	} else if (sc->copy_offset >= mem->inmem_lo && sc->copy_offset < mem->inmem_hi) {
f09f5b26	245	/* What the client wants is in memory */
f09f5b26	246	debug(20, 3) ("storeClientCopy2: Copying from memory\n");
18fe65d0	247	sz = stmemCopy(&mem->data_hdr, sc->copy_offset, sc->copy_buf, sc->copy_size);
f09f5b26	248	#if USE_ASYNC_IO
27002b34	249	if (sc->flags.disk_io_pending) {
f09f5b26	250	if (sc->swapin_fd >= 0)
	251	aioCancel(sc->swapin_fd, NULL);
	252	else
	253	aioCancel(-1, sc);
	254	}
	255	#endif
f115fadd	256	sc->flags.disk_io_pending = 0;
f09f5b26	257	sc->callback = NULL;
	258	callback(sc->callback_data, sc->copy_buf, sz);
	259	} else if (sc->swapin_fd < 0) {
	260	debug(20, 3) ("storeClientCopy2: Need to open swap in file\n");
	261	assert(sc->type == STORE_DISK_CLIENT);
	262	/* gotta open the swapin file */
	263	/* assert(sc->copy_offset == 0); */
27002b34	264	if (!sc->flags.disk_io_pending) {
f115fadd	265	sc->flags.disk_io_pending = 1;
e3ef2b09	266	storeSwapInStart(e, storeClientFileOpened, sc);
f09f5b26	267	} else {
	268	debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
	269	}
	270	} else {
	271	debug(20, 3) ("storeClientCopy: reading from disk FD %d\n",
	272	sc->swapin_fd);
	273	assert(sc->type == STORE_DISK_CLIENT);
27002b34	274	if (!sc->flags.disk_io_pending) {
e3ef2b09	275	storeClientFileRead(sc);
f09f5b26	276	} else {
	277	debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
	278	}
	279	}
67fd69de	280	sc->flags.store_copying = 0;
a455c097	281	cbdataUnlock(sc); /* ick, allow sc to be freed */
f09f5b26	282	}
	283
	284	static void
e3ef2b09	285	storeClientFileOpened(int fd, void *data)
f09f5b26	286	{
	287	store_client *sc = data;
	288	STCB *callback = sc->callback;
	289	if (fd < 0) {
e3ef2b09	290	debug(20, 3) ("storeClientFileOpened: failed\n");
f115fadd	291	sc->flags.disk_io_pending = 0;
f09f5b26	292	sc->callback = NULL;
	293	callback(sc->callback_data, sc->copy_buf, -1);
	294	return;
	295	}
	296	sc->swapin_fd = fd;
e3ef2b09	297	storeClientFileRead(sc);
f09f5b26	298	}
	299
	300	static void
e3ef2b09	301	storeClientFileRead(store_client * sc)
f09f5b26	302	{
07304bf9	303	MemObject *mem = sc->entry->mem_obj;
f09f5b26	304	assert(sc->callback != NULL);
dee51794	305	#ifdef OPTIMISTIC_IO
9d66d521	306	sc->flags.disk_io_pending = 1;
9d66d521	307	#endif
dee51794	308	if (mem->swap_hdr_sz == 0) {
e3ef2b09	309	file_read(sc->swapin_fd,
d8b9a541	310	sc->copy_buf,
d8b9a541	311	sc->copy_size,
e3ef2b09	312	0,
	313	storeClientReadHeader,
	314	sc);
9d66d521	315	} else {
3157c72f	316	if (sc->entry->swap_status == SWAPOUT_WRITING)
0cdcddb9	317	assert(mem->swapout.done_offset > sc->copy_offset + mem->swap_hdr_sz);
e3ef2b09	318	file_read(sc->swapin_fd,
	319	sc->copy_buf,
	320	sc->copy_size,
	321	sc->copy_offset + mem->swap_hdr_sz,
	322	storeClientReadBody,
	323	sc);
3157c72f	324	}
9d66d521	325	#ifndef OPTIMISTIC_IO
f115fadd	326	sc->flags.disk_io_pending = 1;
9d66d521	327	#endif
f09f5b26	328	}
	329
	330	static void
e3ef2b09	331	storeClientReadBody(int fd, const char buf, int len, int flagnotused, void data)
f09f5b26	332	{
f09f5b26	333	store_client *sc = data;
07304bf9	334	MemObject *mem = sc->entry->mem_obj;
f09f5b26	335	STCB *callback = sc->callback;
27002b34	336	assert(sc->flags.disk_io_pending);
f115fadd	337	sc->flags.disk_io_pending = 0;
f09f5b26	338	assert(sc->callback != NULL);
e3ef2b09	339	debug(20, 3) ("storeClientReadBody: FD %d, len %d\n", fd, len);
cb69b4c7	340	if (sc->copy_offset == 0 && len > 0 && mem->reply->sline.status == 0)
cb69b4c7	341	httpReplyParse(mem->reply, sc->copy_buf);
f09f5b26	342	sc->callback = NULL;
	343	callback(sc->callback_data, sc->copy_buf, len);
	344	}
	345
e3ef2b09	346	static void
	347	storeClientReadHeader(int fd, const char buf, int len, int flagnotused, void data)
	348	{
e3ef2b09	349	store_client *sc = data;
07304bf9	350	StoreEntry *e = sc->entry;
07304bf9	351	MemObject *mem = e->mem_obj;
e3ef2b09	352	STCB *callback = sc->callback;
	353	int swap_hdr_sz = 0;
	354	size_t body_sz;
	355	size_t copy_sz;
	356	tlv *tlv_list;
27002b34	357	assert(sc->flags.disk_io_pending);
f115fadd	358	sc->flags.disk_io_pending = 0;
e3ef2b09	359	assert(sc->callback != NULL);
	360	debug(20, 3) ("storeClientReadHeader: FD %d, len %d\n", fd, len);
	361	if (len < 0) {
25535cbe	362	debug(20, 3) ("storeClientReadHeader: FD %d: %s\n", fd, xstrerror());
25535cbe	363	sc->callback = NULL;
25535cbe	364	callback(sc->callback_data, sc->copy_buf, len);
e3ef2b09	365	return;
	366	}
	367	tlv_list = storeSwapMetaUnpack(buf, &swap_hdr_sz);
	368	if (tlv_list == NULL) {
25535cbe	369	debug(20, 1) ("storeClientReadHeader: failed to unpack meta data\n");
25535cbe	370	sc->callback = NULL;
25535cbe	371	callback(sc->callback_data, sc->copy_buf, -1);
e3ef2b09	372	return;
	373	}
	374	/*
	375	* XXX Here we should check the meta data and make sure we got
	376	* the right object.
	377	*/
07304bf9	378	storeSwapTLVFree(tlv_list);
e3ef2b09	379	mem->swap_hdr_sz = swap_hdr_sz;
07304bf9	380	mem->object_sz = e->swap_file_sz - swap_hdr_sz;
e3ef2b09	381	/*
	382	* If our last read got some data the client wants, then give
	383	* it to them, otherwise schedule another read.
	384	*/
	385	body_sz = len - swap_hdr_sz;
	386	if (sc->copy_offset < body_sz) {
	387	/*
25535cbe	388	* we have (part of) what they want
e3ef2b09	389	*/
e3ef2b09	390	copy_sz = XMIN(sc->copy_size, body_sz);
25535cbe	391	debug(20, 3) ("storeClientReadHeader: copying %d bytes of body\n",
25535cbe	392	copy_sz);
d8b9a541	393	xmemmove(sc->copy_buf, sc->copy_buf + swap_hdr_sz, copy_sz);
cb69b4c7	394	if (sc->copy_offset == 0 && len > 0 && mem->reply->sline.status == 0)
cb69b4c7	395	httpReplyParse(mem->reply, sc->copy_buf);
e3ef2b09	396	sc->callback = NULL;
25535cbe	397	callback(sc->callback_data, sc->copy_buf, copy_sz);
e3ef2b09	398	return;
	399	}
	400	/*
	401	* we don't have what the client wants, but at least we now
	402	* know the swap header size.
	403	*/
e3ef2b09	404	storeClientFileRead(sc);
	405	}
	406
f09f5b26	407	int
	408	storeClientCopyPending(StoreEntry * e, void *data)
	409	{
	410	/* return 1 if there is a callback registered for this client */
	411	store_client *sc = storeClientListSearch(e->mem_obj, data);
	412	if (sc == NULL)
	413	return 0;
	414	if (sc->callback == NULL)
	415	return 0;
	416	return 1;
	417	}
	418
	419	int
	420	storeUnregister(StoreEntry * e, void *data)
	421	{
	422	MemObject *mem = e->mem_obj;
	423	store_client *sc;
	424	store_client **S;
	425	STCB *callback;
	426	if (mem == NULL)
	427	return 0;
	428	debug(20, 3) ("storeUnregister: called for '%s'\n", storeKeyText(e->key));
	429	for (S = &mem->clients; (sc = S) != NULL; S = &(S)->next) {
	430	if (sc->callback_data == data)
	431	break;
	432	}
	433	if (sc == NULL)
	434	return 0;
4c454c5c	435	if (sc == mem->clients) {
	436	/*
	437	* If we are unregistering the _first_ client for this
	438	* entry, then we have to reset the client FD to -1.
	439	*/
	440	mem->fd = -1;
	441	}
f09f5b26	442	*S = sc->next;
f09f5b26	443	mem->nclients--;
f115fadd	444	sc->flags.disk_io_pending = 0;
f09f5b26	445	if (e->store_status == STORE_OK && e->swap_status != SWAPOUT_DONE)
	446	storeCheckSwapOut(e);
	447	if (sc->swapin_fd > -1) {
	448	commSetSelect(sc->swapin_fd, COMM_SELECT_READ, NULL, NULL, 0);
	449	file_close(sc->swapin_fd);
fd8fe6ea	450	/* XXX this probably leaks file_read handler structures */
f09f5b26	451	}
	452	#if USE_ASYNC_IO
	453	else
	454	aioCancel(-1, sc);
	455	#endif
	456	if ((callback = sc->callback) != NULL) {
	457	/* callback with ssize = -1 to indicate unexpected termination */
	458	debug(20, 3) ("storeUnregister: store_client for %s has a callback\n",
	459	mem->url);
	460	sc->callback = NULL;
	461	callback(sc->callback_data, sc->copy_buf, -1);
	462	}
	463	cbdataFree(sc);
	464	return 1;
	465	}
	466
	467	off_t
	468	storeLowestMemReaderOffset(const StoreEntry * entry)
	469	{
	470	const MemObject *mem = entry->mem_obj;
	471	off_t lowest = mem->inmem_hi;
	472	store_client *sc;
	473	store_client *nx = NULL;
	474	for (sc = mem->clients; sc; sc = nx) {
	475	nx = sc->next;
	476	if (sc->callback_data == NULL) /* open slot */
	477	continue;
	478	if (sc->type != STORE_MEM_CLIENT)
	479	continue;
	480	if (sc->copy_offset < lowest)
	481	lowest = sc->copy_offset;
	482	}
	483	return lowest;
	484	}
	485
	486	/* Call handlers waiting for data to be appended to E. */
	487	void
	488	InvokeHandlers(StoreEntry * e)
	489	{
	490	int i = 0;
	491	MemObject *mem = e->mem_obj;
	492	store_client *sc;
	493	store_client *nx = NULL;
	494	assert(mem->clients != NULL \|\| mem->nclients == 0);
	495	debug(20, 3) ("InvokeHandlers: %s\n", storeKeyText(e->key));
	496	/* walk the entire list looking for valid callbacks */
	497	for (sc = mem->clients; sc; sc = nx) {
	498	nx = sc->next;
	499	debug(20, 3) ("InvokeHandlers: checking client #%d\n", i++);
	500	if (sc->callback_data == NULL)
	501	continue;
	502	if (sc->callback == NULL)
	503	continue;
	504	storeClientCopy2(e, sc);
	505	}
	506	}
	507
	508	int
	509	storePendingNClients(const StoreEntry * e)
	510	{
f09f5b26	511	MemObject *mem = e->mem_obj;
36547bcf	512	int npend = NULL == mem ? 0 : mem->nclients;
1afe05c5	513	debug(20, 3) ("storePendingNClients: returning %d\n", npend);
f09f5b26	514	return npend;
f09f5b26	515	}