2 * Copyright (c) 2007-2012 Niels Provos and Nick Mathewson
3 * Copyright (c) 2002-2006 Niels Provos <provos@citi.umich.edu>
6 * Redistribution and use in source and binary forms, with or without
7 * modification, are permitted provided that the following conditions
9 * 1. Redistributions of source code must retain the above copyright
10 * notice, this list of conditions and the following disclaimer.
11 * 2. Redistributions in binary form must reproduce the above copyright
12 * notice, this list of conditions and the following disclaimer in the
13 * documentation and/or other materials provided with the distribution.
14 * 3. The name of the author may not be used to endorse or promote products
15 * derived from this software without specific prior written permission.
17 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
18 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
19 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
20 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
21 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
22 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
26 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
29 #include "evconfig-private.h"
31 #include <sys/types.h>
33 #include "event2/event-config.h"
35 #ifdef EVENT__HAVE_SYS_TIME_H
43 #ifdef EVENT__HAVE_STDARG_H
51 #include "event2/util.h"
52 #include "event2/bufferevent.h"
53 #include "event2/buffer.h"
54 #include "event2/bufferevent_struct.h"
55 #include "event2/event.h"
56 #include "log-internal.h"
57 #include "mm-internal.h"
58 #include "bufferevent-internal.h"
59 #include "util-internal.h"
62 static int be_filter_enable(struct bufferevent *, short);
63 static int be_filter_disable(struct bufferevent *, short);
64 static void be_filter_unlink(struct bufferevent *);
65 static void be_filter_destruct(struct bufferevent *);
67 static void be_filter_readcb(struct bufferevent *, void *);
68 static void be_filter_writecb(struct bufferevent *, void *);
69 static void be_filter_eventcb(struct bufferevent *, short, void *);
70 static int be_filter_flush(struct bufferevent *bufev,
71 short iotype, enum bufferevent_flush_mode mode);
72 static int be_filter_ctrl(struct bufferevent *, enum bufferevent_ctrl_op, union bufferevent_ctrl_data *);
74 static void bufferevent_filtered_inbuf_cb(struct evbuffer *buf,
75 const struct evbuffer_cb_info *cbinfo, void *arg);
77 static void bufferevent_filtered_outbuf_cb(struct evbuffer *buf,
78 const struct evbuffer_cb_info *info, void *arg);
80 struct bufferevent_filtered {
81 struct bufferevent_private bev;
83 /** The bufferevent that we read/write filtered data from/to. */
84 struct bufferevent *underlying;
85 /** A callback on our inbuf to notice somebory removes data */
86 struct evbuffer_cb_entry *inbuf_cb;
87 /** A callback on our outbuf to notice when somebody adds data */
88 struct evbuffer_cb_entry *outbuf_cb;
89 /** True iff we have received an EOF callback from the underlying
93 /** Function to free context when we're done. */
94 void (*free_context)(void *);
96 bufferevent_filter_cb process_in;
98 bufferevent_filter_cb process_out;
99 /** User-supplied argument to the filters. */
103 const struct bufferevent_ops bufferevent_ops_filter = {
105 evutil_offsetof(struct bufferevent_filtered, bev.bev),
110 bufferevent_generic_adj_timeouts_,
115 /* Given a bufferevent that's really the bev filter of a bufferevent_filtered,
116 * return that bufferevent_filtered. Returns NULL otherwise.*/
117 static inline struct bufferevent_filtered *
118 upcast(struct bufferevent *bev)
120 struct bufferevent_filtered *bev_f;
121 if (bev->be_ops != &bufferevent_ops_filter)
123 bev_f = (void*)( ((char*)bev) -
124 evutil_offsetof(struct bufferevent_filtered, bev.bev));
125 EVUTIL_ASSERT(bev_f->bev.bev.be_ops == &bufferevent_ops_filter);
129 #define downcast(bev_f) (&(bev_f)->bev.bev)
131 /** Return 1 iff bevf's underlying bufferevent's output buffer is at or
132 * over its high watermark such that we should not write to it in a given
135 be_underlying_writebuf_full(struct bufferevent_filtered *bevf,
136 enum bufferevent_flush_mode state)
138 struct bufferevent *u = bevf->underlying;
139 return state == BEV_NORMAL &&
141 evbuffer_get_length(u->output) >= u->wm_write.high;
144 /** Return 1 if our input buffer is at or over its high watermark such that we
145 * should not write to it in a given flush mode. */
147 be_readbuf_full(struct bufferevent_filtered *bevf,
148 enum bufferevent_flush_mode state)
150 struct bufferevent *bufev = downcast(bevf);
151 return state == BEV_NORMAL &&
152 bufev->wm_read.high &&
153 evbuffer_get_length(bufev->input) >= bufev->wm_read.high;
157 /* Filter to use when we're created with a NULL filter. */
158 static enum bufferevent_filter_result
159 be_null_filter(struct evbuffer *src, struct evbuffer *dst, ev_ssize_t lim,
160 enum bufferevent_flush_mode state, void *ctx)
163 if (evbuffer_remove_buffer(src, dst, lim) == 0)
170 bufferevent_filter_new(struct bufferevent *underlying,
171 bufferevent_filter_cb input_filter,
172 bufferevent_filter_cb output_filter,
174 void (*free_context)(void *),
177 struct bufferevent_filtered *bufev_f;
178 int tmp_options = options & ~BEV_OPT_THREADSAFE;
184 input_filter = be_null_filter;
186 output_filter = be_null_filter;
188 bufev_f = mm_calloc(1, sizeof(struct bufferevent_filtered));
192 if (bufferevent_init_common_(&bufev_f->bev, underlying->ev_base,
193 &bufferevent_ops_filter, tmp_options) < 0) {
197 if (options & BEV_OPT_THREADSAFE) {
198 bufferevent_enable_locking_(downcast(bufev_f), NULL);
201 bufev_f->underlying = underlying;
203 bufev_f->process_in = input_filter;
204 bufev_f->process_out = output_filter;
205 bufev_f->free_context = free_context;
206 bufev_f->context = ctx;
208 bufferevent_setcb(bufev_f->underlying,
209 be_filter_readcb, be_filter_writecb, be_filter_eventcb, bufev_f);
211 bufev_f->inbuf_cb = evbuffer_add_cb(downcast(bufev_f)->input,
212 bufferevent_filtered_inbuf_cb, bufev_f);
213 evbuffer_cb_clear_flags(downcast(bufev_f)->input, bufev_f->inbuf_cb,
214 EVBUFFER_CB_ENABLED);
216 bufev_f->outbuf_cb = evbuffer_add_cb(downcast(bufev_f)->output,
217 bufferevent_filtered_outbuf_cb, bufev_f);
219 bufferevent_init_generic_timeout_cbs_(downcast(bufev_f));
220 bufferevent_incref_(underlying);
222 bufferevent_enable(underlying, EV_READ|EV_WRITE);
223 bufferevent_suspend_read_(underlying, BEV_SUSPEND_FILT_READ);
225 return downcast(bufev_f);
229 be_filter_unlink(struct bufferevent *bev)
231 struct bufferevent_filtered *bevf = upcast(bev);
234 if (bevf->bev.options & BEV_OPT_CLOSE_ON_FREE) {
235 /* Yes, there is also a decref in bufferevent_decref_.
236 * That decref corresponds to the incref when we set
237 * underlying for the first time. This decref is an
238 * extra one to remove the last reference.
240 if (BEV_UPCAST(bevf->underlying)->refcnt < 2) {
241 event_warnx("BEV_OPT_CLOSE_ON_FREE set on an "
242 "bufferevent with too few references");
244 bufferevent_free(bevf->underlying);
247 if (bevf->underlying) {
248 if (bevf->underlying->errorcb == be_filter_eventcb)
249 bufferevent_setcb(bevf->underlying,
250 NULL, NULL, NULL, NULL);
251 bufferevent_unsuspend_read_(bevf->underlying,
252 BEV_SUSPEND_FILT_READ);
258 be_filter_destruct(struct bufferevent *bev)
260 struct bufferevent_filtered *bevf = upcast(bev);
262 if (bevf->free_context)
263 bevf->free_context(bevf->context);
266 evbuffer_remove_cb_entry(bev->input, bevf->inbuf_cb);
269 evbuffer_remove_cb_entry(bev->output, bevf->outbuf_cb);
273 be_filter_enable(struct bufferevent *bev, short event)
275 struct bufferevent_filtered *bevf = upcast(bev);
276 if (event & EV_WRITE)
277 BEV_RESET_GENERIC_WRITE_TIMEOUT(bev);
279 if (event & EV_READ) {
280 BEV_RESET_GENERIC_READ_TIMEOUT(bev);
281 bufferevent_unsuspend_read_(bevf->underlying,
282 BEV_SUSPEND_FILT_READ);
288 be_filter_disable(struct bufferevent *bev, short event)
290 struct bufferevent_filtered *bevf = upcast(bev);
291 if (event & EV_WRITE)
292 BEV_DEL_GENERIC_WRITE_TIMEOUT(bev);
293 if (event & EV_READ) {
294 BEV_DEL_GENERIC_READ_TIMEOUT(bev);
295 bufferevent_suspend_read_(bevf->underlying,
296 BEV_SUSPEND_FILT_READ);
301 static enum bufferevent_filter_result
302 be_filter_process_input(struct bufferevent_filtered *bevf,
303 enum bufferevent_flush_mode state,
306 enum bufferevent_filter_result res;
307 struct bufferevent *bev = downcast(bevf);
309 if (state == BEV_NORMAL) {
310 /* If we're in 'normal' mode, don't urge data on the filter
311 * unless we're reading data and under our high-water mark.*/
312 if (!(bev->enabled & EV_READ) ||
313 be_readbuf_full(bevf, state))
318 ev_ssize_t limit = -1;
319 if (state == BEV_NORMAL && bev->wm_read.high)
320 limit = bev->wm_read.high -
321 evbuffer_get_length(bev->input);
323 res = bevf->process_in(bevf->underlying->input,
324 bev->input, limit, state, bevf->context);
328 } while (res == BEV_OK &&
329 (bev->enabled & EV_READ) &&
330 evbuffer_get_length(bevf->underlying->input) &&
331 !be_readbuf_full(bevf, state));
334 BEV_RESET_GENERIC_READ_TIMEOUT(bev);
340 static enum bufferevent_filter_result
341 be_filter_process_output(struct bufferevent_filtered *bevf,
342 enum bufferevent_flush_mode state,
345 /* Requires references and lock: might call writecb */
346 enum bufferevent_filter_result res = BEV_OK;
347 struct bufferevent *bufev = downcast(bevf);
350 if (state == BEV_NORMAL) {
351 /* If we're in 'normal' mode, don't urge data on the
352 * filter unless we're writing data, and the underlying
353 * bufferevent is accepting data, and we have data to
354 * give the filter. If we're in 'flush' or 'finish',
355 * call the filter no matter what. */
356 if (!(bufev->enabled & EV_WRITE) ||
357 be_underlying_writebuf_full(bevf, state) ||
358 !evbuffer_get_length(bufev->output))
362 /* disable the callback that calls this function
363 when the user adds to the output buffer. */
364 evbuffer_cb_clear_flags(bufev->output, bevf->outbuf_cb,
365 EVBUFFER_CB_ENABLED);
372 ev_ssize_t limit = -1;
373 if (state == BEV_NORMAL &&
374 bevf->underlying->wm_write.high)
375 limit = bevf->underlying->wm_write.high -
376 evbuffer_get_length(bevf->underlying->output);
378 res = bevf->process_out(downcast(bevf)->output,
379 bevf->underlying->output,
385 processed = *processed_out = 1;
386 } while (/* Stop if the filter wasn't successful...*/
388 /* Or if we aren't writing any more. */
389 (bufev->enabled & EV_WRITE) &&
390 /* Of if we have nothing more to write and we are
392 evbuffer_get_length(bufev->output) &&
393 /* Or if we have filled the underlying output buffer. */
394 !be_underlying_writebuf_full(bevf,state));
397 /* call the write callback.*/
398 bufferevent_trigger_nolock_(bufev, EV_WRITE, 0);
401 (bufev->enabled & EV_WRITE) &&
402 evbuffer_get_length(bufev->output) &&
403 !be_underlying_writebuf_full(bevf, state)) {
409 /* reenable the outbuf_cb */
410 evbuffer_cb_set_flags(bufev->output,bevf->outbuf_cb,
411 EVBUFFER_CB_ENABLED);
414 BEV_RESET_GENERIC_WRITE_TIMEOUT(bufev);
419 /* Called when the size of our outbuf changes. */
421 bufferevent_filtered_outbuf_cb(struct evbuffer *buf,
422 const struct evbuffer_cb_info *cbinfo, void *arg)
424 struct bufferevent_filtered *bevf = arg;
425 struct bufferevent *bev = downcast(bevf);
427 if (cbinfo->n_added) {
428 int processed_any = 0;
429 /* Somebody added more data to the output buffer. Try to
430 * process it, if we should. */
431 bufferevent_incref_and_lock_(bev);
432 be_filter_process_output(bevf, BEV_NORMAL, &processed_any);
433 bufferevent_decref_and_unlock_(bev);
438 be_filter_read_nolock_(struct bufferevent *underlying, void *me_)
440 struct bufferevent_filtered *bevf = me_;
441 enum bufferevent_filter_result res;
442 enum bufferevent_flush_mode state;
443 struct bufferevent *bufev = downcast(bevf);
444 struct bufferevent_private *bufev_private = BEV_UPCAST(bufev);
445 int processed_any = 0;
447 // It's possible our refcount is 0 at this point if another thread free'd our filterevent
448 EVUTIL_ASSERT(bufev_private->refcnt >= 0);
450 // If our refcount is > 0
451 if (bufev_private->refcnt > 0) {
454 state = BEV_FINISHED;
458 /* XXXX use return value */
459 res = be_filter_process_input(bevf, state, &processed_any);
462 /* XXX This should be in process_input, not here. There are
463 * other places that can call process-input, and they should
464 * force readcb calls as needed. */
466 bufferevent_trigger_nolock_(bufev, EV_READ, 0);
467 if (evbuffer_get_length(underlying->input) > 0 &&
468 be_readbuf_full(bevf, state)) {
469 /* data left in underlying buffer and filter input buffer
470 * hit its read high watermark.
471 * Schedule callback to avoid data gets stuck in underlying
474 evbuffer_cb_set_flags(bufev->input, bevf->inbuf_cb,
475 EVBUFFER_CB_ENABLED);
481 /* Called when the size of our inbuf changes. */
483 bufferevent_filtered_inbuf_cb(struct evbuffer *buf,
484 const struct evbuffer_cb_info *cbinfo, void *arg)
486 struct bufferevent_filtered *bevf = arg;
487 enum bufferevent_flush_mode state;
488 struct bufferevent *bev = downcast(bevf);
493 state = BEV_FINISHED;
498 if (!be_readbuf_full(bevf, state)) {
499 /* opportunity to read data which was left in underlying
500 * input buffer because filter input buffer hit read
503 evbuffer_cb_clear_flags(bev->input, bevf->inbuf_cb,
504 EVBUFFER_CB_ENABLED);
505 if (evbuffer_get_length(bevf->underlying->input) > 0)
506 be_filter_read_nolock_(bevf->underlying, bevf);
512 /* Called when the underlying socket has read. */
514 be_filter_readcb(struct bufferevent *underlying, void *me_)
516 struct bufferevent_filtered *bevf = me_;
517 struct bufferevent *bev = downcast(bevf);
521 be_filter_read_nolock_(underlying, me_);
526 /* Called when the underlying socket has drained enough that we can write to
529 be_filter_writecb(struct bufferevent *underlying, void *me_)
531 struct bufferevent_filtered *bevf = me_;
532 struct bufferevent *bev = downcast(bevf);
533 struct bufferevent_private *bufev_private = BEV_UPCAST(bev);
534 int processed_any = 0;
538 // It's possible our refcount is 0 at this point if another thread free'd our filterevent
539 EVUTIL_ASSERT(bufev_private->refcnt >= 0);
541 // If our refcount is > 0
542 if (bufev_private->refcnt > 0) {
543 be_filter_process_output(bevf, BEV_NORMAL, &processed_any);
549 /* Called when the underlying socket has given us an error */
551 be_filter_eventcb(struct bufferevent *underlying, short what, void *me_)
553 struct bufferevent_filtered *bevf = me_;
554 struct bufferevent *bev = downcast(bevf);
555 struct bufferevent_private *bufev_private = BEV_UPCAST(bev);
559 // It's possible our refcount is 0 at this point if another thread free'd our filterevent
560 EVUTIL_ASSERT(bufev_private->refcnt >= 0);
562 // If our refcount is > 0
563 if (bufev_private->refcnt > 0) {
565 /* All we can really to is tell our own eventcb. */
566 bufferevent_run_eventcb_(bev, what, 0);
573 be_filter_flush(struct bufferevent *bufev,
574 short iotype, enum bufferevent_flush_mode mode)
576 struct bufferevent_filtered *bevf = upcast(bufev);
577 int processed_any = 0;
580 bufferevent_incref_and_lock_(bufev);
582 if (iotype & EV_READ) {
583 be_filter_process_input(bevf, mode, &processed_any);
585 if (iotype & EV_WRITE) {
586 be_filter_process_output(bevf, mode, &processed_any);
588 /* XXX check the return value? */
589 /* XXX does this want to recursively call lower-level flushes? */
590 bufferevent_flush(bevf->underlying, iotype, mode);
592 bufferevent_decref_and_unlock_(bufev);
594 return processed_any;
598 be_filter_ctrl(struct bufferevent *bev, enum bufferevent_ctrl_op op,
599 union bufferevent_ctrl_data *data)
601 struct bufferevent_filtered *bevf;
603 case BEV_CTRL_GET_UNDERLYING:
605 data->ptr = bevf->underlying;
607 case BEV_CTRL_SET_FD:
610 if (bevf->underlying &&
611 bevf->underlying->be_ops &&
612 bevf->underlying->be_ops->ctrl) {
613 return (bevf->underlying->be_ops->ctrl)(bevf->underlying, op, data);
616 case BEV_CTRL_GET_FD:
617 case BEV_CTRL_CANCEL_ALL: