1 /* $NetBSD: ev_streams.c,v 1.6 2009/04/12 17:07:17 christos Exp $ */
4 * Copyright (c) 2004 by Internet Systems Consortium, Inc. ("ISC")
5 * Copyright (c) 1996-1999 by Internet Software Consortium
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.
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.
20 /* ev_streams.c - implement asynch stream file IO for the eventlib
21 * vix 04mar96 [initial]
24 #include <sys/cdefs.h>
25 #if !defined(LINT) && !defined(CODECENTER) && !defined(lint)
27 static const char rcsid
[] = "Id: ev_streams.c,v 1.5 2005/04/27 04:56:36 sra Exp";
29 __RCSID("$NetBSD: ev_streams.c,v 1.6 2009/04/12 17:07:17 christos Exp $");
33 #include "port_before.h"
34 #include "fd_setsize.h"
36 #include <sys/types.h>
41 #include <isc/eventlib.h>
42 #include <isc/assertions.h>
43 #include "eventlib_p.h"
45 #include "port_after.h"
48 static int copyvec(evStream
*str
, const struct iovec
*iov
, int iocnt
);
49 static void consume(evStream
*str
, size_t bytes
);
50 static void done(evContext opaqueCtx
, evStream
*str
);
51 static void writable(evContext opaqueCtx
, void *uap
, int fd
, int evmask
);
52 static void readable(evContext opaqueCtx
, void *uap
, int fd
, int evmask
);
56 evConsIovec(void *buf
, size_t cnt
) {
59 memset(&ret
, 0xf5, sizeof ret
);
67 evWrite(evContext opaqueCtx
, int fd
, const struct iovec
*iov
, int iocnt
,
68 evStreamFunc func
, void *uap
, evStreamID
*id
)
70 evContext_p
*ctx
= opaqueCtx
.opaque
;
79 if (evSelectFD(opaqueCtx
, fd
, EV_WRITE
, writable
, new, &new->file
) < 0)
81 if (copyvec(new, iov
, iocnt
) < 0)
85 if (ctx
->streams
!= NULL
)
86 ctx
->streams
->prev
= new;
88 new->next
= ctx
->streams
;
101 evRead(evContext opaqueCtx
, int fd
, const struct iovec
*iov
, int iocnt
,
102 evStreamFunc func
, void *uap
, evStreamID
*id
)
104 evContext_p
*ctx
= opaqueCtx
.opaque
;
113 if (evSelectFD(opaqueCtx
, fd
, EV_READ
, readable
, new, &new->file
) < 0)
115 if (copyvec(new, iov
, iocnt
) < 0)
117 new->prevDone
= NULL
;
118 new->nextDone
= NULL
;
119 if (ctx
->streams
!= NULL
)
120 ctx
->streams
->prev
= new;
122 new->next
= ctx
->streams
;
135 evTimeRW(evContext opaqueCtx
, evStreamID id
, evTimerID timer
) /*ARGSUSED*/ {
136 evStream
*str
= id
.opaque
;
141 str
->flags
|= EV_STR_TIMEROK
;
146 evUntimeRW(evContext opaqueCtx
, evStreamID id
) /*ARGSUSED*/ {
147 evStream
*str
= id
.opaque
;
151 str
->flags
&= ~EV_STR_TIMEROK
;
156 evCancelRW(evContext opaqueCtx
, evStreamID id
) {
157 evContext_p
*ctx
= opaqueCtx
.opaque
;
158 evStream
*old
= id
.opaque
;
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.
168 /* Unlink from ctx->streams. */
169 if (old
->prev
!= NULL
)
170 old
->prev
->next
= old
->next
;
172 ctx
->streams
= old
->next
;
173 if (old
->next
!= NULL
)
174 old
->next
->prev
= old
->prev
;
177 * If 'old' is on the ctx->strDone list, remove it. Update
178 * ctx->strLast if necessary.
180 if (old
->prevDone
== NULL
&& old
->nextDone
== NULL
) {
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.
186 if (ctx
->strDone
== old
) {
191 if (old
->prevDone
!= NULL
)
192 old
->prevDone
->nextDone
= old
->nextDone
;
194 ctx
->strDone
= old
->nextDone
;
195 if (old
->nextDone
!= NULL
)
196 old
->nextDone
->prevDone
= old
->prevDone
;
198 ctx
->strLast
= old
->prevDone
;
201 /* Deallocate the stream. */
202 if (old
->file
.opaque
)
203 evDeselectFD(opaqueCtx
, old
->file
);
204 memput(old
->iovOrig
, sizeof (struct iovec
) * old
->iovOrigCount
);
209 /* Copy a scatter/gather vector and initialize a stream handler's IO. */
211 copyvec(evStream
*str
, const struct iovec
*iov
, int iocnt
) {
214 str
->iovOrig
= (struct iovec
*)memget(sizeof(struct iovec
) * iocnt
);
215 if (str
->iovOrig
== NULL
) {
220 for (i
= 0; i
< iocnt
; i
++) {
221 str
->iovOrig
[i
] = iov
[i
];
222 str
->ioTotal
+= iov
[i
].iov_len
;
224 str
->iovOrigCount
= iocnt
;
225 str
->iovCur
= str
->iovOrig
;
226 str
->iovCurCount
= str
->iovOrigCount
;
231 /* Pull off or truncate lead iovec(s). */
233 consume(evStream
*str
, size_t bytes
) {
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
;
242 bytes
-= str
->iovCur
->iov_len
;
243 str
->ioDone
+= str
->iovCur
->iov_len
;
250 /* Add a stream to Done list and deselect the FD. */
252 done(evContext opaqueCtx
, evStream
*str
) {
253 evContext_p
*ctx
= opaqueCtx
.opaque
;
255 if (ctx
->strLast
!= NULL
) {
256 str
->prevDone
= ctx
->strLast
;
257 ctx
->strLast
->nextDone
= str
;
260 INSIST(ctx
->strDone
== NULL
);
261 ctx
->strDone
= ctx
->strLast
= str
;
263 evDeselectFD(opaqueCtx
, str
->file
);
264 str
->file
.opaque
= NULL
;
265 /* evDrop() will call evCancelRW() on us. */
268 /* Dribble out some bytes on the stream. (Called by evDispatch().) */
270 writable(evContext opaqueCtx
, void *uap
, int fd
, int evmask
) {
276 bytes
= writev(fd
, str
->iovCur
, str
->iovCurCount
);
278 if ((str
->flags
& EV_STR_TIMEROK
) != 0)
279 evTouchIdleTimer(opaqueCtx
, str
->timer
);
282 if (bytes
< 0 && errno
!= EINTR
) {
284 str
->ioErrno
= errno
;
287 if (str
->ioDone
== -1 || str
->ioDone
== str
->ioTotal
)
288 done(opaqueCtx
, str
);
291 /* Scoop up some bytes from the stream. (Called by evDispatch().) */
293 readable(evContext opaqueCtx
, void *uap
, int fd
, int evmask
) {
299 bytes
= readv(fd
, str
->iovCur
, str
->iovCurCount
);
301 if ((str
->flags
& EV_STR_TIMEROK
) != 0)
302 evTouchIdleTimer(opaqueCtx
, str
->timer
);
308 if (errno
!= EINTR
) {
310 str
->ioErrno
= errno
;
314 if (str
->ioDone
<= 0 || str
->ioDone
== str
->ioTotal
)
315 done(opaqueCtx
, str
);