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