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

#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);

/* 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;
}

/* 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, 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;
    if (e->store_status == STORE_PENDING && mem->swapout.fd == -1)
	sc->type = STORE_MEM_CLIENT;
    else
	sc->type = STORE_DISK_CLIENT;
    for (T = &mem->clients; *T; T = &(*T)->next);
    *T = sc;
}

static void
storeClientCopyEvent(void *data)
{
    store_client *sc = data;
    int valid = cbdataValid(sc);
    debug(20, 3) ("storeClientCopyEvent: Running\n");
    cbdataUnlock(sc);
    if (!valid)
	return;
    assert(!sc->flags.store_copying);
    sc->flags.store_copying = 1;
    storeClientCopy2(sc->entry, sc);
    sc->flags.store_copying = 0;
}

/* 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;
    if (sc->flags.store_copying) {
	cbdataLock(sc);
	debug(20, 3) ("storeClientCopy: Queueing storeClientCopyEvent()\n");
	eventAdd("storeClientCopyEvent", storeClientCopyEvent, sc, 0);
    } else {
	sc->flags.store_copying = 1;
	storeClientCopy2(e, sc);
	sc->flags.store_copying = 0;
    }
}

/*
 * This function is used below to decide if we have any more data to
 * send to the client.  If the store_status is STORE_ABORTED, that case
 * should be handled before this function gets called.  If the
 * store_status is STORE_PENDING, then we do have more data to send.
 * If its STORE_OK, 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 lenght 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_OK)
	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;
    static int loopdetect = 0;
    assert(++loopdetect < 10);
    debug(20, 3) ("storeClientCopy2: %s\n", storeKeyText(e->key));
    assert(callback != NULL);
    if (e->store_status == STORE_ABORTED) {
#if USE_ASYNC_IO
	if (sc->flags.disk_io_pending == 1) {
	    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 (storeClientNoMoreToSend(e, sc)) {
	/* There is no more to send! */
#if USE_ASYNC_IO
	if (sc->flags.disk_io_pending == 1) {
	    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 && mem->inmem_lo < mem->inmem_hi) {
	/* What the client wants is in memory */
	debug(20, 3) ("storeClientCopy2: Copying from memory\n");
	sz = stmemCopy(mem->data, sc->copy_offset, sc->copy_buf, sc->copy_size);
#if USE_ASYNC_IO
	if (sc->flags.disk_io_pending == 1) {
	    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 == 0) {
	    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 == 0) {
	    storeClientFileRead(sc);
	} else {
	    debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
	}
    }
    --loopdetect;
}

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);
    if (mem->swap_hdr_sz == 0)
	file_read(sc->swapin_fd,
	    sc->copy_buf,
	    sc->copy_size,
	    0,
	    storeClientReadHeader,
	    sc);
    else
	file_read(sc->swapin_fd,
	    sc->copy_buf,
	    sc->copy_size,
	    sc->copy_offset + mem->swap_hdr_sz,
	    storeClientReadBody,
	    sc);
    sc->flags.disk_io_pending = 1;
}

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 != 0);
    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 != 0);
    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;
    *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);
    }
#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;
#if OLD_CODE
    store_client *sc;
    store_client *nx = NULL;
    if (mem == NULL)
	return 0;
    for (sc = mem->clients; sc; sc = nx) {
	nx = sc->next;
	/* Changed from callback_data to just callback.  There can be no use */
	/* for callback_data without a callback, and sc->callback we know */
	/* gets reset, but not necessarily sc->callback_data */
	if (sc->callback == NULL)
	    continue;
	npend++;
    }
#endif
    debug(20, 3) ("storePendingNClients: returning %d\n", npend);
    return npend;
}
Commit	Line	Data
f09f5b26	1	#include "squid.h"
f09f5b26	2
e3ef2b09	3	/*
	4	* NOTE: 'Header' refers to the swapfile metadata header.
	5	* 'Body' refers to the swapfile body, which is the full
	6	* HTTP reply (including HTTP headers and body).
	7	*/
	8	static DRCB storeClientReadBody;
	9	static DRCB storeClientReadHeader;
	10	static SIH storeClientFileOpened;
f09f5b26	11	static void storeClientCopy2(StoreEntry * e, store_client * sc);
e3ef2b09	12	static void storeClientFileRead(store_client * sc);
f09f5b26	13
	14	/* check if there is any client waiting for this object at all */
	15	/* return 1 if there is at least one client */
	16	int
	17	storeClientWaiting(const StoreEntry * e)
	18	{
	19	MemObject *mem = e->mem_obj;
	20	store_client *sc;
	21	for (sc = mem->clients; sc; sc = sc->next) {
	22	if (sc->callback_data != NULL)
	23	return 1;
	24	}
	25	return 0;
	26	}
	27
	28	store_client *
	29	storeClientListSearch(const MemObject * mem, void *data)
	30	{
	31	store_client *sc;
	32	for (sc = mem->clients; sc; sc = sc->next) {
	33	if (sc->callback_data == data)
	34	break;
	35	}
	36	return sc;
	37	}
	38
	39	/* add client with fd to client list */
	40	void
	41	storeClientListAdd(StoreEntry * e, void *data)
	42	{
	43	MemObject *mem = e->mem_obj;
	44	store_client **T;
	45	store_client *sc;
	46	assert(mem);
	47	if (storeClientListSearch(mem, data) != NULL)
	48	return;
	49	mem->nclients++;
7021844c	50	sc = memAllocate(MEM_STORE_CLIENT);
f09f5b26	51	cbdataAdd(sc, MEM_STORE_CLIENT); /* sc is callback_data for file_read */
	52	sc->callback_data = data;
	53	sc->seen_offset = 0;
	54	sc->copy_offset = 0;
	55	sc->swapin_fd = -1;
f115fadd	56	sc->flags.disk_io_pending = 0;
07304bf9	57	sc->entry = e;
f09f5b26	58	if (e->store_status == STORE_PENDING && mem->swapout.fd == -1)
	59	sc->type = STORE_MEM_CLIENT;
	60	else
	61	sc->type = STORE_DISK_CLIENT;
	62	for (T = &mem->clients; T; T = &(T)->next);
	63	*T = sc;
	64	}
	65
f115fadd	66	static void
	67	storeClientCopyEvent(void *data)
	68	{
	69	store_client *sc = data;
	70	int valid = cbdataValid(sc);
fe4a33ac	71	debug(20, 3) ("storeClientCopyEvent: Running\n");
f115fadd	72	cbdataUnlock(sc);
	73	if (!valid)
	74	return;
	75	assert(!sc->flags.store_copying);
	76	sc->flags.store_copying = 1;
	77	storeClientCopy2(sc->entry, sc);
	78	sc->flags.store_copying = 0;
	79	}
	80
f09f5b26	81	/* copy bytes requested by the client */
	82	void
	83	storeClientCopy(StoreEntry * e,
	84	off_t seen_offset,
	85	off_t copy_offset,
	86	size_t size,
	87	char *buf,
	88	STCB * callback,
	89	void *data)
	90	{
	91	store_client *sc;
f09f5b26	92	assert(e->store_status != STORE_ABORTED);
f09f5b26	93	debug(20, 3) ("storeClientCopy: %s, seen %d, want %d, size %d, cb %p, cbdata %p\n",
	94	storeKeyText(e->key),
	95	(int) seen_offset,
	96	(int) copy_offset,
	97	(int) size,
	98	callback,
	99	data);
	100	sc = storeClientListSearch(e->mem_obj, data);
	101	assert(sc != NULL);
	102	assert(sc->callback == NULL);
	103	sc->copy_offset = copy_offset;
	104	sc->seen_offset = seen_offset;
	105	sc->callback = callback;
	106	sc->copy_buf = buf;
	107	sc->copy_size = size;
	108	sc->copy_offset = copy_offset;
f115fadd	109	if (sc->flags.store_copying) {
f115fadd	110	cbdataLock(sc);
fe4a33ac	111	debug(20, 3) ("storeClientCopy: Queueing storeClientCopyEvent()\n");
f115fadd	112	eventAdd("storeClientCopyEvent", storeClientCopyEvent, sc, 0);
f115fadd	113	} else {
fe4a33ac	114	sc->flags.store_copying = 1;
	115	storeClientCopy2(e, sc);
	116	sc->flags.store_copying = 0;
f115fadd	117	}
f09f5b26	118	}
f09f5b26	119
07304bf9	120	/*
	121	* This function is used below to decide if we have any more data to
	122	* send to the client. If the store_status is STORE_ABORTED, that case
	123	* should be handled before this function gets called. If the
	124	* store_status is STORE_PENDING, then we do have more data to send.
	125	* If its STORE_OK, then we continue checking. If the object length is
	126	* negative, then we don't know the real length and must open the swap
	127	* file to find out. If the lenght is >= 0, then we compare it to the
	128	* requested copy offset.
	129	*/
	130	static int
	131	storeClientNoMoreToSend(StoreEntry * e, store_client * sc)
	132	{
	133	ssize_t len;
	134	if (e->store_status != STORE_OK)
	135	return 0;
	136	if ((len = objectLen(e)) < 0)
	137	return 0;
	138	if (sc->copy_offset < len)
	139	return 0;
	140	return 1;
	141	}
	142
f09f5b26	143	static void
	144	storeClientCopy2(StoreEntry * e, store_client * sc)
	145	{
	146	STCB *callback = sc->callback;
	147	MemObject *mem = e->mem_obj;
	148	size_t sz;
	149	static int loopdetect = 0;
	150	assert(++loopdetect < 10);
	151	debug(20, 3) ("storeClientCopy2: %s\n", storeKeyText(e->key));
	152	assert(callback != NULL);
	153	if (e->store_status == STORE_ABORTED) {
	154	#if USE_ASYNC_IO
f115fadd	155	if (sc->flags.disk_io_pending == 1) {
f09f5b26	156	if (sc->swapin_fd >= 0)
	157	aioCancel(sc->swapin_fd, NULL);
	158	else
	159	aioCancel(-1, sc);
	160	}
	161	#endif
f115fadd	162	sc->flags.disk_io_pending = 0;
f09f5b26	163	sc->callback = NULL;
f09f5b26	164	callback(sc->callback_data, sc->copy_buf, 0);
07304bf9	165	} else if (storeClientNoMoreToSend(e, sc)) {
f09f5b26	166	/* There is no more to send! */
f09f5b26	167	#if USE_ASYNC_IO
f115fadd	168	if (sc->flags.disk_io_pending == 1) {
f09f5b26	169	if (sc->swapin_fd >= 0)
	170	aioCancel(sc->swapin_fd, NULL);
	171	else
	172	aioCancel(-1, sc);
	173	}
	174	#endif
f115fadd	175	sc->flags.disk_io_pending = 0;
f09f5b26	176	sc->callback = NULL;
	177	callback(sc->callback_data, sc->copy_buf, 0);
	178	} else if (e->store_status == STORE_PENDING && sc->seen_offset == mem->inmem_hi) {
	179	/* client has already seen this, wait for more */
	180	debug(20, 3) ("storeClientCopy2: Waiting for more\n");
	181	} else if (sc->copy_offset >= mem->inmem_lo && mem->inmem_lo < mem->inmem_hi) {
	182	/* What the client wants is in memory */
	183	debug(20, 3) ("storeClientCopy2: Copying from memory\n");
	184	sz = stmemCopy(mem->data, sc->copy_offset, sc->copy_buf, sc->copy_size);
	185	#if USE_ASYNC_IO
f115fadd	186	if (sc->flags.disk_io_pending == 1) {
f09f5b26	187	if (sc->swapin_fd >= 0)
	188	aioCancel(sc->swapin_fd, NULL);
	189	else
	190	aioCancel(-1, sc);
	191	}
	192	#endif
f115fadd	193	sc->flags.disk_io_pending = 0;
f09f5b26	194	sc->callback = NULL;
	195	callback(sc->callback_data, sc->copy_buf, sz);
	196	} else if (sc->swapin_fd < 0) {
	197	debug(20, 3) ("storeClientCopy2: Need to open swap in file\n");
	198	assert(sc->type == STORE_DISK_CLIENT);
	199	/* gotta open the swapin file */
	200	/* assert(sc->copy_offset == 0); */
f115fadd	201	if (sc->flags.disk_io_pending == 0) {
f115fadd	202	sc->flags.disk_io_pending = 1;
e3ef2b09	203	storeSwapInStart(e, storeClientFileOpened, sc);
f09f5b26	204	} else {
	205	debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
	206	}
	207	} else {
	208	debug(20, 3) ("storeClientCopy: reading from disk FD %d\n",
	209	sc->swapin_fd);
	210	assert(sc->type == STORE_DISK_CLIENT);
f115fadd	211	if (sc->flags.disk_io_pending == 0) {
e3ef2b09	212	storeClientFileRead(sc);
f09f5b26	213	} else {
	214	debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
	215	}
	216	}
	217	--loopdetect;
	218	}
	219
	220	static void
e3ef2b09	221	storeClientFileOpened(int fd, void *data)
f09f5b26	222	{
	223	store_client *sc = data;
	224	STCB *callback = sc->callback;
	225	if (fd < 0) {
e3ef2b09	226	debug(20, 3) ("storeClientFileOpened: failed\n");
f115fadd	227	sc->flags.disk_io_pending = 0;
f09f5b26	228	sc->callback = NULL;
	229	callback(sc->callback_data, sc->copy_buf, -1);
	230	return;
	231	}
	232	sc->swapin_fd = fd;
e3ef2b09	233	storeClientFileRead(sc);
f09f5b26	234	}
	235
	236	static void
e3ef2b09	237	storeClientFileRead(store_client * sc)
f09f5b26	238	{
07304bf9	239	MemObject *mem = sc->entry->mem_obj;
f09f5b26	240	assert(sc->callback != NULL);
e3ef2b09	241	if (mem->swap_hdr_sz == 0)
e3ef2b09	242	file_read(sc->swapin_fd,
d8b9a541	243	sc->copy_buf,
d8b9a541	244	sc->copy_size,
e3ef2b09	245	0,
	246	storeClientReadHeader,
	247	sc);
	248	else
	249	file_read(sc->swapin_fd,
	250	sc->copy_buf,
	251	sc->copy_size,
	252	sc->copy_offset + mem->swap_hdr_sz,
	253	storeClientReadBody,
	254	sc);
f115fadd	255	sc->flags.disk_io_pending = 1;
f09f5b26	256	}
	257
	258	static void
e3ef2b09	259	storeClientReadBody(int fd, const char buf, int len, int flagnotused, void data)
f09f5b26	260	{
f09f5b26	261	store_client *sc = data;
07304bf9	262	MemObject *mem = sc->entry->mem_obj;
f09f5b26	263	STCB *callback = sc->callback;
f115fadd	264	assert(sc->flags.disk_io_pending != 0);
f115fadd	265	sc->flags.disk_io_pending = 0;
f09f5b26	266	assert(sc->callback != NULL);
e3ef2b09	267	debug(20, 3) ("storeClientReadBody: FD %d, len %d\n", fd, len);
cb69b4c7	268	if (sc->copy_offset == 0 && len > 0 && mem->reply->sline.status == 0)
cb69b4c7	269	httpReplyParse(mem->reply, sc->copy_buf);
f09f5b26	270	sc->callback = NULL;
	271	callback(sc->callback_data, sc->copy_buf, len);
	272	}
	273
e3ef2b09	274	static void
	275	storeClientReadHeader(int fd, const char buf, int len, int flagnotused, void data)
	276	{
e3ef2b09	277	store_client *sc = data;
07304bf9	278	StoreEntry *e = sc->entry;
07304bf9	279	MemObject *mem = e->mem_obj;
e3ef2b09	280	STCB *callback = sc->callback;
	281	int swap_hdr_sz = 0;
	282	size_t body_sz;
	283	size_t copy_sz;
	284	tlv *tlv_list;
f115fadd	285	assert(sc->flags.disk_io_pending != 0);
f115fadd	286	sc->flags.disk_io_pending = 0;
e3ef2b09	287	assert(sc->callback != NULL);
	288	debug(20, 3) ("storeClientReadHeader: FD %d, len %d\n", fd, len);
	289	if (len < 0) {
25535cbe	290	debug(20, 3) ("storeClientReadHeader: FD %d: %s\n", fd, xstrerror());
25535cbe	291	sc->callback = NULL;
25535cbe	292	callback(sc->callback_data, sc->copy_buf, len);
e3ef2b09	293	return;
	294	}
	295	tlv_list = storeSwapMetaUnpack(buf, &swap_hdr_sz);
	296	if (tlv_list == NULL) {
25535cbe	297	debug(20, 1) ("storeClientReadHeader: failed to unpack meta data\n");
25535cbe	298	sc->callback = NULL;
25535cbe	299	callback(sc->callback_data, sc->copy_buf, -1);
e3ef2b09	300	return;
	301	}
	302	/*
	303	* XXX Here we should check the meta data and make sure we got
	304	* the right object.
	305	*/
07304bf9	306	storeSwapTLVFree(tlv_list);
e3ef2b09	307	mem->swap_hdr_sz = swap_hdr_sz;
07304bf9	308	mem->object_sz = e->swap_file_sz - swap_hdr_sz;
e3ef2b09	309	/*
	310	* If our last read got some data the client wants, then give
	311	* it to them, otherwise schedule another read.
	312	*/
	313	body_sz = len - swap_hdr_sz;
	314	if (sc->copy_offset < body_sz) {
	315	/*
25535cbe	316	* we have (part of) what they want
e3ef2b09	317	*/
e3ef2b09	318	copy_sz = XMIN(sc->copy_size, body_sz);
25535cbe	319	debug(20, 3) ("storeClientReadHeader: copying %d bytes of body\n",
25535cbe	320	copy_sz);
d8b9a541	321	xmemmove(sc->copy_buf, sc->copy_buf + swap_hdr_sz, copy_sz);
cb69b4c7	322	if (sc->copy_offset == 0 && len > 0 && mem->reply->sline.status == 0)
cb69b4c7	323	httpReplyParse(mem->reply, sc->copy_buf);
e3ef2b09	324	sc->callback = NULL;
25535cbe	325	callback(sc->callback_data, sc->copy_buf, copy_sz);
e3ef2b09	326	return;
	327	}
	328	/*
	329	* we don't have what the client wants, but at least we now
	330	* know the swap header size.
	331	*/
e3ef2b09	332	storeClientFileRead(sc);
	333	}
	334
f09f5b26	335	int
	336	storeClientCopyPending(StoreEntry * e, void *data)
	337	{
	338	/* return 1 if there is a callback registered for this client */
	339	store_client *sc = storeClientListSearch(e->mem_obj, data);
	340	if (sc == NULL)
	341	return 0;
	342	if (sc->callback == NULL)
	343	return 0;
	344	return 1;
	345	}
	346
	347	int
	348	storeUnregister(StoreEntry * e, void *data)
	349	{
	350	MemObject *mem = e->mem_obj;
	351	store_client *sc;
	352	store_client **S;
	353	STCB *callback;
	354	if (mem == NULL)
	355	return 0;
	356	debug(20, 3) ("storeUnregister: called for '%s'\n", storeKeyText(e->key));
	357	for (S = &mem->clients; (sc = S) != NULL; S = &(S)->next) {
	358	if (sc->callback_data == data)
	359	break;
	360	}
	361	if (sc == NULL)
	362	return 0;
	363	*S = sc->next;
	364	mem->nclients--;
f115fadd	365	sc->flags.disk_io_pending = 0;
f09f5b26	366	if (e->store_status == STORE_OK && e->swap_status != SWAPOUT_DONE)
	367	storeCheckSwapOut(e);
	368	if (sc->swapin_fd > -1) {
	369	commSetSelect(sc->swapin_fd, COMM_SELECT_READ, NULL, NULL, 0);
	370	file_close(sc->swapin_fd);
	371	}
	372	#if USE_ASYNC_IO
	373	else
	374	aioCancel(-1, sc);
	375	#endif
	376	if ((callback = sc->callback) != NULL) {
	377	/* callback with ssize = -1 to indicate unexpected termination */
	378	debug(20, 3) ("storeUnregister: store_client for %s has a callback\n",
	379	mem->url);
	380	sc->callback = NULL;
	381	callback(sc->callback_data, sc->copy_buf, -1);
	382	}
	383	cbdataFree(sc);
	384	return 1;
	385	}
	386
	387	off_t
	388	storeLowestMemReaderOffset(const StoreEntry * entry)
	389	{
	390	const MemObject *mem = entry->mem_obj;
	391	off_t lowest = mem->inmem_hi;
	392	store_client *sc;
	393	store_client *nx = NULL;
	394	for (sc = mem->clients; sc; sc = nx) {
	395	nx = sc->next;
	396	if (sc->callback_data == NULL) /* open slot */
	397	continue;
	398	if (sc->type != STORE_MEM_CLIENT)
	399	continue;
	400	if (sc->copy_offset < lowest)
	401	lowest = sc->copy_offset;
	402	}
	403	return lowest;
	404	}
	405
	406	/* Call handlers waiting for data to be appended to E. */
	407	void
	408	InvokeHandlers(StoreEntry * e)
	409	{
	410	int i = 0;
	411	MemObject *mem = e->mem_obj;
	412	store_client *sc;
	413	store_client *nx = NULL;
	414	assert(mem->clients != NULL \|\| mem->nclients == 0);
	415	debug(20, 3) ("InvokeHandlers: %s\n", storeKeyText(e->key));
	416	/* walk the entire list looking for valid callbacks */
	417	for (sc = mem->clients; sc; sc = nx) {
	418	nx = sc->next;
	419	debug(20, 3) ("InvokeHandlers: checking client #%d\n", i++);
	420	if (sc->callback_data == NULL)
	421	continue;
	422	if (sc->callback == NULL)
	423	continue;
	424	storeClientCopy2(e, sc);
	425	}
	426	}
	427
	428	int
	429	storePendingNClients(const StoreEntry * e)
430	{
f09f5b26	431	MemObject *mem = e->mem_obj;
36547bcf	432	int npend = NULL == mem ? 0 : mem->nclients;
36547bcf	433	#if OLD_CODE
f09f5b26	434	store_client *sc;
	435	store_client *nx = NULL;
	436	if (mem == NULL)
	437	return 0;
	438	for (sc = mem->clients; sc; sc = nx) {
	439	nx = sc->next;
	440	/* Changed from callback_data to just callback. There can be no use */
	441	/* for callback_data without a callback, and sc->callback we know */
	442	/* gets reset, but not necessarily sc->callback_data */
	443	if (sc->callback == NULL)
	444	continue;
	445	npend++;
	446	}
36547bcf	447	#endif
1afe05c5	448	debug(20, 3) ("storePendingNClients: returning %d\n", npend);
f09f5b26	449	return npend;
f09f5b26	450	}