1: /*
2: * Copyright (c) 2002-2004 Niels Provos <provos@citi.umich.edu>
3: * All rights reserved.
4: *
5: * Redistribution and use in source and binary forms, with or without
6: * modification, are permitted provided that the following conditions
7: * are met:
8: * 1. Redistributions of source code must retain the above copyright
9: * notice, this list of conditions and the following disclaimer.
10: * 2. Redistributions in binary form must reproduce the above copyright
11: * notice, this list of conditions and the following disclaimer in the
12: * documentation and/or other materials provided with the distribution.
13: * 3. The name of the author may not be used to endorse or promote products
14: * derived from this software without specific prior written permission.
15: *
16: * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
17: * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
18: * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
19: * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
20: * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
21: * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
22: * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
23: * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
24: * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
25: * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
26: */
27:
28: #include <sys/types.h>
29:
30: #ifdef HAVE_CONFIG_H
31: #include "config.h"
32: #endif
33:
34: #ifdef HAVE_SYS_TIME_H
35: #include <sys/time.h>
36: #endif
37:
38: #include <errno.h>
39: #include <stdio.h>
40: #include <stdlib.h>
41: #include <string.h>
42: #ifdef HAVE_STDARG_H
43: #include <stdarg.h>
44: #endif
45:
46: #ifdef WIN32
47: #include <winsock2.h>
48: #endif
49:
50: #include "evutil.h"
51: #include "event.h"
52:
53: /* prototypes */
54:
55: void bufferevent_read_pressure_cb(struct evbuffer *, size_t, size_t, void *);
56:
57: static int
58: bufferevent_add(struct event *ev, int timeout)
59: {
60: struct timeval tv, *ptv = NULL;
61:
62: if (timeout) {
63: evutil_timerclear(&tv);
64: tv.tv_sec = timeout;
65: ptv = &tv;
66: }
67:
68: return (event_add(ev, ptv));
69: }
70:
71: /*
72: * This callback is executed when the size of the input buffer changes.
73: * We use it to apply back pressure on the reading side.
74: */
75:
76: void
77: bufferevent_read_pressure_cb(struct evbuffer *buf, size_t old, size_t now,
78: void *arg) {
79: struct bufferevent *bufev = arg;
80: /*
81: * If we are below the watermark then reschedule reading if it's
82: * still enabled.
83: */
84: if (bufev->wm_read.high == 0 || now < bufev->wm_read.high) {
85: evbuffer_setcb(buf, NULL, NULL);
86:
87: if (bufev->enabled & EV_READ)
88: bufferevent_add(&bufev->ev_read, bufev->timeout_read);
89: }
90: }
91:
92: static void
93: bufferevent_readcb(int fd, short event, void *arg)
94: {
95: struct bufferevent *bufev = arg;
96: int res = 0;
97: short what = EVBUFFER_READ;
98: size_t len;
99: int howmuch = -1;
100:
101: if (event == EV_TIMEOUT) {
102: what |= EVBUFFER_TIMEOUT;
103: goto error;
104: }
105:
106: /*
107: * If we have a high watermark configured then we don't want to
108: * read more data than would make us reach the watermark.
109: */
110: if (bufev->wm_read.high != 0) {
111: howmuch = bufev->wm_read.high - EVBUFFER_LENGTH(bufev->input);
112: /* we might have lowered the watermark, stop reading */
113: if (howmuch <= 0) {
114: struct evbuffer *buf = bufev->input;
115: event_del(&bufev->ev_read);
116: evbuffer_setcb(buf,
117: bufferevent_read_pressure_cb, bufev);
118: return;
119: }
120: }
121:
122: res = evbuffer_read(bufev->input, fd, howmuch);
123: if (res == -1) {
124: if (errno == EAGAIN || errno == EINTR)
125: goto reschedule;
126: /* error case */
127: what |= EVBUFFER_ERROR;
128: } else if (res == 0) {
129: /* eof case */
130: what |= EVBUFFER_EOF;
131: }
132:
133: if (res <= 0)
134: goto error;
135:
136: bufferevent_add(&bufev->ev_read, bufev->timeout_read);
137:
138: /* See if this callbacks meets the water marks */
139: len = EVBUFFER_LENGTH(bufev->input);
140: if (bufev->wm_read.low != 0 && len < bufev->wm_read.low)
141: return;
142: if (bufev->wm_read.high != 0 && len >= bufev->wm_read.high) {
143: struct evbuffer *buf = bufev->input;
144: event_del(&bufev->ev_read);
145:
146: /* Now schedule a callback for us when the buffer changes */
147: evbuffer_setcb(buf, bufferevent_read_pressure_cb, bufev);
148: }
149:
150: /* Invoke the user callback - must always be called last */
151: if (bufev->readcb != NULL)
152: (*bufev->readcb)(bufev, bufev->cbarg);
153: return;
154:
155: reschedule:
156: bufferevent_add(&bufev->ev_read, bufev->timeout_read);
157: return;
158:
159: error:
160: (*bufev->errorcb)(bufev, what, bufev->cbarg);
161: }
162:
163: static void
164: bufferevent_writecb(int fd, short event, void *arg)
165: {
166: struct bufferevent *bufev = arg;
167: int res = 0;
168: short what = EVBUFFER_WRITE;
169:
170: if (event == EV_TIMEOUT) {
171: what |= EVBUFFER_TIMEOUT;
172: goto error;
173: }
174:
175: if (EVBUFFER_LENGTH(bufev->output)) {
176: res = evbuffer_write(bufev->output, fd);
177: if (res == -1) {
178: #ifndef WIN32
179: /*todo. evbuffer uses WriteFile when WIN32 is set. WIN32 system calls do not
180: *set errno. thus this error checking is not portable*/
181: if (errno == EAGAIN ||
182: errno == EINTR ||
183: errno == EINPROGRESS)
184: goto reschedule;
185: /* error case */
186: what |= EVBUFFER_ERROR;
187:
188: #else
189: goto reschedule;
190: #endif
191:
192: } else if (res == 0) {
193: /* eof case */
194: what |= EVBUFFER_EOF;
195: }
196: if (res <= 0)
197: goto error;
198: }
199:
200: if (EVBUFFER_LENGTH(bufev->output) != 0)
201: bufferevent_add(&bufev->ev_write, bufev->timeout_write);
202:
203: /*
204: * Invoke the user callback if our buffer is drained or below the
205: * low watermark.
206: */
207: if (bufev->writecb != NULL &&
208: EVBUFFER_LENGTH(bufev->output) <= bufev->wm_write.low)
209: (*bufev->writecb)(bufev, bufev->cbarg);
210:
211: return;
212:
213: reschedule:
214: if (EVBUFFER_LENGTH(bufev->output) != 0)
215: bufferevent_add(&bufev->ev_write, bufev->timeout_write);
216: return;
217:
218: error:
219: (*bufev->errorcb)(bufev, what, bufev->cbarg);
220: }
221:
222: /*
223: * Create a new buffered event object.
224: *
225: * The read callback is invoked whenever we read new data.
226: * The write callback is invoked whenever the output buffer is drained.
227: * The error callback is invoked on a write/read error or on EOF.
228: *
229: * Both read and write callbacks maybe NULL. The error callback is not
230: * allowed to be NULL and have to be provided always.
231: */
232:
233: struct bufferevent *
234: bufferevent_new(int fd, evbuffercb readcb, evbuffercb writecb,
235: everrorcb errorcb, void *cbarg)
236: {
237: struct bufferevent *bufev;
238:
239: if ((bufev = calloc(1, sizeof(struct bufferevent))) == NULL)
240: return (NULL);
241:
242: if ((bufev->input = evbuffer_new()) == NULL) {
243: free(bufev);
244: return (NULL);
245: }
246:
247: if ((bufev->output = evbuffer_new()) == NULL) {
248: evbuffer_free(bufev->input);
249: free(bufev);
250: return (NULL);
251: }
252:
253: event_set(&bufev->ev_read, fd, EV_READ, bufferevent_readcb, bufev);
254: event_set(&bufev->ev_write, fd, EV_WRITE, bufferevent_writecb, bufev);
255:
256: bufferevent_setcb(bufev, readcb, writecb, errorcb, cbarg);
257:
258: /*
259: * Set to EV_WRITE so that using bufferevent_write is going to
260: * trigger a callback. Reading needs to be explicitly enabled
261: * because otherwise no data will be available.
262: */
263: bufev->enabled = EV_WRITE;
264:
265: return (bufev);
266: }
267:
268: void
269: bufferevent_setcb(struct bufferevent *bufev,
270: evbuffercb readcb, evbuffercb writecb, everrorcb errorcb, void *cbarg)
271: {
272: bufev->readcb = readcb;
273: bufev->writecb = writecb;
274: bufev->errorcb = errorcb;
275:
276: bufev->cbarg = cbarg;
277: }
278:
279: void
280: bufferevent_setfd(struct bufferevent *bufev, int fd)
281: {
282: event_del(&bufev->ev_read);
283: event_del(&bufev->ev_write);
284:
285: event_set(&bufev->ev_read, fd, EV_READ, bufferevent_readcb, bufev);
286: event_set(&bufev->ev_write, fd, EV_WRITE, bufferevent_writecb, bufev);
287: if (bufev->ev_base != NULL) {
288: event_base_set(bufev->ev_base, &bufev->ev_read);
289: event_base_set(bufev->ev_base, &bufev->ev_write);
290: }
291:
292: /* might have to manually trigger event registration */
293: }
294:
295: int
296: bufferevent_priority_set(struct bufferevent *bufev, int priority)
297: {
298: if (event_priority_set(&bufev->ev_read, priority) == -1)
299: return (-1);
300: if (event_priority_set(&bufev->ev_write, priority) == -1)
301: return (-1);
302:
303: return (0);
304: }
305:
306: /* Closing the file descriptor is the responsibility of the caller */
307:
308: void
309: bufferevent_free(struct bufferevent *bufev)
310: {
311: event_del(&bufev->ev_read);
312: event_del(&bufev->ev_write);
313:
314: evbuffer_free(bufev->input);
315: evbuffer_free(bufev->output);
316:
317: free(bufev);
318: }
319:
320: /*
321: * Returns 0 on success;
322: * -1 on failure.
323: */
324:
325: int
326: bufferevent_write(struct bufferevent *bufev, const void *data, size_t size)
327: {
328: int res;
329:
330: res = evbuffer_add(bufev->output, data, size);
331:
332: if (res == -1)
333: return (res);
334:
335: /* If everything is okay, we need to schedule a write */
336: if (size > 0 && (bufev->enabled & EV_WRITE))
337: bufferevent_add(&bufev->ev_write, bufev->timeout_write);
338:
339: return (res);
340: }
341:
342: int
343: bufferevent_write_buffer(struct bufferevent *bufev, struct evbuffer *buf)
344: {
345: int res;
346:
347: res = bufferevent_write(bufev, buf->buffer, buf->off);
348: if (res != -1)
349: evbuffer_drain(buf, buf->off);
350:
351: return (res);
352: }
353:
354: size_t
355: bufferevent_read(struct bufferevent *bufev, void *data, size_t size)
356: {
357: struct evbuffer *buf = bufev->input;
358:
359: if (buf->off < size)
360: size = buf->off;
361:
362: /* Copy the available data to the user buffer */
363: memcpy(data, buf->buffer, size);
364:
365: if (size)
366: evbuffer_drain(buf, size);
367:
368: return (size);
369: }
370:
371: int
372: bufferevent_enable(struct bufferevent *bufev, short event)
373: {
374: if (event & EV_READ) {
375: if (bufferevent_add(&bufev->ev_read, bufev->timeout_read) == -1)
376: return (-1);
377: }
378: if (event & EV_WRITE) {
379: if (bufferevent_add(&bufev->ev_write, bufev->timeout_write) == -1)
380: return (-1);
381: }
382:
383: bufev->enabled |= event;
384: return (0);
385: }
386:
387: int
388: bufferevent_disable(struct bufferevent *bufev, short event)
389: {
390: if (event & EV_READ) {
391: if (event_del(&bufev->ev_read) == -1)
392: return (-1);
393: }
394: if (event & EV_WRITE) {
395: if (event_del(&bufev->ev_write) == -1)
396: return (-1);
397: }
398:
399: bufev->enabled &= ~event;
400: return (0);
401: }
402:
403: /*
404: * Sets the read and write timeout for a buffered event.
405: */
406:
407: void
408: bufferevent_settimeout(struct bufferevent *bufev,
409: int timeout_read, int timeout_write) {
410: bufev->timeout_read = timeout_read;
411: bufev->timeout_write = timeout_write;
412:
413: if (event_pending(&bufev->ev_read, EV_READ, NULL))
414: bufferevent_add(&bufev->ev_read, timeout_read);
415: if (event_pending(&bufev->ev_write, EV_WRITE, NULL))
416: bufferevent_add(&bufev->ev_write, timeout_write);
417: }
418:
419: /*
420: * Sets the water marks
421: */
422:
423: void
424: bufferevent_setwatermark(struct bufferevent *bufev, short events,
425: size_t lowmark, size_t highmark)
426: {
427: if (events & EV_READ) {
428: bufev->wm_read.low = lowmark;
429: bufev->wm_read.high = highmark;
430: }
431:
432: if (events & EV_WRITE) {
433: bufev->wm_write.low = lowmark;
434: bufev->wm_write.high = highmark;
435: }
436:
437: /* If the watermarks changed then see if we should call read again */
438: bufferevent_read_pressure_cb(bufev->input,
439: 0, EVBUFFER_LENGTH(bufev->input), bufev);
440: }
441:
442: int
443: bufferevent_base_set(struct event_base *base, struct bufferevent *bufev)
444: {
445: int res;
446:
447: bufev->ev_base = base;
448:
449: res = event_base_set(base, &bufev->ev_read);
450: if (res == -1)
451: return (res);
452:
453: res = event_base_set(base, &bufev->ev_write);
454: return (res);
455: }
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>
![[BACK]](/icons/cvsweb/back.gif){kind=icon}
Return to evbuffer.c CVS log ![[TXT]](/icons/cvsweb/text.gif){kind=icon}
![[DIR]](/icons/cvsweb/dir.gif){kind=icon}
Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / libevent