summaryrefslogtreecommitdiffstats
path: root/libevent/bufferevent_pair.c
diff options
context:
space:
mode:
Diffstat (limited to 'libevent/bufferevent_pair.c')
-rw-r--r--libevent/bufferevent_pair.c365
1 files changed, 365 insertions, 0 deletions
diff --git a/libevent/bufferevent_pair.c b/libevent/bufferevent_pair.c
new file mode 100644
index 0000000..f88cd75
--- /dev/null
+++ b/libevent/bufferevent_pair.c
@@ -0,0 +1,365 @@
+/*
+ * 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 <sys/types.h>
+
+#ifdef _WIN32
+#include <winsock2.h>
+#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 */
+};