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 <sys/types.h>
30
31#include "event2/event-config.h"
32
33#ifdef _EVENT_HAVE_SYS_TIME_H
34#include <sys/time.h>
35#endif
36
37#include <errno.h>
38#include <stdio.h>
39#include <stdlib.h>
40#include <string.h>
41#ifdef _EVENT_HAVE_STDARG_H
42#include <stdarg.h>
43#endif
44
45#ifdef WIN32
46#include <winsock2.h>
47#endif
48
49#include "event2/util.h"
50#include "event2/bufferevent.h"
51#include "event2/buffer.h"
52#include "event2/bufferevent_struct.h"
53#include "event2/event.h"
54#include "log-internal.h"
55#include "mm-internal.h"
56#include "bufferevent-internal.h"
57#include "util-internal.h"
58
59/* prototypes */
60static int be_filter_enable(struct bufferevent *, short);
61static int be_filter_disable(struct bufferevent *, short);
62static void be_filter_destruct(struct bufferevent *);
63
64static void be_filter_readcb(struct bufferevent *, void *);
65static void be_filter_writecb(struct bufferevent *, void *);
66static void be_filter_eventcb(struct bufferevent *, short, void *);
67static int be_filter_flush(struct bufferevent *bufev,
68    short iotype, enum bufferevent_flush_mode mode);
69static int be_filter_ctrl(struct bufferevent *, enum bufferevent_ctrl_op, union bufferevent_ctrl_data *);
70
71static void bufferevent_filtered_outbuf_cb(struct evbuffer *buf,
72    const struct evbuffer_cb_info *info, void *arg);
73
74struct bufferevent_filtered {
75	struct bufferevent_private bev;
76
77	/** The bufferevent that we read/write filtered data from/to. */
78	struct bufferevent *underlying;
79	/** A callback on our outbuf to notice when somebody adds data */
80	struct evbuffer_cb_entry *outbuf_cb;
81	/** True iff we have received an EOF callback from the underlying
82	 * bufferevent. */
83	unsigned got_eof;
84
85	/** Function to free context when we're done. */
86	void (*free_context)(void *);
87	/** Input filter */
88	bufferevent_filter_cb process_in;
89	/** Output filter */
90	bufferevent_filter_cb process_out;
91	/** User-supplied argument to the filters. */
92	void *context;
93};
94
95const struct bufferevent_ops bufferevent_ops_filter = {
96	"filter",
97	evutil_offsetof(struct bufferevent_filtered, bev.bev),
98	be_filter_enable,
99	be_filter_disable,
100	be_filter_destruct,
101	_bufferevent_generic_adj_timeouts,
102	be_filter_flush,
103	be_filter_ctrl,
104};
105
106/* Given a bufferevent that's really the bev filter of a bufferevent_filtered,
107 * return that bufferevent_filtered. Returns NULL otherwise.*/
108static inline struct bufferevent_filtered *
109upcast(struct bufferevent *bev)
110{
111	struct bufferevent_filtered *bev_f;
112	if (bev->be_ops != &bufferevent_ops_filter)
113		return NULL;
114	bev_f = (void*)( ((char*)bev) -
115			 evutil_offsetof(struct bufferevent_filtered, bev.bev));
116	EVUTIL_ASSERT(bev_f->bev.bev.be_ops == &bufferevent_ops_filter);
117	return bev_f;
118}
119
120#define downcast(bev_f) (&(bev_f)->bev.bev)
121
122/** Return 1 iff bevf's underlying bufferevent's output buffer is at or
123 * over its high watermark such that we should not write to it in a given
124 * flush mode. */
125static int
126be_underlying_writebuf_full(struct bufferevent_filtered *bevf,
127    enum bufferevent_flush_mode state)
128{
129	struct bufferevent *u = bevf->underlying;
130	return state == BEV_NORMAL &&
131	    u->wm_write.high &&
132	    evbuffer_get_length(u->output) >= u->wm_write.high;
133}
134
135/** Return 1 if our input buffer is at or over its high watermark such that we
136 * should not write to it in a given flush mode. */
137static int
138be_readbuf_full(struct bufferevent_filtered *bevf,
139    enum bufferevent_flush_mode state)
140{
141	struct bufferevent *bufev = downcast(bevf);
142	return state == BEV_NORMAL &&
143	    bufev->wm_read.high &&
144	    evbuffer_get_length(bufev->input) >= bufev->wm_read.high;
145}
146
147
148/* Filter to use when we're created with a NULL filter. */
149static enum bufferevent_filter_result
150be_null_filter(struct evbuffer *src, struct evbuffer *dst, ev_ssize_t lim,
151	       enum bufferevent_flush_mode state, void *ctx)
152{
153	(void)state;
154	if (evbuffer_remove_buffer(src, dst, lim) == 0)
155		return BEV_OK;
156	else
157		return BEV_ERROR;
158}
159
160struct bufferevent *
161bufferevent_filter_new(struct bufferevent *underlying,
162		       bufferevent_filter_cb input_filter,
163		       bufferevent_filter_cb output_filter,
164		       int options,
165		       void (*free_context)(void *),
166		       void *ctx)
167{
168	struct bufferevent_filtered *bufev_f;
169	int tmp_options = options & ~BEV_OPT_THREADSAFE;
170
171	if (!underlying)
172		return NULL;
173
174	if (!input_filter)
175		input_filter = be_null_filter;
176	if (!output_filter)
177		output_filter = be_null_filter;
178
179	bufev_f = mm_calloc(1, sizeof(struct bufferevent_filtered));
180	if (!bufev_f)
181		return NULL;
182
183	if (bufferevent_init_common(&bufev_f->bev, underlying->ev_base,
184				    &bufferevent_ops_filter, tmp_options) < 0) {
185		mm_free(bufev_f);
186		return NULL;
187	}
188	if (options & BEV_OPT_THREADSAFE) {
189		bufferevent_enable_locking(downcast(bufev_f), NULL);
190	}
191
192	bufev_f->underlying = underlying;
193
194	bufev_f->process_in = input_filter;
195	bufev_f->process_out = output_filter;
196	bufev_f->free_context = free_context;
197	bufev_f->context = ctx;
198
199	bufferevent_setcb(bufev_f->underlying,
200	    be_filter_readcb, be_filter_writecb, be_filter_eventcb, bufev_f);
201
202	bufev_f->outbuf_cb = evbuffer_add_cb(downcast(bufev_f)->output,
203	   bufferevent_filtered_outbuf_cb, bufev_f);
204
205	_bufferevent_init_generic_timeout_cbs(downcast(bufev_f));
206	bufferevent_incref(underlying);
207
208	bufferevent_enable(underlying, EV_READ|EV_WRITE);
209	bufferevent_suspend_read(underlying, BEV_SUSPEND_FILT_READ);
210
211	return downcast(bufev_f);
212}
213
214static void
215be_filter_destruct(struct bufferevent *bev)
216{
217	struct bufferevent_filtered *bevf = upcast(bev);
218	EVUTIL_ASSERT(bevf);
219	if (bevf->free_context)
220		bevf->free_context(bevf->context);
221
222	if (bevf->bev.options & BEV_OPT_CLOSE_ON_FREE) {
223		/* Yes, there is also a decref in bufferevent_decref.
224		 * That decref corresponds to the incref when we set
225		 * underlying for the first time.  This decref is an
226		 * extra one to remove the last reference.
227		 */
228		if (BEV_UPCAST(bevf->underlying)->refcnt < 2) {
229			event_warnx("BEV_OPT_CLOSE_ON_FREE set on an "
230			    "bufferevent with too few references");
231		} else {
232			bufferevent_free(bevf->underlying);
233		}
234	} else {
235		if (bevf->underlying) {
236			if (bevf->underlying->errorcb == be_filter_eventcb)
237				bufferevent_setcb(bevf->underlying,
238				    NULL, NULL, NULL, NULL);
239			bufferevent_unsuspend_read(bevf->underlying,
240			    BEV_SUSPEND_FILT_READ);
241		}
242	}
243
244	_bufferevent_del_generic_timeout_cbs(bev);
245}
246
247static int
248be_filter_enable(struct bufferevent *bev, short event)
249{
250	struct bufferevent_filtered *bevf = upcast(bev);
251	if (event & EV_WRITE)
252		BEV_RESET_GENERIC_WRITE_TIMEOUT(bev);
253
254	if (event & EV_READ) {
255		BEV_RESET_GENERIC_READ_TIMEOUT(bev);
256		bufferevent_unsuspend_read(bevf->underlying,
257		    BEV_SUSPEND_FILT_READ);
258	}
259	return 0;
260}
261
262static int
263be_filter_disable(struct bufferevent *bev, short event)
264{
265	struct bufferevent_filtered *bevf = upcast(bev);
266	if (event & EV_WRITE)
267		BEV_DEL_GENERIC_WRITE_TIMEOUT(bev);
268	if (event & EV_READ) {
269		BEV_DEL_GENERIC_READ_TIMEOUT(bev);
270		bufferevent_suspend_read(bevf->underlying,
271		    BEV_SUSPEND_FILT_READ);
272	}
273	return 0;
274}
275
276static enum bufferevent_filter_result
277be_filter_process_input(struct bufferevent_filtered *bevf,
278			enum bufferevent_flush_mode state,
279			int *processed_out)
280{
281	enum bufferevent_filter_result res;
282	struct bufferevent *bev = downcast(bevf);
283
284	if (state == BEV_NORMAL) {
285		/* If we're in 'normal' mode, don't urge data on the filter
286		 * unless we're reading data and under our high-water mark.*/
287		if (!(bev->enabled & EV_READ) ||
288		    be_readbuf_full(bevf, state))
289			return BEV_OK;
290	}
291
292	do {
293		ev_ssize_t limit = -1;
294		if (state == BEV_NORMAL && bev->wm_read.high)
295			limit = bev->wm_read.high -
296			    evbuffer_get_length(bev->input);
297
298		res = bevf->process_in(bevf->underlying->input,
299		    bev->input, limit, state, bevf->context);
300
301		if (res == BEV_OK)
302			*processed_out = 1;
303	} while (res == BEV_OK &&
304		 (bev->enabled & EV_READ) &&
305		 evbuffer_get_length(bevf->underlying->input) &&
306		 !be_readbuf_full(bevf, state));
307
308	if (*processed_out)
309		BEV_RESET_GENERIC_READ_TIMEOUT(bev);
310
311	return res;
312}
313
314
315static enum bufferevent_filter_result
316be_filter_process_output(struct bufferevent_filtered *bevf,
317			 enum bufferevent_flush_mode state,
318			 int *processed_out)
319{
320	/* Requires references and lock: might call writecb */
321	enum bufferevent_filter_result res = BEV_OK;
322	struct bufferevent *bufev = downcast(bevf);
323	int again = 0;
324
325	if (state == BEV_NORMAL) {
326		/* If we're in 'normal' mode, don't urge data on the
327		 * filter unless we're writing data, and the underlying
328		 * bufferevent is accepting data, and we have data to
329		 * give the filter.  If we're in 'flush' or 'finish',
330		 * call the filter no matter what. */
331		if (!(bufev->enabled & EV_WRITE) ||
332		    be_underlying_writebuf_full(bevf, state) ||
333		    !evbuffer_get_length(bufev->output))
334			return BEV_OK;
335	}
336
337	/* disable the callback that calls this function
338	   when the user adds to the output buffer. */
339	evbuffer_cb_set_flags(bufev->output, bevf->outbuf_cb, 0);
340
341	do {
342		int processed = 0;
343		again = 0;
344
345		do {
346			ev_ssize_t limit = -1;
347			if (state == BEV_NORMAL &&
348			    bevf->underlying->wm_write.high)
349				limit = bevf->underlying->wm_write.high -
350				    evbuffer_get_length(bevf->underlying->output);
351
352			res = bevf->process_out(downcast(bevf)->output,
353			    bevf->underlying->output,
354			    limit,
355			    state,
356			    bevf->context);
357
358			if (res == BEV_OK)
359				processed = *processed_out = 1;
360		} while (/* Stop if the filter wasn't successful...*/
361			res == BEV_OK &&
362			/* Or if we aren't writing any more. */
363			(bufev->enabled & EV_WRITE) &&
364			/* Of if we have nothing more to write and we are
365			 * not flushing. */
366			evbuffer_get_length(bufev->output) &&
367			/* Or if we have filled the underlying output buffer. */
368			!be_underlying_writebuf_full(bevf,state));
369
370		if (processed &&
371		    evbuffer_get_length(bufev->output) <= bufev->wm_write.low) {
372			/* call the write callback.*/
373			_bufferevent_run_writecb(bufev);
374
375			if (res == BEV_OK &&
376			    (bufev->enabled & EV_WRITE) &&
377			    evbuffer_get_length(bufev->output) &&
378			    !be_underlying_writebuf_full(bevf, state)) {
379				again = 1;
380			}
381		}
382	} while (again);
383
384	/* reenable the outbuf_cb */
385	evbuffer_cb_set_flags(bufev->output,bevf->outbuf_cb,
386	    EVBUFFER_CB_ENABLED);
387
388	if (*processed_out)
389		BEV_RESET_GENERIC_WRITE_TIMEOUT(bufev);
390
391	return res;
392}
393
394/* Called when the size of our outbuf changes. */
395static void
396bufferevent_filtered_outbuf_cb(struct evbuffer *buf,
397    const struct evbuffer_cb_info *cbinfo, void *arg)
398{
399	struct bufferevent_filtered *bevf = arg;
400	struct bufferevent *bev = downcast(bevf);
401
402	if (cbinfo->n_added) {
403		int processed_any = 0;
404		/* Somebody added more data to the output buffer. Try to
405		 * process it, if we should. */
406		_bufferevent_incref_and_lock(bev);
407		be_filter_process_output(bevf, BEV_NORMAL, &processed_any);
408		_bufferevent_decref_and_unlock(bev);
409	}
410}
411
412/* Called when the underlying socket has read. */
413static void
414be_filter_readcb(struct bufferevent *underlying, void *_me)
415{
416	struct bufferevent_filtered *bevf = _me;
417	enum bufferevent_filter_result res;
418	enum bufferevent_flush_mode state;
419	struct bufferevent *bufev = downcast(bevf);
420	int processed_any = 0;
421
422	_bufferevent_incref_and_lock(bufev);
423
424	if (bevf->got_eof)
425		state = BEV_FINISHED;
426	else
427		state = BEV_NORMAL;
428
429	/* XXXX use return value */
430	res = be_filter_process_input(bevf, state, &processed_any);
431	(void)res;
432
433	/* XXX This should be in process_input, not here.  There are
434	 * other places that can call process-input, and they should
435	 * force readcb calls as needed. */
436	if (processed_any &&
437	    evbuffer_get_length(bufev->input) >= bufev->wm_read.low)
438		_bufferevent_run_readcb(bufev);
439
440	_bufferevent_decref_and_unlock(bufev);
441}
442
443/* Called when the underlying socket has drained enough that we can write to
444   it. */
445static void
446be_filter_writecb(struct bufferevent *underlying, void *_me)
447{
448	struct bufferevent_filtered *bevf = _me;
449	struct bufferevent *bev = downcast(bevf);
450	int processed_any = 0;
451
452	_bufferevent_incref_and_lock(bev);
453	be_filter_process_output(bevf, BEV_NORMAL, &processed_any);
454	_bufferevent_decref_and_unlock(bev);
455}
456
457/* Called when the underlying socket has given us an error */
458static void
459be_filter_eventcb(struct bufferevent *underlying, short what, void *_me)
460{
461	struct bufferevent_filtered *bevf = _me;
462	struct bufferevent *bev = downcast(bevf);
463
464	_bufferevent_incref_and_lock(bev);
465	/* All we can really to is tell our own eventcb. */
466	_bufferevent_run_eventcb(bev, what);
467	_bufferevent_decref_and_unlock(bev);
468}
469
470static int
471be_filter_flush(struct bufferevent *bufev,
472    short iotype, enum bufferevent_flush_mode mode)
473{
474	struct bufferevent_filtered *bevf = upcast(bufev);
475	int processed_any = 0;
476	EVUTIL_ASSERT(bevf);
477
478	_bufferevent_incref_and_lock(bufev);
479
480	if (iotype & EV_READ) {
481		be_filter_process_input(bevf, mode, &processed_any);
482	}
483	if (iotype & EV_WRITE) {
484		be_filter_process_output(bevf, mode, &processed_any);
485	}
486	/* XXX check the return value? */
487	/* XXX does this want to recursively call lower-level flushes? */
488	bufferevent_flush(bevf->underlying, iotype, mode);
489
490	_bufferevent_decref_and_unlock(bufev);
491
492	return processed_any;
493}
494
495static int
496be_filter_ctrl(struct bufferevent *bev, enum bufferevent_ctrl_op op,
497    union bufferevent_ctrl_data *data)
498{
499	struct bufferevent_filtered *bevf;
500	switch (op) {
501	case BEV_CTRL_GET_UNDERLYING:
502		bevf = upcast(bev);
503		data->ptr = bevf->underlying;
504		return 0;
505	case BEV_CTRL_GET_FD:
506	case BEV_CTRL_SET_FD:
507	case BEV_CTRL_CANCEL_ALL:
508	default:
509		return -1;
510	}
511}
512