[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, 1);
    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->disk_op_in_progress = 0;
    sc->mem = mem;
    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;
}

/* 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;
    static int recurse_detect = 0;
    assert(e->store_status != STORE_ABORTED);
    assert(recurse_detect < 3);	/* could == 1 for IMS not modified's */
    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);
    recurse_detect--;
}

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->disk_op_in_progress == 1) {
	    if (sc->swapin_fd >= 0)
		aioCancel(sc->swapin_fd, NULL);
	    else
		aioCancel(-1, sc);
	}
#endif
	sc->disk_op_in_progress = 0;
	sc->callback = NULL;
	callback(sc->callback_data, sc->copy_buf, 0);
    } else if (e->store_status == STORE_OK && sc->copy_offset == e->object_len) {
	/* There is no more to send! */
#if USE_ASYNC_IO
	if (sc->disk_op_in_progress == 1) {
	    if (sc->swapin_fd >= 0)
		aioCancel(sc->swapin_fd, NULL);
	    else
		aioCancel(-1, sc);
	}
#endif
	sc->disk_op_in_progress = 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->disk_op_in_progress == 1) {
	    if (sc->swapin_fd >= 0)
		aioCancel(sc->swapin_fd, NULL);
	    else
		aioCancel(-1, sc);
	}
#endif
	sc->disk_op_in_progress = 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->disk_op_in_progress == 0) {
	    sc->disk_op_in_progress = 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->disk_op_in_progress == 0) {
	    sc->disk_op_in_progress = 1;
	    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->disk_op_in_progress = 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->mem;
    assert(sc->callback != NULL);
    if (mem->swap_hdr_sz == 0)
	file_read(sc->swapin_fd,
	    memAllocate(MEM_DISK_BUF, 1),
	    DISK_PAGE_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);
}

static void
storeClientReadBody(int fd, const char *buf, int len, int flagnotused, void *data)
{
    store_client *sc = data;
    MemObject *mem = sc->mem;
    STCB *callback = sc->callback;
    assert(sc->disk_op_in_progress != 0);
    sc->disk_op_in_progress = 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->code == 0)
	httpParseReplyHeaders(sc->copy_buf, mem->reply);
    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)
{
    /*
     * 'buf' should have been allocated by memAllocate(MEM_DISK_BUF)
     */
    store_client *sc = data;
    MemObject *mem = sc->mem;
    STCB *callback = sc->callback;
    int swap_hdr_sz = 0;
    size_t body_sz;
    size_t copy_sz;
    tlv *tlv_list;
    assert(sc->disk_op_in_progress != 0);
    sc->disk_op_in_progress = 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());
	memFree(MEM_DISK_BUF, (void *) buf);
	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");
	memFree(MEM_DISK_BUF, (void *) buf);
	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.
     */
    mem->swap_hdr_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);
	xmemcpy(sc->copy_buf, buf + swap_hdr_sz, copy_sz);
	memFree(MEM_DISK_BUF, (void *) buf);
	if (sc->copy_offset == 0 && len > 0 && mem->reply->code == 0)
	    httpParseReplyHeaders(sc->copy_buf, mem->reply);
	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.
     */
    memFree(MEM_DISK_BUF, (void *) buf);
    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->disk_op_in_progress = 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)
{
    int npend = 0;
    MemObject *mem = e->mem_obj;
    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++;
    }
    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++;
	50	sc = memAllocate(MEM_STORE_CLIENT, 1);
	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;
	56	sc->disk_op_in_progress = 0;
	57	sc->mem = mem;
	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
	66	/* copy bytes requested by the client */
	67	void
	68	storeClientCopy(StoreEntry * e,
	69	off_t seen_offset,
	70	off_t copy_offset,
	71	size_t size,
	72	char *buf,
	73	STCB * callback,
	74	void *data)
	75	{
	76	store_client *sc;
77	static int recurse_detect = 0;
78	assert(e->store_status != STORE_ABORTED);
79	assert(recurse_detect < 3); /* could == 1 for IMS not modified's */
80	debug(20, 3) ("storeClientCopy: %s, seen %d, want %d, size %d, cb %p, cbdata %p\n",
81	storeKeyText(e->key),
82	(int) seen_offset,
83	(int) copy_offset,
84	(int) size,
85	callback,
86	data);
87	sc = storeClientListSearch(e->mem_obj, data);
88	assert(sc != NULL);
89	assert(sc->callback == NULL);
90	sc->copy_offset = copy_offset;
91	sc->seen_offset = seen_offset;
92	sc->callback = callback;
93	sc->copy_buf = buf;
94	sc->copy_size = size;
95	sc->copy_offset = copy_offset;
96	storeClientCopy2(e, sc);
97	recurse_detect--;
98	}
99
100	static void
101	storeClientCopy2(StoreEntry * e, store_client * sc)
102	{
103	STCB *callback = sc->callback;
104	MemObject *mem = e->mem_obj;
105	size_t sz;
106	static int loopdetect = 0;
107	assert(++loopdetect < 10);
108	debug(20, 3) ("storeClientCopy2: %s\n", storeKeyText(e->key));
109	assert(callback != NULL);
110	if (e->store_status == STORE_ABORTED) {
111	#if USE_ASYNC_IO
112	if (sc->disk_op_in_progress == 1) {
113	if (sc->swapin_fd >= 0)
114	aioCancel(sc->swapin_fd, NULL);
115	else
116	aioCancel(-1, sc);
117	}
118	#endif
119	sc->disk_op_in_progress = 0;
120	sc->callback = NULL;
121	callback(sc->callback_data, sc->copy_buf, 0);
122	} else if (e->store_status == STORE_OK && sc->copy_offset == e->object_len) {
123	/* There is no more to send! */
124	#if USE_ASYNC_IO
125	if (sc->disk_op_in_progress == 1) {
126	if (sc->swapin_fd >= 0)
127	aioCancel(sc->swapin_fd, NULL);
128	else
129	aioCancel(-1, sc);
130	}
131	#endif
132	sc->disk_op_in_progress = 0;
133	sc->callback = NULL;
134	callback(sc->callback_data, sc->copy_buf, 0);
135	} else if (e->store_status == STORE_PENDING && sc->seen_offset == mem->inmem_hi) {
136	/* client has already seen this, wait for more */
137	debug(20, 3) ("storeClientCopy2: Waiting for more\n");
138	} else if (sc->copy_offset >= mem->inmem_lo && mem->inmem_lo < mem->inmem_hi) {
139	/* What the client wants is in memory */
140	debug(20, 3) ("storeClientCopy2: Copying from memory\n");
141	sz = stmemCopy(mem->data, sc->copy_offset, sc->copy_buf, sc->copy_size);
142	#if USE_ASYNC_IO
143	if (sc->disk_op_in_progress == 1) {
144	if (sc->swapin_fd >= 0)
145	aioCancel(sc->swapin_fd, NULL);
146	else
147	aioCancel(-1, sc);
148	}
149	#endif
150	sc->disk_op_in_progress = 0;
151	sc->callback = NULL;
152	callback(sc->callback_data, sc->copy_buf, sz);
153	} else if (sc->swapin_fd < 0) {
154	debug(20, 3) ("storeClientCopy2: Need to open swap in file\n");
155	assert(sc->type == STORE_DISK_CLIENT);
156	/* gotta open the swapin file */
157	/* assert(sc->copy_offset == 0); */
158	if (sc->disk_op_in_progress == 0) {
159	sc->disk_op_in_progress = 1;
e3ef2b09	160	storeSwapInStart(e, storeClientFileOpened, sc);
f09f5b26	161	} else {
	162	debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
	163	}
	164	} else {
	165	debug(20, 3) ("storeClientCopy: reading from disk FD %d\n",
	166	sc->swapin_fd);
	167	assert(sc->type == STORE_DISK_CLIENT);
	168	if (sc->disk_op_in_progress == 0) {
	169	sc->disk_op_in_progress = 1;
e3ef2b09	170	storeClientFileRead(sc);
f09f5b26	171	} else {
	172	debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
	173	}
	174	}
	175	--loopdetect;
	176	}
	177
	178	static void
e3ef2b09	179	storeClientFileOpened(int fd, void *data)
f09f5b26	180	{
	181	store_client *sc = data;
	182	STCB *callback = sc->callback;
	183	if (fd < 0) {
e3ef2b09	184	debug(20, 3) ("storeClientFileOpened: failed\n");
f09f5b26	185	sc->disk_op_in_progress = 0;
	186	sc->callback = NULL;
	187	callback(sc->callback_data, sc->copy_buf, -1);
	188	return;
	189	}
	190	sc->swapin_fd = fd;
e3ef2b09	191	storeClientFileRead(sc);
f09f5b26	192	}
	193
	194	static void
e3ef2b09	195	storeClientFileRead(store_client * sc)
f09f5b26	196	{
e3ef2b09	197	MemObject *mem = sc->mem;
f09f5b26	198	assert(sc->callback != NULL);
e3ef2b09	199	if (mem->swap_hdr_sz == 0)
	200	file_read(sc->swapin_fd,
	201	memAllocate(MEM_DISK_BUF, 1),
	202	DISK_PAGE_SIZE,
	203	0,
	204	storeClientReadHeader,
	205	sc);
	206	else
	207	file_read(sc->swapin_fd,
	208	sc->copy_buf,
	209	sc->copy_size,
	210	sc->copy_offset + mem->swap_hdr_sz,
	211	storeClientReadBody,
	212	sc);
f09f5b26	213	}
	214
	215	static void
e3ef2b09	216	storeClientReadBody(int fd, const char buf, int len, int flagnotused, void data)
f09f5b26	217	{
f09f5b26	218	store_client *sc = data;
25535cbe	219	MemObject *mem = sc->mem;
f09f5b26	220	STCB *callback = sc->callback;
f09f5b26	221	assert(sc->disk_op_in_progress != 0);
	222	sc->disk_op_in_progress = 0;
	223	assert(sc->callback != NULL);
e3ef2b09	224	debug(20, 3) ("storeClientReadBody: FD %d, len %d\n", fd, len);
25535cbe	225	if (sc->copy_offset == 0 && len > 0 && mem->reply->code == 0)
25535cbe	226	httpParseReplyHeaders(sc->copy_buf, mem->reply);
f09f5b26	227	sc->callback = NULL;
	228	callback(sc->callback_data, sc->copy_buf, len);
	229	}
	230
e3ef2b09	231	static void
	232	storeClientReadHeader(int fd, const char buf, int len, int flagnotused, void data)
	233	{
	234	/*
	235	* 'buf' should have been allocated by memAllocate(MEM_DISK_BUF)
	236	*/
	237	store_client *sc = data;
	238	MemObject *mem = sc->mem;
	239	STCB *callback = sc->callback;
	240	int swap_hdr_sz = 0;
	241	size_t body_sz;
	242	size_t copy_sz;
	243	tlv *tlv_list;
	244	assert(sc->disk_op_in_progress != 0);
	245	sc->disk_op_in_progress = 0;
	246	assert(sc->callback != NULL);
	247	debug(20, 3) ("storeClientReadHeader: FD %d, len %d\n", fd, len);
	248	if (len < 0) {
25535cbe	249	debug(20, 3) ("storeClientReadHeader: FD %d: %s\n", fd, xstrerror());
e3ef2b09	250	memFree(MEM_DISK_BUF, (void *) buf);
25535cbe	251	sc->callback = NULL;
25535cbe	252	callback(sc->callback_data, sc->copy_buf, len);
e3ef2b09	253	return;
	254	}
	255	tlv_list = storeSwapMetaUnpack(buf, &swap_hdr_sz);
	256	if (tlv_list == NULL) {
25535cbe	257	debug(20, 1) ("storeClientReadHeader: failed to unpack meta data\n");
e3ef2b09	258	memFree(MEM_DISK_BUF, (void *) buf);
25535cbe	259	sc->callback = NULL;
25535cbe	260	callback(sc->callback_data, sc->copy_buf, -1);
e3ef2b09	261	return;
	262	}
	263	/*
	264	* XXX Here we should check the meta data and make sure we got
	265	* the right object.
	266	*/
	267	mem->swap_hdr_sz = swap_hdr_sz;
	268	/*
	269	* If our last read got some data the client wants, then give
	270	* it to them, otherwise schedule another read.
	271	*/
	272	body_sz = len - swap_hdr_sz;
	273	if (sc->copy_offset < body_sz) {
	274	/*
25535cbe	275	* we have (part of) what they want
e3ef2b09	276	*/
e3ef2b09	277	copy_sz = XMIN(sc->copy_size, body_sz);
25535cbe	278	debug(20, 3) ("storeClientReadHeader: copying %d bytes of body\n",
	279	copy_sz);
	280	xmemcpy(sc->copy_buf, buf + swap_hdr_sz, copy_sz);
e3ef2b09	281	memFree(MEM_DISK_BUF, (void *) buf);
25535cbe	282	if (sc->copy_offset == 0 && len > 0 && mem->reply->code == 0)
25535cbe	283	httpParseReplyHeaders(sc->copy_buf, mem->reply);
e3ef2b09	284	sc->callback = NULL;
25535cbe	285	callback(sc->callback_data, sc->copy_buf, copy_sz);
e3ef2b09	286	return;
	287	}
	288	/*
	289	* we don't have what the client wants, but at least we now
	290	* know the swap header size.
	291	*/
	292	memFree(MEM_DISK_BUF, (void *) buf);
	293	storeClientFileRead(sc);
	294	}
	295
f09f5b26	296	int
	297	storeClientCopyPending(StoreEntry * e, void *data)
	298	{
	299	/* return 1 if there is a callback registered for this client */
	300	store_client *sc = storeClientListSearch(e->mem_obj, data);
	301	if (sc == NULL)
	302	return 0;
	303	if (sc->callback == NULL)
	304	return 0;
	305	return 1;
	306	}
	307
	308	int
	309	storeUnregister(StoreEntry * e, void *data)
	310	{
	311	MemObject *mem = e->mem_obj;
	312	store_client *sc;
	313	store_client **S;
	314	STCB *callback;
	315	if (mem == NULL)
	316	return 0;
	317	debug(20, 3) ("storeUnregister: called for '%s'\n", storeKeyText(e->key));
	318	for (S = &mem->clients; (sc = S) != NULL; S = &(S)->next) {
	319	if (sc->callback_data == data)
	320	break;
	321	}
	322	if (sc == NULL)
	323	return 0;
	324	*S = sc->next;
	325	mem->nclients--;
	326	sc->disk_op_in_progress = 0;
	327	if (e->store_status == STORE_OK && e->swap_status != SWAPOUT_DONE)
	328	storeCheckSwapOut(e);
	329	if (sc->swapin_fd > -1) {
	330	commSetSelect(sc->swapin_fd, COMM_SELECT_READ, NULL, NULL, 0);
	331	file_close(sc->swapin_fd);
	332	}
	333	#if USE_ASYNC_IO
	334	else
	335	aioCancel(-1, sc);
	336	#endif
	337	if ((callback = sc->callback) != NULL) {
	338	/* callback with ssize = -1 to indicate unexpected termination */
	339	debug(20, 3) ("storeUnregister: store_client for %s has a callback\n",
	340	mem->url);
	341	sc->callback = NULL;
	342	callback(sc->callback_data, sc->copy_buf, -1);
	343	}
	344	cbdataFree(sc);
	345	return 1;
	346	}
	347
	348	off_t
	349	storeLowestMemReaderOffset(const StoreEntry * entry)
	350	{
	351	const MemObject *mem = entry->mem_obj;
	352	off_t lowest = mem->inmem_hi;
	353	store_client *sc;
	354	store_client *nx = NULL;
	355	for (sc = mem->clients; sc; sc = nx) {
	356	nx = sc->next;
	357	if (sc->callback_data == NULL) /* open slot */
	358	continue;
	359	if (sc->type != STORE_MEM_CLIENT)
360	continue;
361	if (sc->copy_offset < lowest)
362	lowest = sc->copy_offset;
363	}
364	return lowest;
365	}
366
367	/* Call handlers waiting for data to be appended to E. */
368	void
369	InvokeHandlers(StoreEntry * e)
370	{
371	int i = 0;
372	MemObject *mem = e->mem_obj;
373	store_client *sc;
374	store_client *nx = NULL;
375	assert(mem->clients != NULL \|\| mem->nclients == 0);
376	debug(20, 3) ("InvokeHandlers: %s\n", storeKeyText(e->key));
377	/* walk the entire list looking for valid callbacks */
378	for (sc = mem->clients; sc; sc = nx) {
379	nx = sc->next;
380	debug(20, 3) ("InvokeHandlers: checking client #%d\n", i++);
381	if (sc->callback_data == NULL)
382	continue;
383	if (sc->callback == NULL)
384	continue;
385	storeClientCopy2(e, sc);
386	}
387	}
388
389	int
390	storePendingNClients(const StoreEntry * e)
391	{
392	int npend = 0;
393	MemObject *mem = e->mem_obj;
394	store_client *sc;
395	store_client *nx = NULL;
396	if (mem == NULL)
397	return 0;
398	for (sc = mem->clients; sc; sc = nx) {
399	nx = sc->next;
400	/* Changed from callback_data to just callback. There can be no use */
401	/* for callback_data without a callback, and sc->callback we know */
402	/* gets reset, but not necessarily sc->callback_data */
403	if (sc->callback == NULL)
404	continue;
405	npend++;
406	}
407	return npend;
408	}