• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 
45 struct 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.*/
55 static inline struct bufferevent_pair *
upcast(struct bufferevent * bev)56 upcast(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 
68 static inline void
incref_and_lock(struct bufferevent * b)69 incref_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 
78 static inline void
decref_and_unlock(struct bufferevent * b)79 decref_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 
89 static void be_pair_outbuf_cb(struct evbuffer *,
90     const struct evbuffer_cb_info *, void *);
91 
92 static struct bufferevent_pair *
bufferevent_pair_elt_new(struct event_base * base,int options)93 bufferevent_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 
114 int
bufferevent_pair_new(struct event_base * base,int options,struct bufferevent * pair[2])115 bufferevent_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 
152 static void
be_pair_transfer(struct bufferevent * src,struct bufferevent * dst,int ignore_wm)153 be_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);
189 done:
190 	evbuffer_freeze(src->output, 1);
191 	evbuffer_freeze(dst->input, 0);
192 }
193 
194 static inline int
be_pair_wants_to_talk(struct bufferevent_pair * src,struct bufferevent_pair * dst)195 be_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 
204 static void
be_pair_outbuf_cb(struct evbuffer * outbuf,const struct evbuffer_cb_info * info,void * arg)205 be_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 
224 static int
be_pair_enable(struct bufferevent * bufev,short events)225 be_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 
252 static int
be_pair_disable(struct bufferevent * bev,short events)253 be_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 
264 static void
be_pair_unlink(struct bufferevent * bev)265 be_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). */
277 static void
be_pair_destruct(struct bufferevent * bev)278 be_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 
304 static int
be_pair_flush(struct bufferevent * bev,short iotype,enum bufferevent_flush_mode mode)305 be_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 
311 	if (!bev_p->partner)
312 		return -1;
313 
314 	if (mode == BEV_NORMAL)
315 		return 0;
316 
317 	incref_and_lock(bev);
318 
319 	partner = downcast(bev_p->partner);
320 
321 	if ((iotype & EV_READ) != 0)
322 		be_pair_transfer(partner, bev, 1);
323 
324 	if ((iotype & EV_WRITE) != 0)
325 		be_pair_transfer(bev, partner, 1);
326 
327 	if (mode == BEV_FINISHED) {
328 		short what = BEV_EVENT_EOF;
329 		if (iotype & EV_READ)
330 			what |= BEV_EVENT_WRITING;
331 		if (iotype & EV_WRITE)
332 			what |= BEV_EVENT_READING;
333 		bufferevent_run_eventcb_(partner, what, 0);
334 	}
335 	decref_and_unlock(bev);
336 	return 0;
337 }
338 
339 struct bufferevent *
bufferevent_pair_get_partner(struct bufferevent * bev)340 bufferevent_pair_get_partner(struct bufferevent *bev)
341 {
342 	struct bufferevent_pair *bev_p;
343 	struct bufferevent *partner = NULL;
344 	bev_p = upcast(bev);
345 	if (! bev_p)
346 		return NULL;
347 
348 	incref_and_lock(bev);
349 	if (bev_p->partner)
350 		partner = downcast(bev_p->partner);
351 	decref_and_unlock(bev);
352 	return partner;
353 }
354 
355 const struct bufferevent_ops bufferevent_ops_pair = {
356 	"pair_elt",
357 	evutil_offsetof(struct bufferevent_pair, bev.bev),
358 	be_pair_enable,
359 	be_pair_disable,
360 	be_pair_unlink,
361 	be_pair_destruct,
362 	bufferevent_generic_adj_timeouts_,
363 	be_pair_flush,
364 	NULL, /* ctrl */
365 };
366