1 /*
2  * Copyright (c) 2007-2012 Niels Provos and Nick Mathewson
3  * Copyright (c) 2002-2006 Niels Provos <provos@citi.umich.edu>
4  * All rights reserved.
5  *
6  * Redistribution and use in source and binary forms, with or without
7  * modification, are permitted provided that the following conditions
8  * are met:
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.
16  *
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.
27  */
28 
29 #include "evconfig-private.h"
30 
31 #include <sys/types.h>
32 
33 #include "event2/event-config.h"
34 
35 #ifdef EVENT__HAVE_SYS_TIME_H
36 #include <sys/time.h>
37 #endif
38 
39 #include <errno.h>
40 #include <stdio.h>
41 #include <stdlib.h>
42 #include <string.h>
43 #ifdef EVENT__HAVE_STDARG_H
44 #include <stdarg.h>
45 #endif
46 
47 #ifdef _WIN32
48 #include <winsock2.h>
49 #endif
50 
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"
60 
61 /* prototypes */
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 *);
66 
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 *);
73 
74 static void bufferevent_filtered_inbuf_cb(struct evbuffer *buf,
75     const struct evbuffer_cb_info *cbinfo, void *arg);
76 
77 static void bufferevent_filtered_outbuf_cb(struct evbuffer *buf,
78     const struct evbuffer_cb_info *info, void *arg);
79 
80 struct bufferevent_filtered {
81 	struct bufferevent_private bev;
82 
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
90 	 * bufferevent. */
91 	unsigned got_eof;
92 
93 	/** Function to free context when we're done. */
94 	void (*free_context)(void *);
95 	/** Input filter */
96 	bufferevent_filter_cb process_in;
97 	/** Output filter */
98 	bufferevent_filter_cb process_out;
99 	/** User-supplied argument to the filters. */
100 	void *context;
101 };
102 
103 const struct bufferevent_ops bufferevent_ops_filter = {
104 	"filter",
105 	evutil_offsetof(struct bufferevent_filtered, bev.bev),
106 	be_filter_enable,
107 	be_filter_disable,
108 	be_filter_unlink,
109 	be_filter_destruct,
110 	bufferevent_generic_adj_timeouts_,
111 	be_filter_flush,
112 	be_filter_ctrl,
113 };
114 
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 *
upcast(struct bufferevent * bev)118 upcast(struct bufferevent *bev)
119 {
120 	struct bufferevent_filtered *bev_f;
121 	if (bev->be_ops != &bufferevent_ops_filter)
122 		return NULL;
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);
126 	return bev_f;
127 }
128 
129 #define downcast(bev_f) (&(bev_f)->bev.bev)
130 
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
133  * flush mode. */
134 static int
be_underlying_writebuf_full(struct bufferevent_filtered * bevf,enum bufferevent_flush_mode state)135 be_underlying_writebuf_full(struct bufferevent_filtered *bevf,
136     enum bufferevent_flush_mode state)
137 {
138 	struct bufferevent *u = bevf->underlying;
139 	return state == BEV_NORMAL &&
140 	    u->wm_write.high &&
141 	    evbuffer_get_length(u->output) >= u->wm_write.high;
142 }
143 
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. */
146 static int
be_readbuf_full(struct bufferevent_filtered * bevf,enum bufferevent_flush_mode state)147 be_readbuf_full(struct bufferevent_filtered *bevf,
148     enum bufferevent_flush_mode state)
149 {
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;
154 }
155 
156 
157 /* Filter to use when we're created with a NULL filter. */
158 static enum bufferevent_filter_result
be_null_filter(struct evbuffer * src,struct evbuffer * dst,ev_ssize_t lim,enum bufferevent_flush_mode state,void * ctx)159 be_null_filter(struct evbuffer *src, struct evbuffer *dst, ev_ssize_t lim,
160 	       enum bufferevent_flush_mode state, void *ctx)
161 {
162 	(void)state;
163 	if (evbuffer_remove_buffer(src, dst, lim) == 0)
164 		return BEV_OK;
165 	else
166 		return BEV_ERROR;
167 }
168 
169 struct bufferevent *
bufferevent_filter_new(struct bufferevent * underlying,bufferevent_filter_cb input_filter,bufferevent_filter_cb output_filter,int options,void (* free_context)(void *),void * ctx)170 bufferevent_filter_new(struct bufferevent *underlying,
171 		       bufferevent_filter_cb input_filter,
172 		       bufferevent_filter_cb output_filter,
173 		       int options,
174 		       void (*free_context)(void *),
175 		       void *ctx)
176 {
177 	struct bufferevent_filtered *bufev_f;
178 	int tmp_options = options & ~BEV_OPT_THREADSAFE;
179 
180 	if (!underlying)
181 		return NULL;
182 
183 	if (!input_filter)
184 		input_filter = be_null_filter;
185 	if (!output_filter)
186 		output_filter = be_null_filter;
187 
188 	bufev_f = mm_calloc(1, sizeof(struct bufferevent_filtered));
189 	if (!bufev_f)
190 		return NULL;
191 
192 	if (bufferevent_init_common_(&bufev_f->bev, underlying->ev_base,
193 				    &bufferevent_ops_filter, tmp_options) < 0) {
194 		mm_free(bufev_f);
195 		return NULL;
196 	}
197 	if (options & BEV_OPT_THREADSAFE) {
198 		bufferevent_enable_locking_(downcast(bufev_f), NULL);
199 	}
200 
201 	bufev_f->underlying = underlying;
202 
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;
207 
208 	bufferevent_setcb(bufev_f->underlying,
209 	    be_filter_readcb, be_filter_writecb, be_filter_eventcb, bufev_f);
210 
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);
215 
216 	bufev_f->outbuf_cb = evbuffer_add_cb(downcast(bufev_f)->output,
217 	   bufferevent_filtered_outbuf_cb, bufev_f);
218 
219 	bufferevent_init_generic_timeout_cbs_(downcast(bufev_f));
220 	bufferevent_incref_(underlying);
221 
222 	bufferevent_enable(underlying, EV_READ|EV_WRITE);
223 	bufferevent_suspend_read_(underlying, BEV_SUSPEND_FILT_READ);
224 
225 	return downcast(bufev_f);
226 }
227 
228 static void
be_filter_unlink(struct bufferevent * bev)229 be_filter_unlink(struct bufferevent *bev)
230 {
231 	struct bufferevent_filtered *bevf = upcast(bev);
232 	EVUTIL_ASSERT(bevf);
233 
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.
239 		 */
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");
243 		} else {
244 			bufferevent_free(bevf->underlying);
245 		}
246 	} else {
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);
253 		}
254 	}
255 }
256 
257 static void
be_filter_destruct(struct bufferevent * bev)258 be_filter_destruct(struct bufferevent *bev)
259 {
260 	struct bufferevent_filtered *bevf = upcast(bev);
261 	EVUTIL_ASSERT(bevf);
262 	if (bevf->free_context)
263 		bevf->free_context(bevf->context);
264 
265 	if (bevf->inbuf_cb)
266 		evbuffer_remove_cb_entry(bev->input, bevf->inbuf_cb);
267 
268 	if (bevf->outbuf_cb)
269 		evbuffer_remove_cb_entry(bev->output, bevf->outbuf_cb);
270 }
271 
272 static int
be_filter_enable(struct bufferevent * bev,short event)273 be_filter_enable(struct bufferevent *bev, short event)
274 {
275 	struct bufferevent_filtered *bevf = upcast(bev);
276 	if (event & EV_WRITE)
277 		BEV_RESET_GENERIC_WRITE_TIMEOUT(bev);
278 
279 	if (event & EV_READ) {
280 		BEV_RESET_GENERIC_READ_TIMEOUT(bev);
281 		bufferevent_unsuspend_read_(bevf->underlying,
282 		    BEV_SUSPEND_FILT_READ);
283 	}
284 	return 0;
285 }
286 
287 static int
be_filter_disable(struct bufferevent * bev,short event)288 be_filter_disable(struct bufferevent *bev, short event)
289 {
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);
297 	}
298 	return 0;
299 }
300 
301 static enum bufferevent_filter_result
be_filter_process_input(struct bufferevent_filtered * bevf,enum bufferevent_flush_mode state,int * processed_out)302 be_filter_process_input(struct bufferevent_filtered *bevf,
303 			enum bufferevent_flush_mode state,
304 			int *processed_out)
305 {
306 	enum bufferevent_filter_result res;
307 	struct bufferevent *bev = downcast(bevf);
308 
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))
314 			return BEV_OK;
315 	}
316 
317 	do {
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);
322 
323 		res = bevf->process_in(bevf->underlying->input,
324 		    bev->input, limit, state, bevf->context);
325 
326 		if (res == BEV_OK)
327 			*processed_out = 1;
328 	} while (res == BEV_OK &&
329 		 (bev->enabled & EV_READ) &&
330 		 evbuffer_get_length(bevf->underlying->input) &&
331 		 !be_readbuf_full(bevf, state));
332 
333 	if (*processed_out)
334 		BEV_RESET_GENERIC_READ_TIMEOUT(bev);
335 
336 	return res;
337 }
338 
339 
340 static enum bufferevent_filter_result
be_filter_process_output(struct bufferevent_filtered * bevf,enum bufferevent_flush_mode state,int * processed_out)341 be_filter_process_output(struct bufferevent_filtered *bevf,
342 			 enum bufferevent_flush_mode state,
343 			 int *processed_out)
344 {
345 	/* Requires references and lock: might call writecb */
346 	enum bufferevent_filter_result res = BEV_OK;
347 	struct bufferevent *bufev = downcast(bevf);
348 	int again = 0;
349 
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))
359 			return BEV_OK;
360 	}
361 
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);
366 
367 	do {
368 		int processed = 0;
369 		again = 0;
370 
371 		do {
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);
377 
378 			res = bevf->process_out(downcast(bevf)->output,
379 			    bevf->underlying->output,
380 			    limit,
381 			    state,
382 			    bevf->context);
383 
384 			if (res == BEV_OK)
385 				processed = *processed_out = 1;
386 		} while (/* Stop if the filter wasn't successful...*/
387 			res == BEV_OK &&
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
391 			 * not flushing. */
392 			evbuffer_get_length(bufev->output) &&
393 			/* Or if we have filled the underlying output buffer. */
394 			!be_underlying_writebuf_full(bevf,state));
395 
396 		if (processed) {
397 			/* call the write callback.*/
398 			bufferevent_trigger_nolock_(bufev, EV_WRITE, 0);
399 
400 			if (res == BEV_OK &&
401 			    (bufev->enabled & EV_WRITE) &&
402 			    evbuffer_get_length(bufev->output) &&
403 			    !be_underlying_writebuf_full(bevf, state)) {
404 				again = 1;
405 			}
406 		}
407 	} while (again);
408 
409 	/* reenable the outbuf_cb */
410 	evbuffer_cb_set_flags(bufev->output,bevf->outbuf_cb,
411 	    EVBUFFER_CB_ENABLED);
412 
413 	if (*processed_out)
414 		BEV_RESET_GENERIC_WRITE_TIMEOUT(bufev);
415 
416 	return res;
417 }
418 
419 /* Called when the size of our outbuf changes. */
420 static void
bufferevent_filtered_outbuf_cb(struct evbuffer * buf,const struct evbuffer_cb_info * cbinfo,void * arg)421 bufferevent_filtered_outbuf_cb(struct evbuffer *buf,
422     const struct evbuffer_cb_info *cbinfo, void *arg)
423 {
424 	struct bufferevent_filtered *bevf = arg;
425 	struct bufferevent *bev = downcast(bevf);
426 
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);
434 	}
435 }
436 
437 static void
be_filter_read_nolock_(struct bufferevent * underlying,void * me_)438 be_filter_read_nolock_(struct bufferevent *underlying, void *me_)
439 {
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;
446 
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);
449 
450 	// If our refcount is > 0
451 	if (bufev_private->refcnt > 0) {
452 
453 		if (bevf->got_eof)
454 			state = BEV_FINISHED;
455 		else
456 			state = BEV_NORMAL;
457 
458 		/* XXXX use return value */
459 		res = be_filter_process_input(bevf, state, &processed_any);
460 		(void)res;
461 
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. */
465 		if (processed_any) {
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
472 				 * input buffer.
473 				 */
474 				evbuffer_cb_set_flags(bufev->input, bevf->inbuf_cb,
475 					EVBUFFER_CB_ENABLED);
476 			}
477 		}
478 	}
479 }
480 
481 /* Called when the size of our inbuf changes. */
482 static void
bufferevent_filtered_inbuf_cb(struct evbuffer * buf,const struct evbuffer_cb_info * cbinfo,void * arg)483 bufferevent_filtered_inbuf_cb(struct evbuffer *buf,
484     const struct evbuffer_cb_info *cbinfo, void *arg)
485 {
486 	struct bufferevent_filtered *bevf = arg;
487 	enum bufferevent_flush_mode state;
488 	struct bufferevent *bev = downcast(bevf);
489 
490 	BEV_LOCK(bev);
491 
492 	if (bevf->got_eof)
493 		state = BEV_FINISHED;
494 	else
495 		state = BEV_NORMAL;
496 
497 
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
501 		 * high watermark.
502 		 */
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);
507 	}
508 
509 	BEV_UNLOCK(bev);
510 }
511 
512 /* Called when the underlying socket has read. */
513 static void
be_filter_readcb(struct bufferevent * underlying,void * me_)514 be_filter_readcb(struct bufferevent *underlying, void *me_)
515 {
516 	struct bufferevent_filtered *bevf = me_;
517 	struct bufferevent *bev = downcast(bevf);
518 
519 	BEV_LOCK(bev);
520 
521 	be_filter_read_nolock_(underlying, me_);
522 
523 	BEV_UNLOCK(bev);
524 }
525 
526 /* Called when the underlying socket has drained enough that we can write to
527    it. */
528 static void
be_filter_writecb(struct bufferevent * underlying,void * me_)529 be_filter_writecb(struct bufferevent *underlying, void *me_)
530 {
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;
535 
536 	BEV_LOCK(bev);
537 
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);
540 
541 	// If our refcount is > 0
542 	if (bufev_private->refcnt > 0) {
543 		be_filter_process_output(bevf, BEV_NORMAL, &processed_any);
544 	}
545 
546 	BEV_UNLOCK(bev);
547 }
548 
549 /* Called when the underlying socket has given us an error */
550 static void
be_filter_eventcb(struct bufferevent * underlying,short what,void * me_)551 be_filter_eventcb(struct bufferevent *underlying, short what, void *me_)
552 {
553 	struct bufferevent_filtered *bevf = me_;
554 	struct bufferevent *bev = downcast(bevf);
555 	struct bufferevent_private *bufev_private = BEV_UPCAST(bev);
556 
557 	BEV_LOCK(bev);
558 
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);
561 
562 	// If our refcount is > 0
563 	if (bufev_private->refcnt > 0) {
564 
565 		/* All we can really to is tell our own eventcb. */
566 		bufferevent_run_eventcb_(bev, what, 0);
567 	}
568 
569 	BEV_UNLOCK(bev);
570 }
571 
572 static int
be_filter_flush(struct bufferevent * bufev,short iotype,enum bufferevent_flush_mode mode)573 be_filter_flush(struct bufferevent *bufev,
574     short iotype, enum bufferevent_flush_mode mode)
575 {
576 	struct bufferevent_filtered *bevf = upcast(bufev);
577 	int processed_any = 0;
578 	EVUTIL_ASSERT(bevf);
579 
580 	bufferevent_incref_and_lock_(bufev);
581 
582 	if (iotype & EV_READ) {
583 		be_filter_process_input(bevf, mode, &processed_any);
584 	}
585 	if (iotype & EV_WRITE) {
586 		be_filter_process_output(bevf, mode, &processed_any);
587 	}
588 	/* XXX check the return value? */
589 	/* XXX does this want to recursively call lower-level flushes? */
590 	bufferevent_flush(bevf->underlying, iotype, mode);
591 
592 	bufferevent_decref_and_unlock_(bufev);
593 
594 	return processed_any;
595 }
596 
597 static int
be_filter_ctrl(struct bufferevent * bev,enum bufferevent_ctrl_op op,union bufferevent_ctrl_data * data)598 be_filter_ctrl(struct bufferevent *bev, enum bufferevent_ctrl_op op,
599     union bufferevent_ctrl_data *data)
600 {
601 	struct bufferevent_filtered *bevf;
602 	switch (op) {
603 	case BEV_CTRL_GET_UNDERLYING:
604 		bevf = upcast(bev);
605 		data->ptr = bevf->underlying;
606 		return 0;
607 	case BEV_CTRL_SET_FD:
608 		bevf = upcast(bev);
609 
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);
614 		}
615 
616 	case BEV_CTRL_GET_FD:
617 	case BEV_CTRL_CANCEL_ALL:
618 	default:
619 		return -1;
620 	}
621 
622 	return -1;
623 }
624