libevent/bufferevent_pair.c

   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
  45 struct 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.*/
  53 static inline struct bufferevent_pair *
  54 upcast(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
  66 static inline void
  67 incref_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
  76 static inline void
  77 decref_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
  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 {
  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
 112 int
 113 bufferevent_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
 150 static void
 151 be_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         }
 194 done:
 195         evbuffer_freeze(src->output, 1);
 196         evbuffer_freeze(dst->input, 0);
 197 }
 198
 199 static inline int
 200 be_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
 209 static void
 210 be_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
 229 static int
 230 be_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
 257 static int
 258 be_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
 268 static void
 269 be_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
 281 static int
 282 be_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
 309 struct bufferevent *
 310 bufferevent_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
 324 const 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 };