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

#include "squid.h"

static void storeClientCopy2(StoreEntry * e, store_client * sc);
static SIH storeClientCopyFileOpened;
static void storeClientCopyFileRead(store_client * sc);
static DRCB storeClientCopyHandleRead;

/* 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, storeClientCopyFileOpened, 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;
	    storeClientCopyFileRead(sc);
	} else {
	    debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
	}
    }
    --loopdetect;
}

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

static void
storeClientCopyFileRead(store_client * sc)
{
    assert(sc->callback != NULL);
    file_read(sc->swapin_fd,
	sc->copy_buf,
	sc->copy_size,
	sc->copy_offset,
	storeClientCopyHandleRead,
	sc);
}

static void
storeClientCopyHandleRead(int fd, const char *buf, int len, int flagnotused, void *data)
{
    store_client *sc = data;
    MemObject *mem = sc->mem;
    STCB *callback = sc->callback;
    int hdr_len = 0;
    assert(sc->disk_op_in_progress != 0);
    sc->disk_op_in_progress = 0;
    assert(sc->callback != NULL);
    debug(20, 3) ("storeClientCopyHandleRead: FD %d, len %d\n", fd, len);
#if USE_SWAP_HEADERS
	/* XXX: BROKEN */
    if (sc->copy_offset == 0 && len > 0 && mem != NULL) {
	hdr_len = storeGetMetaBuf(buf, mem);
	memmove((char *) buf, (char *) (buf + hdr_len), len - hdr_len);
	len -= hdr_len;
	httpParseReplyHeaders(buf, mem->reply);
    }
#endif
    sc->callback = NULL;
    callback(sc->callback_data, sc->copy_buf, len);
}

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"
	2
	3	static void storeClientCopy2(StoreEntry * e, store_client * sc);
	4	static SIH storeClientCopyFileOpened;
	5	static void storeClientCopyFileRead(store_client * sc);
	6	static DRCB storeClientCopyHandleRead;
	7
	8	/* check if there is any client waiting for this object at all */
	9	/* return 1 if there is at least one client */
	10	int
	11	storeClientWaiting(const StoreEntry * e)
	12	{
	13	MemObject *mem = e->mem_obj;
	14	store_client *sc;
	15	for (sc = mem->clients; sc; sc = sc->next) {
	16	if (sc->callback_data != NULL)
	17	return 1;
	18	}
	19	return 0;
	20	}
	21
	22	store_client *
	23	storeClientListSearch(const MemObject * mem, void *data)
	24	{
	25	store_client *sc;
	26	for (sc = mem->clients; sc; sc = sc->next) {
	27	if (sc->callback_data == data)
	28	break;
	29	}
	30	return sc;
	31	}
	32
	33	/* add client with fd to client list */
	34	void
	35	storeClientListAdd(StoreEntry * e, void *data)
	36	{
	37	MemObject *mem = e->mem_obj;
	38	store_client **T;
	39	store_client *sc;
	40	assert(mem);
	41	if (storeClientListSearch(mem, data) != NULL)
	42	return;
	43	mem->nclients++;
	44	sc = memAllocate(MEM_STORE_CLIENT, 1);
	45	cbdataAdd(sc, MEM_STORE_CLIENT); /* sc is callback_data for file_read */
	46	sc->callback_data = data;
	47	sc->seen_offset = 0;
	48	sc->copy_offset = 0;
	49	sc->swapin_fd = -1;
	50	sc->disk_op_in_progress = 0;
	51	sc->mem = mem;
	52	if (e->store_status == STORE_PENDING && mem->swapout.fd == -1)
	53	sc->type = STORE_MEM_CLIENT;
	54	else
	55	sc->type = STORE_DISK_CLIENT;
	56	for (T = &mem->clients; T; T = &(T)->next);
	57	*T = sc;
	58	}
	59
	60	/* copy bytes requested by the client */
	61	void
	62	storeClientCopy(StoreEntry * e,
	63	off_t seen_offset,
	64	off_t copy_offset,
65	size_t size,
66	char *buf,
67	STCB * callback,
68	void *data)
69	{
70	store_client *sc;
71	static int recurse_detect = 0;
72	assert(e->store_status != STORE_ABORTED);
73	assert(recurse_detect < 3); /* could == 1 for IMS not modified's */
74	debug(20, 3) ("storeClientCopy: %s, seen %d, want %d, size %d, cb %p, cbdata %p\n",
75	storeKeyText(e->key),
76	(int) seen_offset,
77	(int) copy_offset,
78	(int) size,
79	callback,
80	data);
81	sc = storeClientListSearch(e->mem_obj, data);
82	assert(sc != NULL);
83	assert(sc->callback == NULL);
84	sc->copy_offset = copy_offset;
85	sc->seen_offset = seen_offset;
86	sc->callback = callback;
87	sc->copy_buf = buf;
88	sc->copy_size = size;
89	sc->copy_offset = copy_offset;
90	storeClientCopy2(e, sc);
91	recurse_detect--;
92	}
93
94	static void
95	storeClientCopy2(StoreEntry * e, store_client * sc)
96	{
97	STCB *callback = sc->callback;
98	MemObject *mem = e->mem_obj;
99	size_t sz;
100	static int loopdetect = 0;
101	assert(++loopdetect < 10);
102	debug(20, 3) ("storeClientCopy2: %s\n", storeKeyText(e->key));
103	assert(callback != NULL);
104	if (e->store_status == STORE_ABORTED) {
105	#if USE_ASYNC_IO
106	if (sc->disk_op_in_progress == 1) {
107	if (sc->swapin_fd >= 0)
108	aioCancel(sc->swapin_fd, NULL);
109	else
110	aioCancel(-1, sc);
111	}
112	#endif
113	sc->disk_op_in_progress = 0;
114	sc->callback = NULL;
115	callback(sc->callback_data, sc->copy_buf, 0);
116	} else if (e->store_status == STORE_OK && sc->copy_offset == e->object_len) {
117	/* There is no more to send! */
118	#if USE_ASYNC_IO
119	if (sc->disk_op_in_progress == 1) {
120	if (sc->swapin_fd >= 0)
121	aioCancel(sc->swapin_fd, NULL);
122	else
123	aioCancel(-1, sc);
124	}
125	#endif
126	sc->disk_op_in_progress = 0;
127	sc->callback = NULL;
128	callback(sc->callback_data, sc->copy_buf, 0);
129	} else if (e->store_status == STORE_PENDING && sc->seen_offset == mem->inmem_hi) {
130	/* client has already seen this, wait for more */
131	debug(20, 3) ("storeClientCopy2: Waiting for more\n");
132	} else if (sc->copy_offset >= mem->inmem_lo && mem->inmem_lo < mem->inmem_hi) {
133	/* What the client wants is in memory */
134	debug(20, 3) ("storeClientCopy2: Copying from memory\n");
135	sz = stmemCopy(mem->data, sc->copy_offset, sc->copy_buf, sc->copy_size);
136	#if USE_ASYNC_IO
137	if (sc->disk_op_in_progress == 1) {
138	if (sc->swapin_fd >= 0)
139	aioCancel(sc->swapin_fd, NULL);
140	else
141	aioCancel(-1, sc);
142	}
143	#endif
144	sc->disk_op_in_progress = 0;
145	sc->callback = NULL;
146	callback(sc->callback_data, sc->copy_buf, sz);
147	} else if (sc->swapin_fd < 0) {
148	debug(20, 3) ("storeClientCopy2: Need to open swap in file\n");
149	assert(sc->type == STORE_DISK_CLIENT);
150	/* gotta open the swapin file */
151	/* assert(sc->copy_offset == 0); */
152	if (sc->disk_op_in_progress == 0) {
153	sc->disk_op_in_progress = 1;
154	storeSwapInStart(e, storeClientCopyFileOpened, sc);
155	} else {
156	debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
157	}
158	} else {
159	debug(20, 3) ("storeClientCopy: reading from disk FD %d\n",
160	sc->swapin_fd);
161	assert(sc->type == STORE_DISK_CLIENT);
162	if (sc->disk_op_in_progress == 0) {
163	sc->disk_op_in_progress = 1;
164	storeClientCopyFileRead(sc);
165	} else {
166	debug(20, 2) ("storeClientCopy2: Averted multiple fd operation\n");
167	}
168	}
169	--loopdetect;
170	}
171
172	static void
173	storeClientCopyFileOpened(int fd, void *data)
174	{
175	store_client *sc = data;
176	STCB *callback = sc->callback;
177	if (fd < 0) {
178	debug(20, 3) ("storeClientCopyFileOpened: failed\n");
179	sc->disk_op_in_progress = 0;
180	sc->callback = NULL;
181	callback(sc->callback_data, sc->copy_buf, -1);
182	return;
183	}
184	sc->swapin_fd = fd;
185	storeClientCopyFileRead(sc);
186	}
187
188	static void
189	storeClientCopyFileRead(store_client * sc)
190	{
191	assert(sc->callback != NULL);
192	file_read(sc->swapin_fd,
193	sc->copy_buf,
194	sc->copy_size,
195	sc->copy_offset,
196	storeClientCopyHandleRead,
197	sc);
198	}
199
200	static void
201	storeClientCopyHandleRead(int fd, const char buf, int len, int flagnotused, void data)
202	{
203	store_client *sc = data;
204	MemObject *mem = sc->mem;
205	STCB *callback = sc->callback;
206	int hdr_len = 0;
207	assert(sc->disk_op_in_progress != 0);
208	sc->disk_op_in_progress = 0;
209	assert(sc->callback != NULL);
210	debug(20, 3) ("storeClientCopyHandleRead: FD %d, len %d\n", fd, len);
69c01cd7	211	#if USE_SWAP_HEADERS
69c01cd7	212	/* XXX: BROKEN */
f09f5b26	213	if (sc->copy_offset == 0 && len > 0 && mem != NULL) {
	214	hdr_len = storeGetMetaBuf(buf, mem);
	215	memmove((char ) buf, (char ) (buf + hdr_len), len - hdr_len);
	216	len -= hdr_len;
	217	httpParseReplyHeaders(buf, mem->reply);
	218	}
69c01cd7	219	#endif
f09f5b26	220	sc->callback = NULL;
	221	callback(sc->callback_data, sc->copy_buf, len);
	222	}
	223
	224	int
	225	storeClientCopyPending(StoreEntry * e, void *data)
	226	{
	227	/* return 1 if there is a callback registered for this client */
	228	store_client *sc = storeClientListSearch(e->mem_obj, data);
	229	if (sc == NULL)
	230	return 0;
	231	if (sc->callback == NULL)
	232	return 0;
	233	return 1;
	234	}
	235
	236	int
	237	storeUnregister(StoreEntry * e, void *data)
	238	{
	239	MemObject *mem = e->mem_obj;
	240	store_client *sc;
	241	store_client **S;
	242	STCB *callback;
	243	if (mem == NULL)
	244	return 0;
	245	debug(20, 3) ("storeUnregister: called for '%s'\n", storeKeyText(e->key));
	246	for (S = &mem->clients; (sc = S) != NULL; S = &(S)->next) {
	247	if (sc->callback_data == data)
	248	break;
	249	}
	250	if (sc == NULL)
	251	return 0;
	252	*S = sc->next;
	253	mem->nclients--;
	254	sc->disk_op_in_progress = 0;
	255	if (e->store_status == STORE_OK && e->swap_status != SWAPOUT_DONE)
	256	storeCheckSwapOut(e);
	257	if (sc->swapin_fd > -1) {
	258	commSetSelect(sc->swapin_fd, COMM_SELECT_READ, NULL, NULL, 0);
	259	file_close(sc->swapin_fd);
	260	}
	261	#if USE_ASYNC_IO
	262	else
	263	aioCancel(-1, sc);
	264	#endif
	265	if ((callback = sc->callback) != NULL) {
	266	/* callback with ssize = -1 to indicate unexpected termination */
	267	debug(20, 3) ("storeUnregister: store_client for %s has a callback\n",
	268	mem->url);
	269	sc->callback = NULL;
	270	callback(sc->callback_data, sc->copy_buf, -1);
	271	}
	272	cbdataFree(sc);
	273	return 1;
	274	}
	275
	276	off_t
	277	storeLowestMemReaderOffset(const StoreEntry * entry)
	278	{
	279	const MemObject *mem = entry->mem_obj;
	280	off_t lowest = mem->inmem_hi;
	281	store_client *sc;
	282	store_client *nx = NULL;
	283	for (sc = mem->clients; sc; sc = nx) {
284	nx = sc->next;
285	if (sc->callback_data == NULL) /* open slot */
286	continue;
287	if (sc->type != STORE_MEM_CLIENT)
288	continue;
289	if (sc->copy_offset < lowest)
290	lowest = sc->copy_offset;
291	}
292	return lowest;
293	}
294
295	/* Call handlers waiting for data to be appended to E. */
296	void
297	InvokeHandlers(StoreEntry * e)
298	{
299	int i = 0;
300	MemObject *mem = e->mem_obj;
301	store_client *sc;
302	store_client *nx = NULL;
303	assert(mem->clients != NULL \|\| mem->nclients == 0);
304	debug(20, 3) ("InvokeHandlers: %s\n", storeKeyText(e->key));
305	/* walk the entire list looking for valid callbacks */
306	for (sc = mem->clients; sc; sc = nx) {
307	nx = sc->next;
308	debug(20, 3) ("InvokeHandlers: checking client #%d\n", i++);
309	if (sc->callback_data == NULL)
310	continue;
311	if (sc->callback == NULL)
312	continue;
313	storeClientCopy2(e, sc);
314	}
315	}
316
317	int
318	storePendingNClients(const StoreEntry * e)
319	{
320	int npend = 0;
321	MemObject *mem = e->mem_obj;
322	store_client *sc;
323	store_client *nx = NULL;
324	if (mem == NULL)
325	return 0;
326	for (sc = mem->clients; sc; sc = nx) {
327	nx = sc->next;
328	/* Changed from callback_data to just callback. There can be no use */
329	/* for callback_data without a callback, and sc->callback we know */
330	/* gets reset, but not necessarily sc->callback_data */
331	if (sc->callback == NULL)
332	continue;
333	npend++;
334	}
335	return npend;
336	}