/* * Copyright (c) 2009-2012 Niels Provos, Nick Mathewson * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions * are met: * 1. Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * 2. Redistributions in binary form must reproduce the above copyright * notice, this list of conditions and the following disclaimer in the * documentation and/or other materials provided with the distribution. * 3. The name of the author may not be used to endorse or promote products * derived from this software without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. */ #include "event2/event-config.h" #include "evconfig-private.h" #include #ifdef _WIN32 #include #endif #include "event2/util.h" #include "event2/buffer.h" #include "event2/bufferevent.h" #include "event2/bufferevent_struct.h" #include "event2/event.h" #include "defer-internal.h" #include "bufferevent-internal.h" #include "mm-internal.h" #include "util-internal.h" struct bufferevent_pair { struct bufferevent_private bev; struct bufferevent_pair *partner; /* For ->destruct() lock checking */ struct bufferevent_pair *unlinked_partner; }; /* Given a bufferevent that's really a bev part of a bufferevent_pair, * return that bufferevent_filtered. Returns NULL otherwise.*/ static inline struct bufferevent_pair * upcast(struct bufferevent *bev) { struct bufferevent_pair *bev_p; if (!BEV_IS_PAIR(bev)) return NULL; bev_p = EVUTIL_UPCAST(bev, struct bufferevent_pair, bev.bev); EVUTIL_ASSERT(BEV_IS_PAIR(&bev_p->bev.bev)); return bev_p; } #define downcast(bev_pair) (&(bev_pair)->bev.bev) static inline void incref_and_lock(struct bufferevent *b) { struct bufferevent_pair *bevp; bufferevent_incref_and_lock_(b); bevp = upcast(b); if (bevp->partner) bufferevent_incref_and_lock_(downcast(bevp->partner)); } static inline void decref_and_unlock(struct bufferevent *b) { struct bufferevent_pair *bevp = upcast(b); if (bevp->partner) bufferevent_decref_and_unlock_(downcast(bevp->partner)); bufferevent_decref_and_unlock_(b); } /* XXX Handle close */ static void be_pair_outbuf_cb(struct evbuffer *, const struct evbuffer_cb_info *, void *); static struct bufferevent_pair * bufferevent_pair_elt_new(struct event_base *base, int options) { struct bufferevent_pair *bufev; if (! (bufev = mm_calloc(1, sizeof(struct bufferevent_pair)))) return NULL; if (bufferevent_init_common_(&bufev->bev, base, &bufferevent_ops_pair, options)) { mm_free(bufev); return NULL; } if (!evbuffer_add_cb(bufev->bev.bev.output, be_pair_outbuf_cb, bufev)) { bufferevent_free(downcast(bufev)); return NULL; } bufferevent_init_generic_timeout_cbs_(&bufev->bev.bev); return bufev; } int bufferevent_pair_new(struct event_base *base, int options, struct bufferevent *pair[2]) { struct bufferevent_pair *bufev1 = NULL, *bufev2 = NULL; int tmp_options; options |= BEV_OPT_DEFER_CALLBACKS; tmp_options = options & ~BEV_OPT_THREADSAFE; bufev1 = bufferevent_pair_elt_new(base, options); if (!bufev1) return -1; bufev2 = bufferevent_pair_elt_new(base, tmp_options); if (!bufev2) { bufferevent_free(downcast(bufev1)); return -1; } if (options & BEV_OPT_THREADSAFE) { /*XXXX check return */ bufferevent_enable_locking_(downcast(bufev2), bufev1->bev.lock); } bufev1->partner = bufev2; bufev2->partner = bufev1; evbuffer_freeze(downcast(bufev1)->input, 0); evbuffer_freeze(downcast(bufev1)->output, 1); evbuffer_freeze(downcast(bufev2)->input, 0); evbuffer_freeze(downcast(bufev2)->output, 1); pair[0] = downcast(bufev1); pair[1] = downcast(bufev2); return 0; } static void be_pair_transfer(struct bufferevent *src, struct bufferevent *dst, int ignore_wm) { size_t dst_size; size_t n; evbuffer_unfreeze(src->output, 1); evbuffer_unfreeze(dst->input, 0); if (dst->wm_read.high) { dst_size = evbuffer_get_length(dst->input); if (dst_size < dst->wm_read.high) { n = dst->wm_read.high - dst_size; evbuffer_remove_buffer(src->output, dst->input, n); } else { if (!ignore_wm) goto done; n = evbuffer_get_length(src->output); evbuffer_add_buffer(dst->input, src->output); } } else { n = evbuffer_get_length(src->output); evbuffer_add_buffer(dst->input, src->output); } if (n) { BEV_RESET_GENERIC_READ_TIMEOUT(dst); if (evbuffer_get_length(dst->output)) BEV_RESET_GENERIC_WRITE_TIMEOUT(dst); else BEV_DEL_GENERIC_WRITE_TIMEOUT(dst); } bufferevent_trigger_nolock_(dst, EV_READ, 0); bufferevent_trigger_nolock_(src, EV_WRITE, 0); done: evbuffer_freeze(src->output, 1); evbuffer_freeze(dst->input, 0); } static inline int be_pair_wants_to_talk(struct bufferevent_pair *src, struct bufferevent_pair *dst) { return (downcast(src)->enabled & EV_WRITE) && (downcast(dst)->enabled & EV_READ) && !dst->bev.read_suspended && evbuffer_get_length(downcast(src)->output); } static void be_pair_outbuf_cb(struct evbuffer *outbuf, const struct evbuffer_cb_info *info, void *arg) { struct bufferevent_pair *bev_pair = arg; struct bufferevent_pair *partner = bev_pair->partner; incref_and_lock(downcast(bev_pair)); if (info->n_added > info->n_deleted && partner) { /* We got more data. If the other side's reading, then hand it over. */ if (be_pair_wants_to_talk(bev_pair, partner)) { be_pair_transfer(downcast(bev_pair), downcast(partner), 0); } } decref_and_unlock(downcast(bev_pair)); } static int be_pair_enable(struct bufferevent *bufev, short events) { struct bufferevent_pair *bev_p = upcast(bufev); struct bufferevent_pair *partner = bev_p->partner; incref_and_lock(bufev); if (events & EV_READ) { BEV_RESET_GENERIC_READ_TIMEOUT(bufev); } if ((events & EV_WRITE) && evbuffer_get_length(bufev->output)) BEV_RESET_GENERIC_WRITE_TIMEOUT(bufev); /* We're starting to read! Does the other side have anything to write?*/ if ((events & EV_READ) && partner && be_pair_wants_to_talk(partner, bev_p)) { be_pair_transfer(downcast(partner), bufev, 0); } /* We're starting to write! Does the other side want to read? */ if ((events & EV_WRITE) && partner && be_pair_wants_to_talk(bev_p, partner)) { be_pair_transfer(bufev, downcast(partner), 0); } decref_and_unlock(bufev); return 0; } static int be_pair_disable(struct bufferevent *bev, short events) { if (events & EV_READ) { BEV_DEL_GENERIC_READ_TIMEOUT(bev); } if (events & EV_WRITE) { BEV_DEL_GENERIC_WRITE_TIMEOUT(bev); } return 0; } static void be_pair_unlink(struct bufferevent *bev) { struct bufferevent_pair *bev_p = upcast(bev); if (bev_p->partner) { bev_p->unlinked_partner = bev_p->partner; bev_p->partner->partner = NULL; bev_p->partner = NULL; } } /* Free *shared* lock in the latest be (since we share it between two of them). */ static void be_pair_destruct(struct bufferevent *bev) { struct bufferevent_pair *bev_p = upcast(bev); /* Transfer ownership of the lock into partner, otherwise we will use * already free'd lock during freeing second bev, see next example: * * bev1->own_lock = 1 * bev2->own_lock = 0 * bev2->lock = bev1->lock * * bufferevent_free(bev1) # refcnt == 0 -> unlink * bufferevent_free(bev2) # refcnt == 0 -> unlink * * event_base_free() -> finilizers -> EVTHREAD_FREE_LOCK(bev1->lock) * -> BEV_LOCK(bev2->lock) <-- already freed * * Where bev1 == pair[0], bev2 == pair[1]. */ if (bev_p->unlinked_partner && bev_p->bev.own_lock) { bev_p->unlinked_partner->bev.own_lock = 1; bev_p->bev.own_lock = 0; } bev_p->unlinked_partner = NULL; } static int be_pair_flush(struct bufferevent *bev, short iotype, enum bufferevent_flush_mode mode) { struct bufferevent_pair *bev_p = upcast(bev); struct bufferevent *partner; if (!bev_p->partner) return -1; if (mode == BEV_NORMAL) return 0; incref_and_lock(bev); partner = downcast(bev_p->partner); if ((iotype & EV_READ) != 0) be_pair_transfer(partner, bev, 1); if ((iotype & EV_WRITE) != 0) be_pair_transfer(bev, partner, 1); if (mode == BEV_FINISHED) { short what = BEV_EVENT_EOF; if (iotype & EV_READ) what |= BEV_EVENT_WRITING; if (iotype & EV_WRITE) what |= BEV_EVENT_READING; bufferevent_run_eventcb_(partner, what, 0); } decref_and_unlock(bev); return 0; } struct bufferevent * bufferevent_pair_get_partner(struct bufferevent *bev) { struct bufferevent_pair *bev_p; struct bufferevent *partner = NULL; bev_p = upcast(bev); if (! bev_p) return NULL; incref_and_lock(bev); if (bev_p->partner) partner = downcast(bev_p->partner); decref_and_unlock(bev); return partner; } const struct bufferevent_ops bufferevent_ops_pair = { "pair_elt", evutil_offsetof(struct bufferevent_pair, bev.bev), be_pair_enable, be_pair_disable, be_pair_unlink, be_pair_destruct, bufferevent_generic_adj_timeouts_, be_pair_flush, NULL, /* ctrl */ };