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
27#include <sys/types.h>
28
29#ifdef WIN32
30#include <winsock2.h>
31#endif
32
33#include "event2/event-config.h"
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};
49
50
51/* Given a bufferevent that's really a bev part of a bufferevent_pair,
52 * return that bufferevent_filtered. Returns NULL otherwise.*/
53static inline struct bufferevent_pair *
54upcast(struct bufferevent *bev)
55{
56	struct bufferevent_pair *bev_p;
57	if (bev->be_ops != &bufferevent_ops_pair)
58		return NULL;
59	bev_p = EVUTIL_UPCAST(bev, struct bufferevent_pair, bev.bev);
60	EVUTIL_ASSERT(bev_p->bev.bev.be_ops == &bufferevent_ops_pair);
61	return bev_p;
62}
63
64#define downcast(bev_pair) (&(bev_pair)->bev.bev)
65
66static inline void
67incref_and_lock(struct bufferevent *b)
68{
69	struct bufferevent_pair *bevp;
70	_bufferevent_incref_and_lock(b);
71	bevp = upcast(b);
72	if (bevp->partner)
73		_bufferevent_incref_and_lock(downcast(bevp->partner));
74}
75
76static inline void
77decref_and_unlock(struct bufferevent *b)
78{
79	struct bufferevent_pair *bevp = upcast(b);
80	if (bevp->partner)
81		_bufferevent_decref_and_unlock(downcast(bevp->partner));
82	_bufferevent_decref_and_unlock(b);
83}
84
85/* XXX Handle close */
86
87static void be_pair_outbuf_cb(struct evbuffer *,
88    const struct evbuffer_cb_info *, void *);
89
90static struct bufferevent_pair *
91bufferevent_pair_elt_new(struct event_base *base,
92    int options)
93{
94	struct bufferevent_pair *bufev;
95	if (! (bufev = mm_calloc(1, sizeof(struct bufferevent_pair))))
96		return NULL;
97	if (bufferevent_init_common(&bufev->bev, base, &bufferevent_ops_pair,
98		options)) {
99		mm_free(bufev);
100		return NULL;
101	}
102	if (!evbuffer_add_cb(bufev->bev.bev.output, be_pair_outbuf_cb, bufev)) {
103		bufferevent_free(downcast(bufev));
104		return NULL;
105	}
106
107	_bufferevent_init_generic_timeout_cbs(&bufev->bev.bev);
108
109	return bufev;
110}
111
112int
113bufferevent_pair_new(struct event_base *base, int options,
114    struct bufferevent *pair[2])
115{
116	struct bufferevent_pair *bufev1 = NULL, *bufev2 = NULL;
117	int tmp_options;
118
119	options |= BEV_OPT_DEFER_CALLBACKS;
120	tmp_options = options & ~BEV_OPT_THREADSAFE;
121
122	bufev1 = bufferevent_pair_elt_new(base, options);
123	if (!bufev1)
124		return -1;
125	bufev2 = bufferevent_pair_elt_new(base, tmp_options);
126	if (!bufev2) {
127		bufferevent_free(downcast(bufev1));
128		return -1;
129	}
130
131	if (options & BEV_OPT_THREADSAFE) {
132		/*XXXX check return */
133		bufferevent_enable_locking(downcast(bufev2), bufev1->bev.lock);
134	}
135
136	bufev1->partner = bufev2;
137	bufev2->partner = bufev1;
138
139	evbuffer_freeze(downcast(bufev1)->input, 0);
140	evbuffer_freeze(downcast(bufev1)->output, 1);
141	evbuffer_freeze(downcast(bufev2)->input, 0);
142	evbuffer_freeze(downcast(bufev2)->output, 1);
143
144	pair[0] = downcast(bufev1);
145	pair[1] = downcast(bufev2);
146
147	return 0;
148}
149
150static void
151be_pair_transfer(struct bufferevent *src, struct bufferevent *dst,
152    int ignore_wm)
153{
154	size_t src_size, dst_size;
155	size_t n;
156
157	evbuffer_unfreeze(src->output, 1);
158	evbuffer_unfreeze(dst->input, 0);
159
160	if (dst->wm_read.high) {
161		dst_size = evbuffer_get_length(dst->input);
162		if (dst_size < dst->wm_read.high) {
163			n = dst->wm_read.high - dst_size;
164			evbuffer_remove_buffer(src->output, dst->input, n);
165		} else {
166			if (!ignore_wm)
167				goto done;
168			n = evbuffer_get_length(src->output);
169			evbuffer_add_buffer(dst->input, src->output);
170		}
171	} else {
172		n = evbuffer_get_length(src->output);
173		evbuffer_add_buffer(dst->input, src->output);
174	}
175
176	if (n) {
177		BEV_RESET_GENERIC_READ_TIMEOUT(dst);
178
179		if (evbuffer_get_length(dst->output))
180			BEV_RESET_GENERIC_WRITE_TIMEOUT(dst);
181		else
182			BEV_DEL_GENERIC_WRITE_TIMEOUT(dst);
183	}
184
185	src_size = evbuffer_get_length(src->output);
186	dst_size = evbuffer_get_length(dst->input);
187
188	if (dst_size >= dst->wm_read.low) {
189		_bufferevent_run_readcb(dst);
190	}
191	if (src_size <= src->wm_write.low) {
192		_bufferevent_run_writecb(src);
193	}
194done:
195	evbuffer_freeze(src->output, 1);
196	evbuffer_freeze(dst->input, 0);
197}
198
199static inline int
200be_pair_wants_to_talk(struct bufferevent_pair *src,
201    struct bufferevent_pair *dst)
202{
203	return (downcast(src)->enabled & EV_WRITE) &&
204	    (downcast(dst)->enabled & EV_READ) &&
205	    !dst->bev.read_suspended &&
206	    evbuffer_get_length(downcast(src)->output);
207}
208
209static void
210be_pair_outbuf_cb(struct evbuffer *outbuf,
211    const struct evbuffer_cb_info *info, void *arg)
212{
213	struct bufferevent_pair *bev_pair = arg;
214	struct bufferevent_pair *partner = bev_pair->partner;
215
216	incref_and_lock(downcast(bev_pair));
217
218	if (info->n_added > info->n_deleted && partner) {
219		/* We got more data.  If the other side's reading, then
220		   hand it over. */
221		if (be_pair_wants_to_talk(bev_pair, partner)) {
222			be_pair_transfer(downcast(bev_pair), downcast(partner), 0);
223		}
224	}
225
226	decref_and_unlock(downcast(bev_pair));
227}
228
229static int
230be_pair_enable(struct bufferevent *bufev, short events)
231{
232	struct bufferevent_pair *bev_p = upcast(bufev);
233	struct bufferevent_pair *partner = bev_p->partner;
234
235	incref_and_lock(bufev);
236
237	if (events & EV_READ) {
238		BEV_RESET_GENERIC_READ_TIMEOUT(bufev);
239	}
240	if ((events & EV_WRITE) && evbuffer_get_length(bufev->output))
241		BEV_RESET_GENERIC_WRITE_TIMEOUT(bufev);
242
243	/* We're starting to read! Does the other side have anything to write?*/
244	if ((events & EV_READ) && partner &&
245	    be_pair_wants_to_talk(partner, bev_p)) {
246		be_pair_transfer(downcast(partner), bufev, 0);
247	}
248	/* We're starting to write! Does the other side want to read? */
249	if ((events & EV_WRITE) && partner &&
250	    be_pair_wants_to_talk(bev_p, partner)) {
251		be_pair_transfer(bufev, downcast(partner), 0);
252	}
253	decref_and_unlock(bufev);
254	return 0;
255}
256
257static int
258be_pair_disable(struct bufferevent *bev, short events)
259{
260	if (events & EV_READ) {
261		BEV_DEL_GENERIC_READ_TIMEOUT(bev);
262	}
263	if (events & EV_WRITE)
264		BEV_DEL_GENERIC_WRITE_TIMEOUT(bev);
265	return 0;
266}
267
268static void
269be_pair_destruct(struct bufferevent *bev)
270{
271	struct bufferevent_pair *bev_p = upcast(bev);
272
273	if (bev_p->partner) {
274		bev_p->partner->partner = NULL;
275		bev_p->partner = NULL;
276	}
277
278	_bufferevent_del_generic_timeout_cbs(bev);
279}
280
281static int
282be_pair_flush(struct bufferevent *bev, short iotype,
283    enum bufferevent_flush_mode mode)
284{
285	struct bufferevent_pair *bev_p = upcast(bev);
286	struct bufferevent *partner;
287	incref_and_lock(bev);
288	if (!bev_p->partner)
289		return -1;
290
291	partner = downcast(bev_p->partner);
292
293	if (mode == BEV_NORMAL)
294		return 0;
295
296	if ((iotype & EV_READ) != 0)
297		be_pair_transfer(partner, bev, 1);
298
299	if ((iotype & EV_WRITE) != 0)
300		be_pair_transfer(bev, partner, 1);
301
302	if (mode == BEV_FINISHED) {
303		_bufferevent_run_eventcb(partner, iotype|BEV_EVENT_EOF);
304	}
305	decref_and_unlock(bev);
306	return 0;
307}
308
309struct bufferevent *
310bufferevent_pair_get_partner(struct bufferevent *bev)
311{
312	struct bufferevent_pair *bev_p;
313	struct bufferevent *partner;
314	bev_p = upcast(bev);
315	if (! bev_p)
316		return NULL;
317
318	incref_and_lock(bev);
319	partner = downcast(bev_p->partner);
320	decref_and_unlock(bev);
321	return partner;
322}
323
324const struct bufferevent_ops bufferevent_ops_pair = {
325	"pair_elt",
326	evutil_offsetof(struct bufferevent_pair, bev.bev),
327	be_pair_enable,
328	be_pair_disable,
329	be_pair_destruct,
330	_bufferevent_generic_adj_timeouts,
331	be_pair_flush,
332	NULL, /* ctrl */
333};
334