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 "evconfig-private.h" 30 31 #include <sys/types.h> 32 33 #include "event2/event-config.h" 34 35 #ifdef EVENT__HAVE_SYS_TIME_H 36 #include <sys/time.h> 37 #endif 38 39 #include <errno.h> 40 #include <stdio.h> 41 #include <stdlib.h> 42 #include <string.h> 43 #ifdef EVENT__HAVE_STDARG_H 44 #include <stdarg.h> 45 #endif 46 47 #ifdef _WIN32 48 #include <winsock2.h> 49 #endif 50 51 #include "event2/util.h" 52 #include "event2/bufferevent.h" 53 #include "event2/buffer.h" 54 #include "event2/bufferevent_struct.h" 55 #include "event2/event.h" 56 #include "log-internal.h" 57 #include "mm-internal.h" 58 #include "bufferevent-internal.h" 59 #include "util-internal.h" 60 61 /* prototypes */ 62 static int be_filter_enable(struct bufferevent *, short); 63 static int be_filter_disable(struct bufferevent *, short); 64 static void be_filter_unlink(struct bufferevent *); 65 static void be_filter_destruct(struct bufferevent *); 66 67 static void be_filter_readcb(struct bufferevent *, void *); 68 static void be_filter_writecb(struct bufferevent *, void *); 69 static void be_filter_eventcb(struct bufferevent *, short, void *); 70 static int be_filter_flush(struct bufferevent *bufev, 71 short iotype, enum bufferevent_flush_mode mode); 72 static int be_filter_ctrl(struct bufferevent *, enum bufferevent_ctrl_op, union bufferevent_ctrl_data *); 73 74 static void bufferevent_filtered_outbuf_cb(struct evbuffer *buf, 75 const struct evbuffer_cb_info *info, void *arg); 76 77 struct bufferevent_filtered { 78 struct bufferevent_private bev; 79 80 /** The bufferevent that we read/write filtered data from/to. */ 81 struct bufferevent *underlying; 82 /** A callback on our outbuf to notice when somebody adds data */ 83 struct evbuffer_cb_entry *outbuf_cb; 84 /** True iff we have received an EOF callback from the underlying 85 * bufferevent. */ 86 unsigned got_eof; 87 88 /** Function to free context when we're done. */ 89 void (*free_context)(void *); 90 /** Input filter */ 91 bufferevent_filter_cb process_in; 92 /** Output filter */ 93 bufferevent_filter_cb process_out; 94 /** User-supplied argument to the filters. */ 95 void *context; 96 }; 97 98 const struct bufferevent_ops bufferevent_ops_filter = { 99 "filter", 100 evutil_offsetof(struct bufferevent_filtered, bev.bev), 101 be_filter_enable, 102 be_filter_disable, 103 be_filter_unlink, 104 be_filter_destruct, 105 bufferevent_generic_adj_timeouts_, 106 be_filter_flush, 107 be_filter_ctrl, 108 }; 109 110 /* Given a bufferevent that's really the bev filter of a bufferevent_filtered, 111 * return that bufferevent_filtered. Returns NULL otherwise.*/ 112 static inline struct bufferevent_filtered * 113 upcast(struct bufferevent *bev) 114 { 115 struct bufferevent_filtered *bev_f; 116 if (bev->be_ops != &bufferevent_ops_filter) 117 return NULL; 118 bev_f = (void*)( ((char*)bev) - 119 evutil_offsetof(struct bufferevent_filtered, bev.bev)); 120 EVUTIL_ASSERT(bev_f->bev.bev.be_ops == &bufferevent_ops_filter); 121 return bev_f; 122 } 123 124 #define downcast(bev_f) (&(bev_f)->bev.bev) 125 126 /** Return 1 iff bevf's underlying bufferevent's output buffer is at or 127 * over its high watermark such that we should not write to it in a given 128 * flush mode. */ 129 static int 130 be_underlying_writebuf_full(struct bufferevent_filtered *bevf, 131 enum bufferevent_flush_mode state) 132 { 133 struct bufferevent *u = bevf->underlying; 134 return state == BEV_NORMAL && 135 u->wm_write.high && 136 evbuffer_get_length(u->output) >= u->wm_write.high; 137 } 138 139 /** Return 1 if our input buffer is at or over its high watermark such that we 140 * should not write to it in a given flush mode. */ 141 static int 142 be_readbuf_full(struct bufferevent_filtered *bevf, 143 enum bufferevent_flush_mode state) 144 { 145 struct bufferevent *bufev = downcast(bevf); 146 return state == BEV_NORMAL && 147 bufev->wm_read.high && 148 evbuffer_get_length(bufev->input) >= bufev->wm_read.high; 149 } 150 151 152 /* Filter to use when we're created with a NULL filter. */ 153 static enum bufferevent_filter_result 154 be_null_filter(struct evbuffer *src, struct evbuffer *dst, ev_ssize_t lim, 155 enum bufferevent_flush_mode state, void *ctx) 156 { 157 (void)state; 158 if (evbuffer_remove_buffer(src, dst, lim) == 0) 159 return BEV_OK; 160 else 161 return BEV_ERROR; 162 } 163 164 struct bufferevent * 165 bufferevent_filter_new(struct bufferevent *underlying, 166 bufferevent_filter_cb input_filter, 167 bufferevent_filter_cb output_filter, 168 int options, 169 void (*free_context)(void *), 170 void *ctx) 171 { 172 struct bufferevent_filtered *bufev_f; 173 int tmp_options = options & ~BEV_OPT_THREADSAFE; 174 175 if (!underlying) 176 return NULL; 177 178 if (!input_filter) 179 input_filter = be_null_filter; 180 if (!output_filter) 181 output_filter = be_null_filter; 182 183 bufev_f = mm_calloc(1, sizeof(struct bufferevent_filtered)); 184 if (!bufev_f) 185 return NULL; 186 187 if (bufferevent_init_common_(&bufev_f->bev, underlying->ev_base, 188 &bufferevent_ops_filter, tmp_options) < 0) { 189 mm_free(bufev_f); 190 return NULL; 191 } 192 if (options & BEV_OPT_THREADSAFE) { 193 bufferevent_enable_locking_(downcast(bufev_f), NULL); 194 } 195 196 bufev_f->underlying = underlying; 197 198 bufev_f->process_in = input_filter; 199 bufev_f->process_out = output_filter; 200 bufev_f->free_context = free_context; 201 bufev_f->context = ctx; 202 203 bufferevent_setcb(bufev_f->underlying, 204 be_filter_readcb, be_filter_writecb, be_filter_eventcb, bufev_f); 205 206 bufev_f->outbuf_cb = evbuffer_add_cb(downcast(bufev_f)->output, 207 bufferevent_filtered_outbuf_cb, bufev_f); 208 209 bufferevent_init_generic_timeout_cbs_(downcast(bufev_f)); 210 bufferevent_incref_(underlying); 211 212 bufferevent_enable(underlying, EV_READ|EV_WRITE); 213 bufferevent_suspend_read_(underlying, BEV_SUSPEND_FILT_READ); 214 215 return downcast(bufev_f); 216 } 217 218 static void 219 be_filter_unlink(struct bufferevent *bev) 220 { 221 struct bufferevent_filtered *bevf = upcast(bev); 222 EVUTIL_ASSERT(bevf); 223 224 if (bevf->bev.options & BEV_OPT_CLOSE_ON_FREE) { 225 /* Yes, there is also a decref in bufferevent_decref_. 226 * That decref corresponds to the incref when we set 227 * underlying for the first time. This decref is an 228 * extra one to remove the last reference. 229 */ 230 if (BEV_UPCAST(bevf->underlying)->refcnt < 2) { 231 event_warnx("BEV_OPT_CLOSE_ON_FREE set on an " 232 "bufferevent with too few references"); 233 } else { 234 bufferevent_free(bevf->underlying); 235 } 236 } else { 237 if (bevf->underlying) { 238 if (bevf->underlying->errorcb == be_filter_eventcb) 239 bufferevent_setcb(bevf->underlying, 240 NULL, NULL, NULL, NULL); 241 bufferevent_unsuspend_read_(bevf->underlying, 242 BEV_SUSPEND_FILT_READ); 243 } 244 } 245 } 246 247 static void 248 be_filter_destruct(struct bufferevent *bev) 249 { 250 struct bufferevent_filtered *bevf = upcast(bev); 251 EVUTIL_ASSERT(bevf); 252 if (bevf->free_context) 253 bevf->free_context(bevf->context); 254 } 255 256 static int 257 be_filter_enable(struct bufferevent *bev, short event) 258 { 259 struct bufferevent_filtered *bevf = upcast(bev); 260 if (event & EV_WRITE) 261 BEV_RESET_GENERIC_WRITE_TIMEOUT(bev); 262 263 if (event & EV_READ) { 264 BEV_RESET_GENERIC_READ_TIMEOUT(bev); 265 bufferevent_unsuspend_read_(bevf->underlying, 266 BEV_SUSPEND_FILT_READ); 267 } 268 return 0; 269 } 270 271 static int 272 be_filter_disable(struct bufferevent *bev, short event) 273 { 274 struct bufferevent_filtered *bevf = upcast(bev); 275 if (event & EV_WRITE) 276 BEV_DEL_GENERIC_WRITE_TIMEOUT(bev); 277 if (event & EV_READ) { 278 BEV_DEL_GENERIC_READ_TIMEOUT(bev); 279 bufferevent_suspend_read_(bevf->underlying, 280 BEV_SUSPEND_FILT_READ); 281 } 282 return 0; 283 } 284 285 static enum bufferevent_filter_result 286 be_filter_process_input(struct bufferevent_filtered *bevf, 287 enum bufferevent_flush_mode state, 288 int *processed_out) 289 { 290 enum bufferevent_filter_result res; 291 struct bufferevent *bev = downcast(bevf); 292 293 if (state == BEV_NORMAL) { 294 /* If we're in 'normal' mode, don't urge data on the filter 295 * unless we're reading data and under our high-water mark.*/ 296 if (!(bev->enabled & EV_READ) || 297 be_readbuf_full(bevf, state)) 298 return BEV_OK; 299 } 300 301 do { 302 ev_ssize_t limit = -1; 303 if (state == BEV_NORMAL && bev->wm_read.high) 304 limit = bev->wm_read.high - 305 evbuffer_get_length(bev->input); 306 307 res = bevf->process_in(bevf->underlying->input, 308 bev->input, limit, state, bevf->context); 309 310 if (res == BEV_OK) 311 *processed_out = 1; 312 } while (res == BEV_OK && 313 (bev->enabled & EV_READ) && 314 evbuffer_get_length(bevf->underlying->input) && 315 !be_readbuf_full(bevf, state)); 316 317 if (*processed_out) 318 BEV_RESET_GENERIC_READ_TIMEOUT(bev); 319 320 return res; 321 } 322 323 324 static enum bufferevent_filter_result 325 be_filter_process_output(struct bufferevent_filtered *bevf, 326 enum bufferevent_flush_mode state, 327 int *processed_out) 328 { 329 /* Requires references and lock: might call writecb */ 330 enum bufferevent_filter_result res = BEV_OK; 331 struct bufferevent *bufev = downcast(bevf); 332 int again = 0; 333 334 if (state == BEV_NORMAL) { 335 /* If we're in 'normal' mode, don't urge data on the 336 * filter unless we're writing data, and the underlying 337 * bufferevent is accepting data, and we have data to 338 * give the filter. If we're in 'flush' or 'finish', 339 * call the filter no matter what. */ 340 if (!(bufev->enabled & EV_WRITE) || 341 be_underlying_writebuf_full(bevf, state) || 342 !evbuffer_get_length(bufev->output)) 343 return BEV_OK; 344 } 345 346 /* disable the callback that calls this function 347 when the user adds to the output buffer. */ 348 evbuffer_cb_set_flags(bufev->output, bevf->outbuf_cb, 0); 349 350 do { 351 int processed = 0; 352 again = 0; 353 354 do { 355 ev_ssize_t limit = -1; 356 if (state == BEV_NORMAL && 357 bevf->underlying->wm_write.high) 358 limit = bevf->underlying->wm_write.high - 359 evbuffer_get_length(bevf->underlying->output); 360 361 res = bevf->process_out(downcast(bevf)->output, 362 bevf->underlying->output, 363 limit, 364 state, 365 bevf->context); 366 367 if (res == BEV_OK) 368 processed = *processed_out = 1; 369 } while (/* Stop if the filter wasn't successful...*/ 370 res == BEV_OK && 371 /* Or if we aren't writing any more. */ 372 (bufev->enabled & EV_WRITE) && 373 /* Of if we have nothing more to write and we are 374 * not flushing. */ 375 evbuffer_get_length(bufev->output) && 376 /* Or if we have filled the underlying output buffer. */ 377 !be_underlying_writebuf_full(bevf,state)); 378 379 if (processed) { 380 /* call the write callback.*/ 381 bufferevent_trigger_nolock_(bufev, EV_WRITE, 0); 382 383 if (res == BEV_OK && 384 (bufev->enabled & EV_WRITE) && 385 evbuffer_get_length(bufev->output) && 386 !be_underlying_writebuf_full(bevf, state)) { 387 again = 1; 388 } 389 } 390 } while (again); 391 392 /* reenable the outbuf_cb */ 393 evbuffer_cb_set_flags(bufev->output,bevf->outbuf_cb, 394 EVBUFFER_CB_ENABLED); 395 396 if (*processed_out) 397 BEV_RESET_GENERIC_WRITE_TIMEOUT(bufev); 398 399 return res; 400 } 401 402 /* Called when the size of our outbuf changes. */ 403 static void 404 bufferevent_filtered_outbuf_cb(struct evbuffer *buf, 405 const struct evbuffer_cb_info *cbinfo, void *arg) 406 { 407 struct bufferevent_filtered *bevf = arg; 408 struct bufferevent *bev = downcast(bevf); 409 410 if (cbinfo->n_added) { 411 int processed_any = 0; 412 /* Somebody added more data to the output buffer. Try to 413 * process it, if we should. */ 414 bufferevent_incref_and_lock_(bev); 415 be_filter_process_output(bevf, BEV_NORMAL, &processed_any); 416 bufferevent_decref_and_unlock_(bev); 417 } 418 } 419 420 /* Called when the underlying socket has read. */ 421 static void 422 be_filter_readcb(struct bufferevent *underlying, void *me_) 423 { 424 struct bufferevent_filtered *bevf = me_; 425 enum bufferevent_filter_result res; 426 enum bufferevent_flush_mode state; 427 struct bufferevent *bufev = downcast(bevf); 428 struct bufferevent_private *bufev_private = BEV_UPCAST(bufev); 429 int processed_any = 0; 430 431 BEV_LOCK(bufev); 432 433 // It's possible our refcount is 0 at this point if another thread free'd our filterevent 434 EVUTIL_ASSERT(bufev_private->refcnt >= 0); 435 436 // If our refcount is > 0 437 if (bufev_private->refcnt > 0) { 438 439 if (bevf->got_eof) 440 state = BEV_FINISHED; 441 else 442 state = BEV_NORMAL; 443 444 /* XXXX use return value */ 445 res = be_filter_process_input(bevf, state, &processed_any); 446 (void)res; 447 448 /* XXX This should be in process_input, not here. There are 449 * other places that can call process-input, and they should 450 * force readcb calls as needed. */ 451 if (processed_any) 452 bufferevent_trigger_nolock_(bufev, EV_READ, 0); 453 } 454 455 BEV_UNLOCK(bufev); 456 } 457 458 /* Called when the underlying socket has drained enough that we can write to 459 it. */ 460 static void 461 be_filter_writecb(struct bufferevent *underlying, void *me_) 462 { 463 struct bufferevent_filtered *bevf = me_; 464 struct bufferevent *bev = downcast(bevf); 465 struct bufferevent_private *bufev_private = BEV_UPCAST(bev); 466 int processed_any = 0; 467 468 BEV_LOCK(bev); 469 470 // It's possible our refcount is 0 at this point if another thread free'd our filterevent 471 EVUTIL_ASSERT(bufev_private->refcnt >= 0); 472 473 // If our refcount is > 0 474 if (bufev_private->refcnt > 0) { 475 be_filter_process_output(bevf, BEV_NORMAL, &processed_any); 476 } 477 478 BEV_UNLOCK(bev); 479 } 480 481 /* Called when the underlying socket has given us an error */ 482 static void 483 be_filter_eventcb(struct bufferevent *underlying, short what, void *me_) 484 { 485 struct bufferevent_filtered *bevf = me_; 486 struct bufferevent *bev = downcast(bevf); 487 struct bufferevent_private *bufev_private = BEV_UPCAST(bev); 488 489 BEV_LOCK(bev); 490 491 // It's possible our refcount is 0 at this point if another thread free'd our filterevent 492 EVUTIL_ASSERT(bufev_private->refcnt >= 0); 493 494 // If our refcount is > 0 495 if (bufev_private->refcnt > 0) { 496 497 /* All we can really to is tell our own eventcb. */ 498 bufferevent_run_eventcb_(bev, what, 0); 499 } 500 501 BEV_UNLOCK(bev); 502 } 503 504 static int 505 be_filter_flush(struct bufferevent *bufev, 506 short iotype, enum bufferevent_flush_mode mode) 507 { 508 struct bufferevent_filtered *bevf = upcast(bufev); 509 int processed_any = 0; 510 EVUTIL_ASSERT(bevf); 511 512 bufferevent_incref_and_lock_(bufev); 513 514 if (iotype & EV_READ) { 515 be_filter_process_input(bevf, mode, &processed_any); 516 } 517 if (iotype & EV_WRITE) { 518 be_filter_process_output(bevf, mode, &processed_any); 519 } 520 /* XXX check the return value? */ 521 /* XXX does this want to recursively call lower-level flushes? */ 522 bufferevent_flush(bevf->underlying, iotype, mode); 523 524 bufferevent_decref_and_unlock_(bufev); 525 526 return processed_any; 527 } 528 529 static int 530 be_filter_ctrl(struct bufferevent *bev, enum bufferevent_ctrl_op op, 531 union bufferevent_ctrl_data *data) 532 { 533 struct bufferevent_filtered *bevf; 534 switch (op) { 535 case BEV_CTRL_GET_UNDERLYING: 536 bevf = upcast(bev); 537 data->ptr = bevf->underlying; 538 return 0; 539 case BEV_CTRL_GET_FD: 540 case BEV_CTRL_SET_FD: 541 case BEV_CTRL_CANCEL_ALL: 542 default: 543 return -1; 544 } 545 } 546