bufferevent_pair.c revision 290001 
1/*
2 * Copyright (c) 2009-2012 Niels Provos, Nick Mathewson
3 *
4 * Redistribution and use in source and binary forms, with or without
5 * modification, are permitted provided that the following conditions
6 * are met:
7 * 1. Redistributions of source code must retain the above copyright
8 *    notice, this list of conditions and the following disclaimer.
9 * 2. Redistributions in binary form must reproduce the above copyright
10 *    notice, this list of conditions and the following disclaimer in the
11 *    documentation and/or other materials provided with the distribution.
12 * 3. The name of the author may not be used to endorse or promote products
13 *    derived from this software without specific prior written permission.
14 *
15 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
16 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
17 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
18 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
19 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
20 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
21 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
22 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
23 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
24 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
25 */
26#include "event2/event-config.h"
27#include "evconfig-private.h"
28
29#include <sys/types.h>
30
31#ifdef _WIN32
32#include <winsock2.h>
33#endif
34
35#include "event2/util.h"
36#include "event2/buffer.h"
37#include "event2/bufferevent.h"
38#include "event2/bufferevent_struct.h"
39#include "event2/event.h"
40#include "defer-internal.h"
41#include "bufferevent-internal.h"
42#include "mm-internal.h"
43#include "util-internal.h"
44
45struct bufferevent_pair {
46	struct bufferevent_private bev;
47	struct bufferevent_pair *partner;
48	/* For ->destruct() lock checking */
49	struct bufferevent_pair *unlinked_partner;
50};
51
52
53/* Given a bufferevent that's really a bev part of a bufferevent_pair,
54 * return that bufferevent_filtered. Returns NULL otherwise.*/
55static inline struct bufferevent_pair *
56upcast(struct bufferevent *bev)
57{
58	struct bufferevent_pair *bev_p;
59	if (bev->be_ops != &bufferevent_ops_pair)
60		return NULL;
61	bev_p = EVUTIL_UPCAST(bev, struct bufferevent_pair, bev.bev);
62	EVUTIL_ASSERT(bev_p->bev.bev.be_ops == &bufferevent_ops_pair);
63	return bev_p;
64}
65
66#define downcast(bev_pair) (&(bev_pair)->bev.bev)
67
68static inline void
69incref_and_lock(struct bufferevent *b)
70{
71	struct bufferevent_pair *bevp;
72	bufferevent_incref_and_lock_(b);
73	bevp = upcast(b);
74	if (bevp->partner)
75		bufferevent_incref_and_lock_(downcast(bevp->partner));
76}
77
78static inline void
79decref_and_unlock(struct bufferevent *b)
80{
81	struct bufferevent_pair *bevp = upcast(b);
82	if (bevp->partner)
83		bufferevent_decref_and_unlock_(downcast(bevp->partner));
84	bufferevent_decref_and_unlock_(b);
85}
86
87/* XXX Handle close */
88
89static void be_pair_outbuf_cb(struct evbuffer *,
90    const struct evbuffer_cb_info *, void *);
91
92static struct bufferevent_pair *
93bufferevent_pair_elt_new(struct event_base *base,
94    int options)
95{
96	struct bufferevent_pair *bufev;
97	if (! (bufev = mm_calloc(1, sizeof(struct bufferevent_pair))))
98		return NULL;
99	if (bufferevent_init_common_(&bufev->bev, base, &bufferevent_ops_pair,
100		options)) {
101		mm_free(bufev);
102		return NULL;
103	}
104	if (!evbuffer_add_cb(bufev->bev.bev.output, be_pair_outbuf_cb, bufev)) {
105		bufferevent_free(downcast(bufev));
106		return NULL;
107	}
108
109	bufferevent_init_generic_timeout_cbs_(&bufev->bev.bev);
110
111	return bufev;
112}
113
114int
115bufferevent_pair_new(struct event_base *base, int options,
116    struct bufferevent *pair[2])
117{
118	struct bufferevent_pair *bufev1 = NULL, *bufev2 = NULL;
119	int tmp_options;
120
121	options |= BEV_OPT_DEFER_CALLBACKS;
122	tmp_options = options & ~BEV_OPT_THREADSAFE;
123
124	bufev1 = bufferevent_pair_elt_new(base, options);
125	if (!bufev1)
126		return -1;
127	bufev2 = bufferevent_pair_elt_new(base, tmp_options);
128	if (!bufev2) {
129		bufferevent_free(downcast(bufev1));
130		return -1;
131	}
132
133	if (options & BEV_OPT_THREADSAFE) {
134		/*XXXX check return */
135		bufferevent_enable_locking_(downcast(bufev2), bufev1->bev.lock);
136	}
137
138	bufev1->partner = bufev2;
139	bufev2->partner = bufev1;
140
141	evbuffer_freeze(downcast(bufev1)->input, 0);
142	evbuffer_freeze(downcast(bufev1)->output, 1);
143	evbuffer_freeze(downcast(bufev2)->input, 0);
144	evbuffer_freeze(downcast(bufev2)->output, 1);
145
146	pair[0] = downcast(bufev1);
147	pair[1] = downcast(bufev2);
148
149	return 0;
150}
151
152static void
153be_pair_transfer(struct bufferevent *src, struct bufferevent *dst,
154    int ignore_wm)
155{
156	size_t dst_size;
157	size_t n;
158
159	evbuffer_unfreeze(src->output, 1);
160	evbuffer_unfreeze(dst->input, 0);
161
162	if (dst->wm_read.high) {
163		dst_size = evbuffer_get_length(dst->input);
164		if (dst_size < dst->wm_read.high) {
165			n = dst->wm_read.high - dst_size;
166			evbuffer_remove_buffer(src->output, dst->input, n);
167		} else {
168			if (!ignore_wm)
169				goto done;
170			n = evbuffer_get_length(src->output);
171			evbuffer_add_buffer(dst->input, src->output);
172		}
173	} else {
174		n = evbuffer_get_length(src->output);
175		evbuffer_add_buffer(dst->input, src->output);
176	}
177
178	if (n) {
179		BEV_RESET_GENERIC_READ_TIMEOUT(dst);
180
181		if (evbuffer_get_length(dst->output))
182			BEV_RESET_GENERIC_WRITE_TIMEOUT(dst);
183		else
184			BEV_DEL_GENERIC_WRITE_TIMEOUT(dst);
185	}
186
187	bufferevent_trigger_nolock_(dst, EV_READ, 0);
188	bufferevent_trigger_nolock_(src, EV_WRITE, 0);
189done:
190	evbuffer_freeze(src->output, 1);
191	evbuffer_freeze(dst->input, 0);
192}
193
194static inline int
195be_pair_wants_to_talk(struct bufferevent_pair *src,
196    struct bufferevent_pair *dst)
197{
198	return (downcast(src)->enabled & EV_WRITE) &&
199	    (downcast(dst)->enabled & EV_READ) &&
200	    !dst->bev.read_suspended &&
201	    evbuffer_get_length(downcast(src)->output);
202}
203
204static void
205be_pair_outbuf_cb(struct evbuffer *outbuf,
206    const struct evbuffer_cb_info *info, void *arg)
207{
208	struct bufferevent_pair *bev_pair = arg;
209	struct bufferevent_pair *partner = bev_pair->partner;
210
211	incref_and_lock(downcast(bev_pair));
212
213	if (info->n_added > info->n_deleted && partner) {
214		/* We got more data.  If the other side's reading, then
215		   hand it over. */
216		if (be_pair_wants_to_talk(bev_pair, partner)) {
217			be_pair_transfer(downcast(bev_pair), downcast(partner), 0);
218		}
219	}
220
221	decref_and_unlock(downcast(bev_pair));
222}
223
224static int
225be_pair_enable(struct bufferevent *bufev, short events)
226{
227	struct bufferevent_pair *bev_p = upcast(bufev);
228	struct bufferevent_pair *partner = bev_p->partner;
229
230	incref_and_lock(bufev);
231
232	if (events & EV_READ) {
233		BEV_RESET_GENERIC_READ_TIMEOUT(bufev);
234	}
235	if ((events & EV_WRITE) && evbuffer_get_length(bufev->output))
236		BEV_RESET_GENERIC_WRITE_TIMEOUT(bufev);
237
238	/* We're starting to read! Does the other side have anything to write?*/
239	if ((events & EV_READ) && partner &&
240	    be_pair_wants_to_talk(partner, bev_p)) {
241		be_pair_transfer(downcast(partner), bufev, 0);
242	}
243	/* We're starting to write! Does the other side want to read? */
244	if ((events & EV_WRITE) && partner &&
245	    be_pair_wants_to_talk(bev_p, partner)) {
246		be_pair_transfer(bufev, downcast(partner), 0);
247	}
248	decref_and_unlock(bufev);
249	return 0;
250}
251
252static int
253be_pair_disable(struct bufferevent *bev, short events)
254{
255	if (events & EV_READ) {
256		BEV_DEL_GENERIC_READ_TIMEOUT(bev);
257	}
258	if (events & EV_WRITE) {
259		BEV_DEL_GENERIC_WRITE_TIMEOUT(bev);
260	}
261	return 0;
262}
263
264static void
265be_pair_unlink(struct bufferevent *bev)
266{
267	struct bufferevent_pair *bev_p = upcast(bev);
268
269	if (bev_p->partner) {
270		bev_p->unlinked_partner = bev_p->partner;
271		bev_p->partner->partner = NULL;
272		bev_p->partner = NULL;
273	}
274}
275
276/* Free *shared* lock in the latest be (since we share it between two of them). */
277static void
278be_pair_destruct(struct bufferevent *bev)
279{
280	struct bufferevent_pair *bev_p = upcast(bev);
281
282	/* Transfer ownership of the lock into partner, otherwise we will use
283	 * already free'd lock during freeing second bev, see next example:
284	 *
285	 * bev1->own_lock = 1
286	 * bev2->own_lock = 0
287	 * bev2->lock = bev1->lock
288	 *
289	 * bufferevent_free(bev1) # refcnt == 0 -> unlink
290	 * bufferevent_free(bev2) # refcnt == 0 -> unlink
291	 *
292	 * event_base_free() -> finilizers -> EVTHREAD_FREE_LOCK(bev1->lock)
293	 *                                 -> BEV_LOCK(bev2->lock) <-- already freed
294	 *
295	 * Where bev1 == pair[0], bev2 == pair[1].
296	 */
297	if (bev_p->unlinked_partner && bev_p->bev.own_lock) {
298		bev_p->unlinked_partner->bev.own_lock = 1;
299		bev_p->bev.own_lock = 0;
300	}
301	bev_p->unlinked_partner = NULL;
302}
303
304static int
305be_pair_flush(struct bufferevent *bev, short iotype,
306    enum bufferevent_flush_mode mode)
307{
308	struct bufferevent_pair *bev_p = upcast(bev);
309	struct bufferevent *partner;
310	incref_and_lock(bev);
311	if (!bev_p->partner)
312		return -1;
313
314	partner = downcast(bev_p->partner);
315
316	if (mode == BEV_NORMAL)
317		return 0;
318
319	if ((iotype & EV_READ) != 0)
320		be_pair_transfer(partner, bev, 1);
321
322	if ((iotype & EV_WRITE) != 0)
323		be_pair_transfer(bev, partner, 1);
324
325	if (mode == BEV_FINISHED) {
326		bufferevent_run_eventcb_(partner, iotype|BEV_EVENT_EOF, 0);
327	}
328	decref_and_unlock(bev);
329	return 0;
330}
331
332struct bufferevent *
333bufferevent_pair_get_partner(struct bufferevent *bev)
334{
335	struct bufferevent_pair *bev_p;
336	struct bufferevent *partner = NULL;
337	bev_p = upcast(bev);
338	if (! bev_p)
339		return NULL;
340
341	incref_and_lock(bev);
342	if (bev_p->partner)
343		partner = downcast(bev_p->partner);
344	decref_and_unlock(bev);
345	return partner;
346}
347
348const struct bufferevent_ops bufferevent_ops_pair = {
349	"pair_elt",
350	evutil_offsetof(struct bufferevent_pair, bev.bev),
351	be_pair_enable,
352	be_pair_disable,
353	be_pair_unlink,
354	be_pair_destruct,
355	bufferevent_generic_adj_timeouts_,
356	be_pair_flush,
357	NULL, /* ctrl */
358};
359