/src/libevent/bufferevent_pair.c
Line | Count | Source (jump to first uncovered line) |
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 * |
56 | | upcast(struct bufferevent *bev) |
57 | 3.95k | { |
58 | 3.95k | struct bufferevent_pair *bev_p; |
59 | 3.95k | bev_p = EVUTIL_UPCAST(bev, struct bufferevent_pair, bev.bev); |
60 | 3.95k | EVUTIL_ASSERT(BEV_IS_PAIR(&bev_p->bev.bev)); |
61 | 3.95k | return bev_p; |
62 | 3.95k | } |
63 | | |
64 | 6.75k | #define downcast(bev_pair) (&(bev_pair)->bev.bev) |
65 | | |
66 | | static inline void |
67 | | incref_and_lock(struct bufferevent *b) |
68 | 1.02k | { |
69 | 1.02k | struct bufferevent_pair *bevp; |
70 | 1.02k | bufferevent_incref_and_lock_(b); |
71 | 1.02k | bevp = upcast(b); |
72 | 1.02k | if (bevp->partner) |
73 | 1.01k | bufferevent_incref_and_lock_(downcast(bevp->partner)); |
74 | 1.02k | } |
75 | | |
76 | | static inline void |
77 | | decref_and_unlock(struct bufferevent *b) |
78 | 1.02k | { |
79 | 1.02k | struct bufferevent_pair *bevp = upcast(b); |
80 | 1.02k | if (bevp->partner) |
81 | 1.01k | bufferevent_decref_and_unlock_(downcast(bevp->partner)); |
82 | 1.02k | bufferevent_decref_and_unlock_(b); |
83 | 1.02k | } |
84 | | |
85 | | /* XXX Handle close */ |
86 | | |
87 | | static void be_pair_outbuf_cb(struct evbuffer *, |
88 | | const struct evbuffer_cb_info *, void *); |
89 | | |
90 | | static struct bufferevent_pair * |
91 | | bufferevent_pair_elt_new(struct event_base *base, |
92 | | int options) |
93 | 460 | { |
94 | 460 | struct bufferevent_pair *bufev; |
95 | 460 | if (! (bufev = mm_calloc(1, sizeof(struct bufferevent_pair)))) |
96 | 0 | return NULL; |
97 | 460 | if (bufferevent_init_common_(&bufev->bev, base, &bufferevent_ops_pair, |
98 | 460 | options)) { |
99 | 18 | mm_free(bufev); |
100 | 18 | return NULL; |
101 | 18 | } |
102 | 442 | if (!evbuffer_add_cb(bufev->bev.bev.output, be_pair_outbuf_cb, bufev)) { |
103 | 0 | bufferevent_free(downcast(bufev)); |
104 | 0 | return NULL; |
105 | 0 | } |
106 | | |
107 | 442 | bufferevent_init_generic_timeout_cbs_(&bufev->bev.bev); |
108 | | |
109 | 442 | return bufev; |
110 | 442 | } |
111 | | |
112 | | int |
113 | | bufferevent_pair_new(struct event_base *base, int options, |
114 | | struct bufferevent *pair[2]) |
115 | 239 | { |
116 | 239 | struct bufferevent_pair *bufev1 = NULL, *bufev2 = NULL; |
117 | 239 | int tmp_options; |
118 | | |
119 | 239 | options |= BEV_OPT_DEFER_CALLBACKS; |
120 | 239 | tmp_options = options & ~BEV_OPT_THREADSAFE; |
121 | | |
122 | 239 | bufev1 = bufferevent_pair_elt_new(base, options); |
123 | 239 | if (!bufev1) |
124 | 18 | return -1; |
125 | 221 | bufev2 = bufferevent_pair_elt_new(base, tmp_options); |
126 | 221 | if (!bufev2) { |
127 | 0 | bufferevent_free(downcast(bufev1)); |
128 | 0 | return -1; |
129 | 0 | } |
130 | | |
131 | 221 | if (options & BEV_OPT_THREADSAFE) { |
132 | | /*XXXX check return */ |
133 | 0 | bufferevent_enable_locking_(downcast(bufev2), bufev1->bev.lock); |
134 | 0 | } |
135 | | |
136 | 221 | bufev1->partner = bufev2; |
137 | 221 | bufev2->partner = bufev1; |
138 | | |
139 | 221 | evbuffer_freeze(downcast(bufev1)->input, 0); |
140 | 221 | evbuffer_freeze(downcast(bufev1)->output, 1); |
141 | 221 | evbuffer_freeze(downcast(bufev2)->input, 0); |
142 | 221 | evbuffer_freeze(downcast(bufev2)->output, 1); |
143 | | |
144 | 221 | pair[0] = downcast(bufev1); |
145 | 221 | pair[1] = downcast(bufev2); |
146 | | |
147 | 221 | return 0; |
148 | 221 | } |
149 | | |
150 | | static void |
151 | | be_pair_transfer(struct bufferevent *src, struct bufferevent *dst, |
152 | | int ignore_wm) |
153 | 0 | { |
154 | 0 | size_t dst_size; |
155 | 0 | size_t n; |
156 | |
|
157 | 0 | evbuffer_unfreeze(src->output, 1); |
158 | 0 | evbuffer_unfreeze(dst->input, 0); |
159 | |
|
160 | 0 | if (dst->wm_read.high) { |
161 | 0 | dst_size = evbuffer_get_length(dst->input); |
162 | 0 | if (dst_size < dst->wm_read.high) { |
163 | 0 | n = dst->wm_read.high - dst_size; |
164 | 0 | evbuffer_remove_buffer(src->output, dst->input, n); |
165 | 0 | } else { |
166 | 0 | if (!ignore_wm) |
167 | 0 | goto done; |
168 | 0 | n = evbuffer_get_length(src->output); |
169 | 0 | evbuffer_add_buffer(dst->input, src->output); |
170 | 0 | } |
171 | 0 | } else { |
172 | 0 | n = evbuffer_get_length(src->output); |
173 | 0 | evbuffer_add_buffer(dst->input, src->output); |
174 | 0 | } |
175 | | |
176 | 0 | if (n) { |
177 | 0 | BEV_RESET_GENERIC_READ_TIMEOUT(dst); |
178 | |
|
179 | 0 | if (evbuffer_get_length(dst->output)) |
180 | 0 | BEV_RESET_GENERIC_WRITE_TIMEOUT(dst); |
181 | 0 | else |
182 | 0 | BEV_DEL_GENERIC_WRITE_TIMEOUT(dst); |
183 | 0 | } |
184 | |
|
185 | 0 | bufferevent_trigger_nolock_(dst, EV_READ, 0); |
186 | 0 | bufferevent_trigger_nolock_(src, EV_WRITE, 0); |
187 | 0 | done: |
188 | 0 | evbuffer_freeze(src->output, 1); |
189 | 0 | evbuffer_freeze(dst->input, 0); |
190 | 0 | } |
191 | | |
192 | | static inline int |
193 | | be_pair_wants_to_talk(struct bufferevent_pair *src, |
194 | | struct bufferevent_pair *dst) |
195 | 1.19k | { |
196 | 1.19k | return (downcast(src)->enabled & EV_WRITE) && |
197 | 1.19k | (downcast(dst)->enabled & EV_READ) && |
198 | 1.19k | !dst->bev.read_suspended && |
199 | 1.19k | evbuffer_get_length(downcast(src)->output); |
200 | 1.19k | } |
201 | | |
202 | | static void |
203 | | be_pair_outbuf_cb(struct evbuffer *outbuf, |
204 | | const struct evbuffer_cb_info *info, void *arg) |
205 | 0 | { |
206 | 0 | struct bufferevent_pair *bev_pair = arg; |
207 | 0 | struct bufferevent_pair *partner = bev_pair->partner; |
208 | |
|
209 | 0 | incref_and_lock(downcast(bev_pair)); |
210 | |
|
211 | 0 | if (info->n_added > info->n_deleted && partner) { |
212 | | /* We got more data. If the other side's reading, then |
213 | | hand it over. */ |
214 | 0 | if (be_pair_wants_to_talk(bev_pair, partner)) { |
215 | 0 | be_pair_transfer(downcast(bev_pair), downcast(partner), 0); |
216 | 0 | } |
217 | 0 | } |
218 | |
|
219 | 0 | decref_and_unlock(downcast(bev_pair)); |
220 | 0 | } |
221 | | |
222 | | static int |
223 | | be_pair_enable(struct bufferevent *bufev, short events) |
224 | 802 | { |
225 | 802 | struct bufferevent_pair *bev_p = upcast(bufev); |
226 | 802 | struct bufferevent_pair *partner = bev_p->partner; |
227 | | |
228 | 802 | incref_and_lock(bufev); |
229 | | |
230 | 802 | if (events & EV_READ) { |
231 | 802 | BEV_RESET_GENERIC_READ_TIMEOUT(bufev); |
232 | 802 | } |
233 | 802 | if ((events & EV_WRITE) && evbuffer_get_length(bufev->output)) |
234 | 0 | BEV_RESET_GENERIC_WRITE_TIMEOUT(bufev); |
235 | | |
236 | | /* We're starting to read! Does the other side have anything to write?*/ |
237 | 802 | if ((events & EV_READ) && partner && |
238 | 802 | be_pair_wants_to_talk(partner, bev_p)) { |
239 | 0 | be_pair_transfer(downcast(partner), bufev, 0); |
240 | 0 | } |
241 | | /* We're starting to write! Does the other side want to read? */ |
242 | 802 | if ((events & EV_WRITE) && partner && |
243 | 802 | be_pair_wants_to_talk(bev_p, partner)) { |
244 | 0 | be_pair_transfer(bufev, downcast(partner), 0); |
245 | 0 | } |
246 | 802 | decref_and_unlock(bufev); |
247 | 802 | return 0; |
248 | 802 | } |
249 | | |
250 | | static int |
251 | | be_pair_disable(struct bufferevent *bev, short events) |
252 | 401 | { |
253 | 401 | if (events & EV_READ) { |
254 | 401 | BEV_DEL_GENERIC_READ_TIMEOUT(bev); |
255 | 401 | } |
256 | 401 | if (events & EV_WRITE) { |
257 | 0 | BEV_DEL_GENERIC_WRITE_TIMEOUT(bev); |
258 | 0 | } |
259 | 401 | return 0; |
260 | 401 | } |
261 | | |
262 | | static void |
263 | | be_pair_unlink(struct bufferevent *bev) |
264 | 442 | { |
265 | 442 | struct bufferevent_pair *bev_p = upcast(bev); |
266 | | |
267 | 442 | if (bev_p->partner) { |
268 | 221 | bev_p->unlinked_partner = bev_p->partner; |
269 | 221 | bev_p->partner->partner = NULL; |
270 | 221 | bev_p->partner = NULL; |
271 | 221 | } |
272 | 442 | } |
273 | | |
274 | | /* Free *shared* lock in the latest be (since we share it between two of them). */ |
275 | | static void |
276 | | be_pair_destruct(struct bufferevent *bev) |
277 | 442 | { |
278 | 442 | struct bufferevent_pair *bev_p = upcast(bev); |
279 | | |
280 | | /* Transfer ownership of the lock into partner, otherwise we will use |
281 | | * already free'd lock during freeing second bev, see next example: |
282 | | * |
283 | | * bev1->own_lock = 1 |
284 | | * bev2->own_lock = 0 |
285 | | * bev2->lock = bev1->lock |
286 | | * |
287 | | * bufferevent_free(bev1) # refcnt == 0 -> unlink |
288 | | * bufferevent_free(bev2) # refcnt == 0 -> unlink |
289 | | * |
290 | | * event_base_free() -> finalizers -> EVTHREAD_FREE_LOCK(bev1->lock) |
291 | | * -> BEV_LOCK(bev2->lock) <-- already freed |
292 | | * |
293 | | * Where bev1 == pair[0], bev2 == pair[1]. |
294 | | */ |
295 | 442 | if (bev_p->unlinked_partner && bev_p->bev.own_lock) { |
296 | 0 | bev_p->unlinked_partner->bev.own_lock = 1; |
297 | 0 | bev_p->bev.own_lock = 0; |
298 | 0 | } |
299 | 442 | bev_p->unlinked_partner = NULL; |
300 | 442 | } |
301 | | |
302 | | static int |
303 | | be_pair_flush(struct bufferevent *bev, short iotype, |
304 | | enum bufferevent_flush_mode mode) |
305 | 0 | { |
306 | 0 | struct bufferevent_pair *bev_p = upcast(bev); |
307 | 0 | struct bufferevent *partner; |
308 | |
|
309 | 0 | if (!bev_p->partner) |
310 | 0 | return -1; |
311 | | |
312 | 0 | if (mode == BEV_NORMAL) |
313 | 0 | return 0; |
314 | | |
315 | 0 | incref_and_lock(bev); |
316 | |
|
317 | 0 | partner = downcast(bev_p->partner); |
318 | |
|
319 | 0 | if ((iotype & EV_READ) != 0) |
320 | 0 | be_pair_transfer(partner, bev, 1); |
321 | |
|
322 | 0 | if ((iotype & EV_WRITE) != 0) |
323 | 0 | be_pair_transfer(bev, partner, 1); |
324 | |
|
325 | 0 | if (mode == BEV_FINISHED) { |
326 | 0 | short what = BEV_EVENT_EOF; |
327 | 0 | if (iotype & EV_READ) |
328 | 0 | what |= BEV_EVENT_WRITING; |
329 | 0 | if (iotype & EV_WRITE) |
330 | 0 | what |= BEV_EVENT_READING; |
331 | 0 | bufferevent_run_eventcb_(partner, what, 0); |
332 | 0 | } |
333 | 0 | decref_and_unlock(bev); |
334 | 0 | return 0; |
335 | 0 | } |
336 | | |
337 | | struct bufferevent * |
338 | | bufferevent_pair_get_partner(struct bufferevent *bev) |
339 | 221 | { |
340 | 221 | struct bufferevent_pair *bev_p; |
341 | 221 | struct bufferevent *partner = NULL; |
342 | 221 | if (!BEV_IS_PAIR(bev)) |
343 | 0 | return NULL; |
344 | 221 | bev_p = upcast(bev); |
345 | 221 | incref_and_lock(bev); |
346 | 221 | if (bev_p->partner) |
347 | 221 | partner = downcast(bev_p->partner); |
348 | 221 | decref_and_unlock(bev); |
349 | 221 | return partner; |
350 | 221 | } |
351 | | |
352 | | const struct bufferevent_ops bufferevent_ops_pair = { |
353 | | "pair_elt", |
354 | | evutil_offsetof(struct bufferevent_pair, bev.bev), |
355 | | be_pair_enable, |
356 | | be_pair_disable, |
357 | | be_pair_unlink, |
358 | | be_pair_destruct, |
359 | | bufferevent_generic_adj_timeouts_, |
360 | | be_pair_flush, |
361 | | NULL, /* ctrl */ |
362 | | }; |