1/*	$NetBSD: ev_streams.c,v 1.6 2009/04/12 17:07:17 christos Exp $	*/
2
3/*
4 * Copyright (c) 2004 by Internet Systems Consortium, Inc. ("ISC")
5 * Copyright (c) 1996-1999 by Internet Software Consortium
6 *
7 * Permission to use, copy, modify, and distribute this software for any
8 * purpose with or without fee is hereby granted, provided that the above
9 * copyright notice and this permission notice appear in all copies.
10 *
11 * THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES
12 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
13 * MERCHANTABILITY AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR
14 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
15 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
16 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
17 * OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
18 */
19
20/* ev_streams.c - implement asynch stream file IO for the eventlib
21 * vix 04mar96 [initial]
22 */
23
24#include <sys/cdefs.h>
25#if !defined(LINT) && !defined(CODECENTER) && !defined(lint)
26#ifdef notdef
27static const char rcsid[] = "Id: ev_streams.c,v 1.5 2005/04/27 04:56:36 sra Exp";
28#else
29__RCSID("$NetBSD: ev_streams.c,v 1.6 2009/04/12 17:07:17 christos Exp $");
30#endif
31#endif
32
33#include "port_before.h"
34#include "fd_setsize.h"
35
36#include <sys/types.h>
37#include <sys/uio.h>
38
39#include <errno.h>
40
41#include <isc/eventlib.h>
42#include <isc/assertions.h>
43#include "eventlib_p.h"
44
45#include "port_after.h"
46
47#ifndef _LIBC
48static int	copyvec(evStream *str, const struct iovec *iov, int iocnt);
49static void	consume(evStream *str, size_t bytes);
50static void	done(evContext opaqueCtx, evStream *str);
51static void	writable(evContext opaqueCtx, void *uap, int fd, int evmask);
52static void	readable(evContext opaqueCtx, void *uap, int fd, int evmask);
53#endif
54
55struct iovec
56evConsIovec(void *buf, size_t cnt) {
57	struct iovec ret;
58
59	memset(&ret, 0xf5, sizeof ret);
60	ret.iov_base = buf;
61	ret.iov_len = cnt;
62	return (ret);
63}
64
65#ifndef _LIBC
66int
67evWrite(evContext opaqueCtx, int fd, const struct iovec *iov, int iocnt,
68	evStreamFunc func, void *uap, evStreamID *id)
69{
70	evContext_p *ctx = opaqueCtx.opaque;
71	evStream *new;
72	int save;
73
74	OKNEW(new);
75	new->func = func;
76	new->uap = uap;
77	new->fd = fd;
78	new->flags = 0;
79	if (evSelectFD(opaqueCtx, fd, EV_WRITE, writable, new, &new->file) < 0)
80		goto free;
81	if (copyvec(new, iov, iocnt) < 0)
82		goto free;
83	new->prevDone = NULL;
84	new->nextDone = NULL;
85	if (ctx->streams != NULL)
86		ctx->streams->prev = new;
87	new->prev = NULL;
88	new->next = ctx->streams;
89	ctx->streams = new;
90	if (id != NULL)
91		id->opaque = new;
92	return (0);
93 free:
94	save = errno;
95	FREE(new);
96	errno = save;
97	return (-1);
98}
99
100int
101evRead(evContext opaqueCtx, int fd, const struct iovec *iov, int iocnt,
102       evStreamFunc func, void *uap, evStreamID *id)
103{
104	evContext_p *ctx = opaqueCtx.opaque;
105	evStream *new;
106	int save;
107
108	OKNEW(new);
109	new->func = func;
110	new->uap = uap;
111	new->fd = fd;
112	new->flags = 0;
113	if (evSelectFD(opaqueCtx, fd, EV_READ, readable, new, &new->file) < 0)
114		goto free;
115	if (copyvec(new, iov, iocnt) < 0)
116		goto free;
117	new->prevDone = NULL;
118	new->nextDone = NULL;
119	if (ctx->streams != NULL)
120		ctx->streams->prev = new;
121	new->prev = NULL;
122	new->next = ctx->streams;
123	ctx->streams = new;
124	if (id)
125		id->opaque = new;
126	return (0);
127 free:
128	save = errno;
129	FREE(new);
130	errno = save;
131	return (-1);
132}
133
134int
135evTimeRW(evContext opaqueCtx, evStreamID id, evTimerID timer) /*ARGSUSED*/ {
136	evStream *str = id.opaque;
137
138	UNUSED(opaqueCtx);
139
140	str->timer = timer;
141	str->flags |= EV_STR_TIMEROK;
142	return (0);
143}
144
145int
146evUntimeRW(evContext opaqueCtx, evStreamID id) /*ARGSUSED*/ {
147	evStream *str = id.opaque;
148
149	UNUSED(opaqueCtx);
150
151	str->flags &= ~EV_STR_TIMEROK;
152	return (0);
153}
154
155int
156evCancelRW(evContext opaqueCtx, evStreamID id) {
157	evContext_p *ctx = opaqueCtx.opaque;
158	evStream *old = id.opaque;
159
160	/*
161	 * The streams list is doubly threaded.  First, there's ctx->streams
162	 * that's used by evDestroy() to find and cancel all streams.  Second,
163	 * there's ctx->strDone (head) and ctx->strLast (tail) which thread
164	 * through the potentially smaller number of "IO completed" streams,
165	 * used in evGetNext() to avoid scanning the entire list.
166	 */
167
168	/* Unlink from ctx->streams. */
169	if (old->prev != NULL)
170		old->prev->next = old->next;
171	else
172		ctx->streams = old->next;
173	if (old->next != NULL)
174		old->next->prev = old->prev;
175
176	/*
177	 * If 'old' is on the ctx->strDone list, remove it.  Update
178	 * ctx->strLast if necessary.
179	 */
180	if (old->prevDone == NULL && old->nextDone == NULL) {
181		/*
182		 * Either 'old' is the only item on the done list, or it's
183		 * not on the done list.  If the former, then we unlink it
184		 * from the list.  If the latter, we leave the list alone.
185		 */
186		if (ctx->strDone == old) {
187			ctx->strDone = NULL;
188			ctx->strLast = NULL;
189		}
190	} else {
191		if (old->prevDone != NULL)
192			old->prevDone->nextDone = old->nextDone;
193		else
194			ctx->strDone = old->nextDone;
195		if (old->nextDone != NULL)
196			old->nextDone->prevDone = old->prevDone;
197		else
198			ctx->strLast = old->prevDone;
199	}
200
201	/* Deallocate the stream. */
202	if (old->file.opaque)
203		evDeselectFD(opaqueCtx, old->file);
204	memput(old->iovOrig, sizeof (struct iovec) * old->iovOrigCount);
205	FREE(old);
206	return (0);
207}
208
209/* Copy a scatter/gather vector and initialize a stream handler's IO. */
210static int
211copyvec(evStream *str, const struct iovec *iov, int iocnt) {
212	int i;
213
214	str->iovOrig = (struct iovec *)memget(sizeof(struct iovec) * iocnt);
215	if (str->iovOrig == NULL) {
216		errno = ENOMEM;
217		return (-1);
218	}
219	str->ioTotal = 0;
220	for (i = 0; i < iocnt; i++) {
221		str->iovOrig[i] = iov[i];
222		str->ioTotal += iov[i].iov_len;
223	}
224	str->iovOrigCount = iocnt;
225	str->iovCur = str->iovOrig;
226	str->iovCurCount = str->iovOrigCount;
227	str->ioDone = 0;
228	return (0);
229}
230
231/* Pull off or truncate lead iovec(s). */
232static void
233consume(evStream *str, size_t bytes) {
234	while (bytes > 0U) {
235		if (bytes < (size_t)str->iovCur->iov_len) {
236			str->iovCur->iov_len -= bytes;
237			str->iovCur->iov_base = (void *)
238				((u_char *)str->iovCur->iov_base + bytes);
239			str->ioDone += bytes;
240			bytes = 0;
241		} else {
242			bytes -= str->iovCur->iov_len;
243			str->ioDone += str->iovCur->iov_len;
244			str->iovCur++;
245			str->iovCurCount--;
246		}
247	}
248}
249
250/* Add a stream to Done list and deselect the FD. */
251static void
252done(evContext opaqueCtx, evStream *str) {
253	evContext_p *ctx = opaqueCtx.opaque;
254
255	if (ctx->strLast != NULL) {
256		str->prevDone = ctx->strLast;
257		ctx->strLast->nextDone = str;
258		ctx->strLast = str;
259	} else {
260		INSIST(ctx->strDone == NULL);
261		ctx->strDone = ctx->strLast = str;
262	}
263	evDeselectFD(opaqueCtx, str->file);
264	str->file.opaque = NULL;
265	/* evDrop() will call evCancelRW() on us. */
266}
267
268/* Dribble out some bytes on the stream.  (Called by evDispatch().) */
269static void
270writable(evContext opaqueCtx, void *uap, int fd, int evmask) {
271	evStream *str = uap;
272	int bytes;
273
274	UNUSED(evmask);
275
276	bytes = writev(fd, str->iovCur, str->iovCurCount);
277	if (bytes > 0) {
278		if ((str->flags & EV_STR_TIMEROK) != 0)
279			evTouchIdleTimer(opaqueCtx, str->timer);
280		consume(str, bytes);
281	} else {
282		if (bytes < 0 && errno != EINTR) {
283			str->ioDone = -1;
284			str->ioErrno = errno;
285		}
286	}
287	if (str->ioDone == -1 || str->ioDone == str->ioTotal)
288		done(opaqueCtx, str);
289}
290
291/* Scoop up some bytes from the stream.  (Called by evDispatch().) */
292static void
293readable(evContext opaqueCtx, void *uap, int fd, int evmask) {
294	evStream *str = uap;
295	int bytes;
296
297	UNUSED(evmask);
298
299	bytes = readv(fd, str->iovCur, str->iovCurCount);
300	if (bytes > 0) {
301		if ((str->flags & EV_STR_TIMEROK) != 0)
302			evTouchIdleTimer(opaqueCtx, str->timer);
303		consume(str, bytes);
304	} else {
305		if (bytes == 0)
306			str->ioDone = 0;
307		else {
308			if (errno != EINTR) {
309				str->ioDone = -1;
310				str->ioErrno = errno;
311			}
312		}
313	}
314	if (str->ioDone <= 0 || str->ioDone == str->ioTotal)
315		done(opaqueCtx, str);
316}
317#endif
318
319/*! \file */
320