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