libevent/bufferevent_filter.c

   1 /*
   2  * Copyright (c) 2007-2012 Niels Provos and Nick Mathewson
   3  * Copyright (c) 2002-2006 Niels Provos <provos@citi.umich.edu>
   4  * All rights reserved.
   5  *
   6  * Redistribution and use in source and binary forms, with or without
   7  * modification, are permitted provided that the following conditions
   8  * are met:
   9  * 1. Redistributions of source code must retain the above copyright
  10  *    notice, this list of conditions and the following disclaimer.
  11  * 2. Redistributions in binary form must reproduce the above copyright
  12  *    notice, this list of conditions and the following disclaimer in the
  13  *    documentation and/or other materials provided with the distribution.
  14  * 3. The name of the author may not be used to endorse or promote products
  15  *    derived from this software without specific prior written permission.
  16  *
  17  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
  18  * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
  19  * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
  20  * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
  21  * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
  22  * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  23  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  24  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  25  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
  26  * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  27  */
  28
  29 #include <sys/types.h>
  30
  31 #include "event2/event-config.h"
  32
  33 #ifdef _EVENT_HAVE_SYS_TIME_H
  34 #include <sys/time.h>
  35 #endif
  36
  37 #include <errno.h>
  38 #include <stdio.h>
  39 #include <stdlib.h>
  40 #include <string.h>
  41 #ifdef _EVENT_HAVE_STDARG_H
  42 #include <stdarg.h>
  43 #endif
  44
  45 #ifdef WIN32
  46 #include <winsock2.h>
  47 #endif
  48
  49 #include "event2/util.h"
  50 #include "event2/bufferevent.h"
  51 #include "event2/buffer.h"
  52 #include "event2/bufferevent_struct.h"
  53 #include "event2/event.h"
  54 #include "log-internal.h"
  55 #include "mm-internal.h"
  56 #include "bufferevent-internal.h"
  57 #include "util-internal.h"
  58
  59 /* prototypes */
  60 static int be_filter_enable(struct bufferevent *, short);
  61 static int be_filter_disable(struct bufferevent *, short);
  62 static void be_filter_destruct(struct bufferevent *);
  63
  64 static void be_filter_readcb(struct bufferevent *, void *);
  65 static void be_filter_writecb(struct bufferevent *, void *);
  66 static void be_filter_eventcb(struct bufferevent *, short, void *);
  67 static int be_filter_flush(struct bufferevent *bufev,
  68     short iotype, enum bufferevent_flush_mode mode);
  69 static int be_filter_ctrl(struct bufferevent *, enum bufferevent_ctrl_op, union bufferevent_ctrl_data *);
  70
  71 static void bufferevent_filtered_outbuf_cb(struct evbuffer *buf,
  72     const struct evbuffer_cb_info *info, void *arg);
  73
  74 struct bufferevent_filtered {
  75         struct bufferevent_private bev;
  76
  77         /** The bufferevent that we read/write filtered data from/to. */
  78         struct bufferevent *underlying;
  79         /** A callback on our outbuf to notice when somebody adds data */
  80         struct evbuffer_cb_entry *outbuf_cb;
  81         /** True iff we have received an EOF callback from the underlying
  82          * bufferevent. */
  83         unsigned got_eof;
  84
  85         /** Function to free context when we're done. */
  86         void (*free_context)(void *);
  87         /** Input filter */
  88         bufferevent_filter_cb process_in;
  89         /** Output filter */
  90         bufferevent_filter_cb process_out;
  91         /** User-supplied argument to the filters. */
  92         void *context;
  93 };
  94
  95 const struct bufferevent_ops bufferevent_ops_filter = {
  96         "filter",
  97         evutil_offsetof(struct bufferevent_filtered, bev.bev),
  98         be_filter_enable,
  99         be_filter_disable,
 100         be_filter_destruct,
 101         _bufferevent_generic_adj_timeouts,
 102         be_filter_flush,
 103         be_filter_ctrl,
 104 };
 105
 106 /* Given a bufferevent that's really the bev filter of a bufferevent_filtered,
 107  * return that bufferevent_filtered. Returns NULL otherwise.*/
 108 static inline struct bufferevent_filtered *
 109 upcast(struct bufferevent *bev)
 110 {
 111         struct bufferevent_filtered *bev_f;
 112         if (bev->be_ops != &bufferevent_ops_filter)
 113                 return NULL;
 114         bev_f = (void*)( ((char*)bev) -
 115                          evutil_offsetof(struct bufferevent_filtered, bev.bev));
 116         EVUTIL_ASSERT(bev_f->bev.bev.be_ops == &bufferevent_ops_filter);
 117         return bev_f;
 118 }
 119
 120 #define downcast(bev_f) (&(bev_f)->bev.bev)
 121
 122 /** Return 1 iff bevf's underlying bufferevent's output buffer is at or
 123  * over its high watermark such that we should not write to it in a given
 124  * flush mode. */
 125 static int
 126 be_underlying_writebuf_full(struct bufferevent_filtered *bevf,
 127     enum bufferevent_flush_mode state)
 128 {
 129         struct bufferevent *u = bevf->underlying;
 130         return state == BEV_NORMAL &&
 131             u->wm_write.high &&
 132             evbuffer_get_length(u->output) >= u->wm_write.high;
 133 }
 134
 135 /** Return 1 if our input buffer is at or over its high watermark such that we
 136  * should not write to it in a given flush mode. */
 137 static int
 138 be_readbuf_full(struct bufferevent_filtered *bevf,
 139     enum bufferevent_flush_mode state)
 140 {
 141         struct bufferevent *bufev = downcast(bevf);
 142         return state == BEV_NORMAL &&
 143             bufev->wm_read.high &&
 144             evbuffer_get_length(bufev->input) >= bufev->wm_read.high;
 145 }
 146
 147
 148 /* Filter to use when we're created with a NULL filter. */
 149 static enum bufferevent_filter_result
 150 be_null_filter(struct evbuffer *src, struct evbuffer *dst, ev_ssize_t lim,
 151                enum bufferevent_flush_mode state, void *ctx)
 152 {
 153         (void)state;
 154         if (evbuffer_remove_buffer(src, dst, lim) == 0)
 155                 return BEV_OK;
 156         else
 157                 return BEV_ERROR;
 158 }
 159
 160 struct bufferevent *
 161 bufferevent_filter_new(struct bufferevent *underlying,
 162                        bufferevent_filter_cb input_filter,
 163                        bufferevent_filter_cb output_filter,
 164                        int options,
 165                        void (*free_context)(void *),
 166                        void *ctx)
 167 {
 168         struct bufferevent_filtered *bufev_f;
 169         int tmp_options = options & ~BEV_OPT_THREADSAFE;
 170
 171         if (!underlying)
 172                 return NULL;
 173
 174         if (!input_filter)
 175                 input_filter = be_null_filter;
 176         if (!output_filter)
 177                 output_filter = be_null_filter;
 178
 179         bufev_f = mm_calloc(1, sizeof(struct bufferevent_filtered));
 180         if (!bufev_f)
 181                 return NULL;
 182
 183         if (bufferevent_init_common(&bufev_f->bev, underlying->ev_base,
 184                                     &bufferevent_ops_filter, tmp_options) < 0) {
 185                 mm_free(bufev_f);
 186                 return NULL;
 187         }
 188         if (options & BEV_OPT_THREADSAFE) {
 189                 bufferevent_enable_locking(downcast(bufev_f), NULL);
 190         }
 191
 192         bufev_f->underlying = underlying;
 193
 194         bufev_f->process_in = input_filter;
 195         bufev_f->process_out = output_filter;
 196         bufev_f->free_context = free_context;
 197         bufev_f->context = ctx;
 198
 199         bufferevent_setcb(bufev_f->underlying,
 200             be_filter_readcb, be_filter_writecb, be_filter_eventcb, bufev_f);
 201
 202         bufev_f->outbuf_cb = evbuffer_add_cb(downcast(bufev_f)->output,
 203            bufferevent_filtered_outbuf_cb, bufev_f);
 204
 205         _bufferevent_init_generic_timeout_cbs(downcast(bufev_f));
 206         bufferevent_incref(underlying);
 207
 208         bufferevent_enable(underlying, EV_READ|EV_WRITE);
 209         bufferevent_suspend_read(underlying, BEV_SUSPEND_FILT_READ);
 210
 211         return downcast(bufev_f);
 212 }
 213
 214 static void
 215 be_filter_destruct(struct bufferevent *bev)
 216 {
 217         struct bufferevent_filtered *bevf = upcast(bev);
 218         EVUTIL_ASSERT(bevf);
 219         if (bevf->free_context)
 220                 bevf->free_context(bevf->context);
 221
 222         if (bevf->bev.options & BEV_OPT_CLOSE_ON_FREE) {
 223                 /* Yes, there is also a decref in bufferevent_decref.
 224                  * That decref corresponds to the incref when we set
 225                  * underlying for the first time.  This decref is an
 226                  * extra one to remove the last reference.
 227                  */
 228                 if (BEV_UPCAST(bevf->underlying)->refcnt < 2) {
 229                         event_warnx("BEV_OPT_CLOSE_ON_FREE set on an "
 230                             "bufferevent with too few references");
 231                 } else {
 232                         bufferevent_free(bevf->underlying);
 233                 }
 234         } else {
 235                 if (bevf->underlying) {
 236                         if (bevf->underlying->errorcb == be_filter_eventcb)
 237                                 bufferevent_setcb(bevf->underlying,
 238                                     NULL, NULL, NULL, NULL);
 239                         bufferevent_unsuspend_read(bevf->underlying,
 240                             BEV_SUSPEND_FILT_READ);
 241                 }
 242         }
 243
 244         _bufferevent_del_generic_timeout_cbs(bev);
 245 }
 246
 247 static int
 248 be_filter_enable(struct bufferevent *bev, short event)
 249 {
 250         struct bufferevent_filtered *bevf = upcast(bev);
 251         if (event & EV_WRITE)
 252                 BEV_RESET_GENERIC_WRITE_TIMEOUT(bev);
 253
 254         if (event & EV_READ) {
 255                 BEV_RESET_GENERIC_READ_TIMEOUT(bev);
 256                 bufferevent_unsuspend_read(bevf->underlying,
 257                     BEV_SUSPEND_FILT_READ);
 258         }
 259         return 0;
 260 }
 261
 262 static int
 263 be_filter_disable(struct bufferevent *bev, short event)
 264 {
 265         struct bufferevent_filtered *bevf = upcast(bev);
 266         if (event & EV_WRITE)
 267                 BEV_DEL_GENERIC_WRITE_TIMEOUT(bev);
 268         if (event & EV_READ) {
 269                 BEV_DEL_GENERIC_READ_TIMEOUT(bev);
 270                 bufferevent_suspend_read(bevf->underlying,
 271                     BEV_SUSPEND_FILT_READ);
 272         }
 273         return 0;
 274 }
 275
 276 static enum bufferevent_filter_result
 277 be_filter_process_input(struct bufferevent_filtered *bevf,
 278                         enum bufferevent_flush_mode state,
 279                         int *processed_out)
 280 {
 281         enum bufferevent_filter_result res;
 282         struct bufferevent *bev = downcast(bevf);
 283
 284         if (state == BEV_NORMAL) {
 285                 /* If we're in 'normal' mode, don't urge data on the filter
 286                  * unless we're reading data and under our high-water mark.*/
 287                 if (!(bev->enabled & EV_READ) ||
 288                     be_readbuf_full(bevf, state))
 289                         return BEV_OK;
 290         }
 291
 292         do {
 293                 ev_ssize_t limit = -1;
 294                 if (state == BEV_NORMAL && bev->wm_read.high)
 295                         limit = bev->wm_read.high -
 296                             evbuffer_get_length(bev->input);
 297
 298                 res = bevf->process_in(bevf->underlying->input,
 299                     bev->input, limit, state, bevf->context);
 300
 301                 if (res == BEV_OK)
 302                         *processed_out = 1;
 303         } while (res == BEV_OK &&
 304                  (bev->enabled & EV_READ) &&
 305                  evbuffer_get_length(bevf->underlying->input) &&
 306                  !be_readbuf_full(bevf, state));
 307
 308         if (*processed_out)
 309                 BEV_RESET_GENERIC_READ_TIMEOUT(bev);
 310
 311         return res;
 312 }
 313
 314
 315 static enum bufferevent_filter_result
 316 be_filter_process_output(struct bufferevent_filtered *bevf,
 317                          enum bufferevent_flush_mode state,
 318                          int *processed_out)
 319 {
 320         /* Requires references and lock: might call writecb */
 321         enum bufferevent_filter_result res = BEV_OK;
 322         struct bufferevent *bufev = downcast(bevf);
 323         int again = 0;
 324
 325         if (state == BEV_NORMAL) {
 326                 /* If we're in 'normal' mode, don't urge data on the
 327                  * filter unless we're writing data, and the underlying
 328                  * bufferevent is accepting data, and we have data to
 329                  * give the filter.  If we're in 'flush' or 'finish',
 330                  * call the filter no matter what. */
 331                 if (!(bufev->enabled & EV_WRITE) ||
 332                     be_underlying_writebuf_full(bevf, state) ||
 333                     !evbuffer_get_length(bufev->output))
 334                         return BEV_OK;
 335         }
 336
 337         /* disable the callback that calls this function
 338            when the user adds to the output buffer. */
 339         evbuffer_cb_set_flags(bufev->output, bevf->outbuf_cb, 0);
 340
 341         do {
 342                 int processed = 0;
 343                 again = 0;
 344
 345                 do {
 346                         ev_ssize_t limit = -1;
 347                         if (state == BEV_NORMAL &&
 348                             bevf->underlying->wm_write.high)
 349                                 limit = bevf->underlying->wm_write.high -
 350                                     evbuffer_get_length(bevf->underlying->output);
 351
 352                         res = bevf->process_out(downcast(bevf)->output,
 353                             bevf->underlying->output,
 354                             limit,
 355                             state,
 356                             bevf->context);
 357
 358                         if (res == BEV_OK)
 359                                 processed = *processed_out = 1;
 360                 } while (/* Stop if the filter wasn't successful...*/
 361                         res == BEV_OK &&
 362                         /* Or if we aren't writing any more. */
 363                         (bufev->enabled & EV_WRITE) &&
 364                         /* Of if we have nothing more to write and we are
 365                          * not flushing. */
 366                         evbuffer_get_length(bufev->output) &&
 367                         /* Or if we have filled the underlying output buffer. */
 368                         !be_underlying_writebuf_full(bevf,state));
 369
 370                 if (processed &&
 371                     evbuffer_get_length(bufev->output) <= bufev->wm_write.low) {
 372                         /* call the write callback.*/
 373                         _bufferevent_run_writecb(bufev);
 374
 375                         if (res == BEV_OK &&
 376                             (bufev->enabled & EV_WRITE) &&
 377                             evbuffer_get_length(bufev->output) &&
 378                             !be_underlying_writebuf_full(bevf, state)) {
 379                                 again = 1;
 380                         }
 381                 }
 382         } while (again);
 383
 384         /* reenable the outbuf_cb */
 385         evbuffer_cb_set_flags(bufev->output,bevf->outbuf_cb,
 386             EVBUFFER_CB_ENABLED);
 387
 388         if (*processed_out)
 389                 BEV_RESET_GENERIC_WRITE_TIMEOUT(bufev);
 390
 391         return res;
 392 }
 393
 394 /* Called when the size of our outbuf changes. */
 395 static void
 396 bufferevent_filtered_outbuf_cb(struct evbuffer *buf,
 397     const struct evbuffer_cb_info *cbinfo, void *arg)
 398 {
 399         struct bufferevent_filtered *bevf = arg;
 400         struct bufferevent *bev = downcast(bevf);
 401
 402         if (cbinfo->n_added) {
 403                 int processed_any = 0;
 404                 /* Somebody added more data to the output buffer. Try to
 405                  * process it, if we should. */
 406                 _bufferevent_incref_and_lock(bev);
 407                 be_filter_process_output(bevf, BEV_NORMAL, &processed_any);
 408                 _bufferevent_decref_and_unlock(bev);
 409         }
 410 }
 411
 412 /* Called when the underlying socket has read. */
 413 static void
 414 be_filter_readcb(struct bufferevent *underlying, void *_me)
 415 {
 416         struct bufferevent_filtered *bevf = _me;
 417         enum bufferevent_filter_result res;
 418         enum bufferevent_flush_mode state;
 419         struct bufferevent *bufev = downcast(bevf);
 420         int processed_any = 0;
 421
 422         _bufferevent_incref_and_lock(bufev);
 423
 424         if (bevf->got_eof)
 425                 state = BEV_FINISHED;
 426         else
 427                 state = BEV_NORMAL;
 428
 429         /* XXXX use return value */
 430         res = be_filter_process_input(bevf, state, &processed_any);
 431         (void)res;
 432
 433         /* XXX This should be in process_input, not here.  There are
 434          * other places that can call process-input, and they should
 435          * force readcb calls as needed. */
 436         if (processed_any &&
 437             evbuffer_get_length(bufev->input) >= bufev->wm_read.low)
 438                 _bufferevent_run_readcb(bufev);
 439
 440         _bufferevent_decref_and_unlock(bufev);
 441 }
 442
 443 /* Called when the underlying socket has drained enough that we can write to
 444    it. */
 445 static void
 446 be_filter_writecb(struct bufferevent *underlying, void *_me)
 447 {
 448         struct bufferevent_filtered *bevf = _me;
 449         struct bufferevent *bev = downcast(bevf);
 450         int processed_any = 0;
 451
 452         _bufferevent_incref_and_lock(bev);
 453         be_filter_process_output(bevf, BEV_NORMAL, &processed_any);
 454         _bufferevent_decref_and_unlock(bev);
 455 }
 456
 457 /* Called when the underlying socket has given us an error */
 458 static void
 459 be_filter_eventcb(struct bufferevent *underlying, short what, void *_me)
 460 {
 461         struct bufferevent_filtered *bevf = _me;
 462         struct bufferevent *bev = downcast(bevf);
 463
 464         _bufferevent_incref_and_lock(bev);
 465         /* All we can really to is tell our own eventcb. */
 466         _bufferevent_run_eventcb(bev, what);
 467         _bufferevent_decref_and_unlock(bev);
 468 }
 469
 470 static int
 471 be_filter_flush(struct bufferevent *bufev,
 472     short iotype, enum bufferevent_flush_mode mode)
 473 {
 474         struct bufferevent_filtered *bevf = upcast(bufev);
 475         int processed_any = 0;
 476         EVUTIL_ASSERT(bevf);
 477
 478         _bufferevent_incref_and_lock(bufev);
 479
 480         if (iotype & EV_READ) {
 481                 be_filter_process_input(bevf, mode, &processed_any);
 482         }
 483         if (iotype & EV_WRITE) {
 484                 be_filter_process_output(bevf, mode, &processed_any);
 485         }
 486         /* XXX check the return value? */
 487         /* XXX does this want to recursively call lower-level flushes? */
 488         bufferevent_flush(bevf->underlying, iotype, mode);
 489
 490         _bufferevent_decref_and_unlock(bufev);
 491
 492         return processed_any;
 493 }
 494
 495 static int
 496 be_filter_ctrl(struct bufferevent *bev, enum bufferevent_ctrl_op op,
 497     union bufferevent_ctrl_data *data)
 498 {
 499         struct bufferevent_filtered *bevf;
 500         switch (op) {
 501         case BEV_CTRL_GET_UNDERLYING:
 502                 bevf = upcast(bev);
 503                 data->ptr = bevf->underlying;
 504                 return 0;
 505         case BEV_CTRL_GET_FD:
 506         case BEV_CTRL_SET_FD:
 507         case BEV_CTRL_CANCEL_ALL:
 508         default:
 509                 return -1;
 510         }
 511 }