libevent/bufferevent_async.c

   1 /*
   2  * Copyright (c) 2009-2010 Niels Provos and Nick Mathewson
   3  *
   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 "event2/event-config.h"
  30
  31 #ifdef _EVENT_HAVE_SYS_TIME_H
  32 #include <sys/time.h>
  33 #endif
  34
  35 #include <errno.h>
  36 #include <stdio.h>
  37 #include <stdlib.h>
  38 #include <string.h>
  39 #ifdef _EVENT_HAVE_STDARG_H
  40 #include <stdarg.h>
  41 #endif
  42 #ifdef _EVENT_HAVE_UNISTD_H
  43 #include <unistd.h>
  44 #endif
  45
  46 #ifdef WIN32
  47 #include <winsock2.h>
  48 #include <ws2tcpip.h>
  49 #endif
  50
  51 #include <sys/queue.h>
  52
  53 #include "event2/util.h"
  54 #include "event2/bufferevent.h"
  55 #include "event2/buffer.h"
  56 #include "event2/bufferevent_struct.h"
  57 #include "event2/event.h"
  58 #include "event2/util.h"
  59 #include "event-internal.h"
  60 #include "log-internal.h"
  61 #include "mm-internal.h"
  62 #include "bufferevent-internal.h"
  63 #include "util-internal.h"
  64 #include "iocp-internal.h"
  65
  66 #ifndef SO_UPDATE_CONNECT_CONTEXT
  67 /* Mingw is sometimes missing this */
  68 #define SO_UPDATE_CONNECT_CONTEXT 0x7010
  69 #endif
  70
  71 /* prototypes */
  72 static int be_async_enable(struct bufferevent *, short);
  73 static int be_async_disable(struct bufferevent *, short);
  74 static void be_async_destruct(struct bufferevent *);
  75 static int be_async_flush(struct bufferevent *, short, enum bufferevent_flush_mode);
  76 static int be_async_ctrl(struct bufferevent *, enum bufferevent_ctrl_op, union bufferevent_ctrl_data *);
  77
  78 struct bufferevent_async {
  79         struct bufferevent_private bev;
  80         struct event_overlapped connect_overlapped;
  81         struct event_overlapped read_overlapped;
  82         struct event_overlapped write_overlapped;
  83         unsigned read_in_progress : 1;
  84         unsigned write_in_progress : 1;
  85         unsigned ok : 1;
  86         unsigned read_added : 1;
  87         unsigned write_added : 1;
  88 };
  89
  90 const struct bufferevent_ops bufferevent_ops_async = {
  91         "socket_async",
  92         evutil_offsetof(struct bufferevent_async, bev.bev),
  93         be_async_enable,
  94         be_async_disable,
  95         be_async_destruct,
  96         _bufferevent_generic_adj_timeouts,
  97         be_async_flush,
  98         be_async_ctrl,
  99 };
 100
 101 static inline struct bufferevent_async *
 102 upcast(struct bufferevent *bev)
 103 {
 104         struct bufferevent_async *bev_a;
 105         if (bev->be_ops != &bufferevent_ops_async)
 106                 return NULL;
 107         bev_a = EVUTIL_UPCAST(bev, struct bufferevent_async, bev.bev);
 108         return bev_a;
 109 }
 110
 111 static inline struct bufferevent_async *
 112 upcast_connect(struct event_overlapped *eo)
 113 {
 114         struct bufferevent_async *bev_a;
 115         bev_a = EVUTIL_UPCAST(eo, struct bufferevent_async, connect_overlapped);
 116         EVUTIL_ASSERT(BEV_IS_ASYNC(&bev_a->bev.bev));
 117         return bev_a;
 118 }
 119
 120 static inline struct bufferevent_async *
 121 upcast_read(struct event_overlapped *eo)
 122 {
 123         struct bufferevent_async *bev_a;
 124         bev_a = EVUTIL_UPCAST(eo, struct bufferevent_async, read_overlapped);
 125         EVUTIL_ASSERT(BEV_IS_ASYNC(&bev_a->bev.bev));
 126         return bev_a;
 127 }
 128
 129 static inline struct bufferevent_async *
 130 upcast_write(struct event_overlapped *eo)
 131 {
 132         struct bufferevent_async *bev_a;
 133         bev_a = EVUTIL_UPCAST(eo, struct bufferevent_async, write_overlapped);
 134         EVUTIL_ASSERT(BEV_IS_ASYNC(&bev_a->bev.bev));
 135         return bev_a;
 136 }
 137
 138 static void
 139 bev_async_del_write(struct bufferevent_async *beva)
 140 {
 141         struct bufferevent *bev = &beva->bev.bev;
 142
 143         if (beva->write_added) {
 144                 beva->write_added = 0;
 145                 event_base_del_virtual(bev->ev_base);
 146         }
 147 }
 148
 149 static void
 150 bev_async_del_read(struct bufferevent_async *beva)
 151 {
 152         struct bufferevent *bev = &beva->bev.bev;
 153
 154         if (beva->read_added) {
 155                 beva->read_added = 0;
 156                 event_base_del_virtual(bev->ev_base);
 157         }
 158 }
 159
 160 static void
 161 bev_async_add_write(struct bufferevent_async *beva)
 162 {
 163         struct bufferevent *bev = &beva->bev.bev;
 164
 165         if (!beva->write_added) {
 166                 beva->write_added = 1;
 167                 event_base_add_virtual(bev->ev_base);
 168         }
 169 }
 170
 171 static void
 172 bev_async_add_read(struct bufferevent_async *beva)
 173 {
 174         struct bufferevent *bev = &beva->bev.bev;
 175
 176         if (!beva->read_added) {
 177                 beva->read_added = 1;
 178                 event_base_add_virtual(bev->ev_base);
 179         }
 180 }
 181
 182 static void
 183 bev_async_consider_writing(struct bufferevent_async *beva)
 184 {
 185         size_t at_most;
 186         int limit;
 187         struct bufferevent *bev = &beva->bev.bev;
 188
 189         /* Don't write if there's a write in progress, or we do not
 190          * want to write, or when there's nothing left to write. */
 191         if (beva->write_in_progress)
 192                 return;
 193         if (!beva->ok || !(bev->enabled&EV_WRITE) ||
 194             !evbuffer_get_length(bev->output)) {
 195                 bev_async_del_write(beva);
 196                 return;
 197         }
 198
 199         at_most = evbuffer_get_length(bev->output);
 200
 201         /* XXXX This over-commits. */
 202         /* This is safe so long as bufferevent_get_write_max never returns
 203          * more than INT_MAX.  That's true for now. XXXX */
 204         limit = (int)_bufferevent_get_write_max(&beva->bev);
 205         if (at_most >= (size_t)limit && limit >= 0)
 206                 at_most = limit;
 207
 208         if (beva->bev.write_suspended) {
 209                 bev_async_del_write(beva);
 210                 return;
 211         }
 212
 213         /*  XXXX doesn't respect low-water mark very well. */
 214         bufferevent_incref(bev);
 215         if (evbuffer_launch_write(bev->output, at_most,
 216             &beva->write_overlapped)) {
 217                 bufferevent_decref(bev);
 218                 beva->ok = 0;
 219                 _bufferevent_run_eventcb(bev, BEV_EVENT_ERROR);
 220         } else {
 221                 beva->write_in_progress = 1;
 222                 bev_async_add_write(beva);
 223         }
 224 }
 225
 226 static void
 227 bev_async_consider_reading(struct bufferevent_async *beva)
 228 {
 229         size_t cur_size;
 230         size_t read_high;
 231         size_t at_most;
 232         int limit;
 233         struct bufferevent *bev = &beva->bev.bev;
 234
 235         /* Don't read if there is a read in progress, or we do not
 236          * want to read. */
 237         if (beva->read_in_progress)
 238                 return;
 239         if (!beva->ok || !(bev->enabled&EV_READ)) {
 240                 bev_async_del_read(beva);
 241                 return;
 242         }
 243
 244         /* Don't read if we're full */
 245         cur_size = evbuffer_get_length(bev->input);
 246         read_high = bev->wm_read.high;
 247         if (read_high) {
 248                 if (cur_size >= read_high) {
 249                         bev_async_del_read(beva);
 250                         return;
 251                 }
 252                 at_most = read_high - cur_size;
 253         } else {
 254                 at_most = 16384; /* FIXME totally magic. */
 255         }
 256
 257         /* XXXX This over-commits. */
 258         /* XXXX see also not above on cast on _bufferevent_get_write_max() */
 259         limit = (int)_bufferevent_get_read_max(&beva->bev);
 260         if (at_most >= (size_t)limit && limit >= 0)
 261                 at_most = limit;
 262
 263         if (beva->bev.read_suspended) {
 264                 bev_async_del_read(beva);
 265                 return;
 266         }
 267
 268         bufferevent_incref(bev);
 269         if (evbuffer_launch_read(bev->input, at_most, &beva->read_overlapped)) {
 270                 beva->ok = 0;
 271                 bufferevent_decref(bev);
 272                 _bufferevent_run_eventcb(bev, BEV_EVENT_ERROR);
 273         } else {
 274                 beva->read_in_progress = 1;
 275                 bev_async_add_read(beva);
 276         }
 277
 278         return;
 279 }
 280
 281 static void
 282 be_async_outbuf_callback(struct evbuffer *buf,
 283     const struct evbuffer_cb_info *cbinfo,
 284     void *arg)
 285 {
 286         struct bufferevent *bev = arg;
 287         struct bufferevent_async *bev_async = upcast(bev);
 288
 289         /* If we added data to the outbuf and were not writing before,
 290          * we may want to write now. */
 291
 292         _bufferevent_incref_and_lock(bev);
 293
 294         if (cbinfo->n_added)
 295                 bev_async_consider_writing(bev_async);
 296
 297         _bufferevent_decref_and_unlock(bev);
 298 }
 299
 300 static void
 301 be_async_inbuf_callback(struct evbuffer *buf,
 302     const struct evbuffer_cb_info *cbinfo,
 303     void *arg)
 304 {
 305         struct bufferevent *bev = arg;
 306         struct bufferevent_async *bev_async = upcast(bev);
 307
 308         /* If we drained data from the inbuf and were not reading before,
 309          * we may want to read now */
 310
 311         _bufferevent_incref_and_lock(bev);
 312
 313         if (cbinfo->n_deleted)
 314                 bev_async_consider_reading(bev_async);
 315
 316         _bufferevent_decref_and_unlock(bev);
 317 }
 318
 319 static int
 320 be_async_enable(struct bufferevent *buf, short what)
 321 {
 322         struct bufferevent_async *bev_async = upcast(buf);
 323
 324         if (!bev_async->ok)
 325                 return -1;
 326
 327         /* NOTE: This interferes with non-blocking connect */
 328         if (what & EV_READ)
 329                 BEV_RESET_GENERIC_READ_TIMEOUT(buf);
 330         if (what & EV_WRITE)
 331                 BEV_RESET_GENERIC_WRITE_TIMEOUT(buf);
 332
 333         /* If we newly enable reading or writing, and we aren't reading or
 334            writing already, consider launching a new read or write. */
 335
 336         if (what & EV_READ)
 337                 bev_async_consider_reading(bev_async);
 338         if (what & EV_WRITE)
 339                 bev_async_consider_writing(bev_async);
 340         return 0;
 341 }
 342
 343 static int
 344 be_async_disable(struct bufferevent *bev, short what)
 345 {
 346         struct bufferevent_async *bev_async = upcast(bev);
 347         /* XXXX If we disable reading or writing, we may want to consider
 348          * canceling any in-progress read or write operation, though it might
 349          * not work. */
 350
 351         if (what & EV_READ) {
 352                 BEV_DEL_GENERIC_READ_TIMEOUT(bev);
 353                 bev_async_del_read(bev_async);
 354         }
 355         if (what & EV_WRITE) {
 356                 BEV_DEL_GENERIC_WRITE_TIMEOUT(bev);
 357                 bev_async_del_write(bev_async);
 358         }
 359
 360         return 0;
 361 }
 362
 363 static void
 364 be_async_destruct(struct bufferevent *bev)
 365 {
 366         struct bufferevent_async *bev_async = upcast(bev);
 367         struct bufferevent_private *bev_p = BEV_UPCAST(bev);
 368         evutil_socket_t fd;
 369
 370         EVUTIL_ASSERT(!upcast(bev)->write_in_progress &&
 371                         !upcast(bev)->read_in_progress);
 372
 373         bev_async_del_read(bev_async);
 374         bev_async_del_write(bev_async);
 375
 376         fd = _evbuffer_overlapped_get_fd(bev->input);
 377         if (bev_p->options & BEV_OPT_CLOSE_ON_FREE)
 378                 evutil_closesocket(fd);
 379         /* delete this in case non-blocking connect was used */
 380         if (event_initialized(&bev->ev_write)) {
 381                 event_del(&bev->ev_write);
 382                 _bufferevent_del_generic_timeout_cbs(bev);
 383         }
 384 }
 385
 386 /* GetQueuedCompletionStatus doesn't reliably yield WSA error codes, so
 387  * we use WSAGetOverlappedResult to translate. */
 388 static void
 389 bev_async_set_wsa_error(struct bufferevent *bev, struct event_overlapped *eo)
 390 {
 391         DWORD bytes, flags;
 392         evutil_socket_t fd;
 393
 394         fd = _evbuffer_overlapped_get_fd(bev->input);
 395         WSAGetOverlappedResult(fd, &eo->overlapped, &bytes, FALSE, &flags);
 396 }
 397
 398 static int
 399 be_async_flush(struct bufferevent *bev, short what,
 400     enum bufferevent_flush_mode mode)
 401 {
 402         return 0;
 403 }
 404
 405 static void
 406 connect_complete(struct event_overlapped *eo, ev_uintptr_t key,
 407     ev_ssize_t nbytes, int ok)
 408 {
 409         struct bufferevent_async *bev_a = upcast_connect(eo);
 410         struct bufferevent *bev = &bev_a->bev.bev;
 411         evutil_socket_t sock;
 412
 413         BEV_LOCK(bev);
 414
 415         EVUTIL_ASSERT(bev_a->bev.connecting);
 416         bev_a->bev.connecting = 0;
 417         sock = _evbuffer_overlapped_get_fd(bev_a->bev.bev.input);
 418         /* XXXX Handle error? */
 419         setsockopt(sock, SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT, NULL, 0);
 420
 421         if (ok)
 422                 bufferevent_async_set_connected(bev);
 423         else
 424                 bev_async_set_wsa_error(bev, eo);
 425
 426         _bufferevent_run_eventcb(bev,
 427                         ok? BEV_EVENT_CONNECTED : BEV_EVENT_ERROR);
 428
 429         event_base_del_virtual(bev->ev_base);
 430
 431         _bufferevent_decref_and_unlock(bev);
 432 }
 433
 434 static void
 435 read_complete(struct event_overlapped *eo, ev_uintptr_t key,
 436     ev_ssize_t nbytes, int ok)
 437 {
 438         struct bufferevent_async *bev_a = upcast_read(eo);
 439         struct bufferevent *bev = &bev_a->bev.bev;
 440         short what = BEV_EVENT_READING;
 441
 442         BEV_LOCK(bev);
 443         EVUTIL_ASSERT(bev_a->read_in_progress);
 444
 445         evbuffer_commit_read(bev->input, nbytes);
 446         bev_a->read_in_progress = 0;
 447
 448         if (!ok)
 449                 bev_async_set_wsa_error(bev, eo);
 450
 451         if (bev_a->ok) {
 452                 if (ok && nbytes) {
 453                         BEV_RESET_GENERIC_READ_TIMEOUT(bev);
 454                         _bufferevent_decrement_read_buckets(&bev_a->bev,
 455                                         nbytes);
 456                         if (evbuffer_get_length(bev->input) >= bev->wm_read.low)
 457                                 _bufferevent_run_readcb(bev);
 458                         bev_async_consider_reading(bev_a);
 459                 } else if (!ok) {
 460                         what |= BEV_EVENT_ERROR;
 461                         bev_a->ok = 0;
 462                         _bufferevent_run_eventcb(bev, what);
 463                 } else if (!nbytes) {
 464                         what |= BEV_EVENT_EOF;
 465                         bev_a->ok = 0;
 466                         _bufferevent_run_eventcb(bev, what);
 467                 }
 468         }
 469
 470         _bufferevent_decref_and_unlock(bev);
 471 }
 472
 473 static void
 474 write_complete(struct event_overlapped *eo, ev_uintptr_t key,
 475     ev_ssize_t nbytes, int ok)
 476 {
 477         struct bufferevent_async *bev_a = upcast_write(eo);
 478         struct bufferevent *bev = &bev_a->bev.bev;
 479         short what = BEV_EVENT_WRITING;
 480
 481         BEV_LOCK(bev);
 482         EVUTIL_ASSERT(bev_a->write_in_progress);
 483         evbuffer_commit_write(bev->output, nbytes);
 484         bev_a->write_in_progress = 0;
 485
 486         if (!ok)
 487                 bev_async_set_wsa_error(bev, eo);
 488
 489         if (bev_a->ok) {
 490                 if (ok && nbytes) {
 491                         BEV_RESET_GENERIC_WRITE_TIMEOUT(bev);
 492                         _bufferevent_decrement_write_buckets(&bev_a->bev,
 493                                         nbytes);
 494                         if (evbuffer_get_length(bev->output) <=
 495                             bev->wm_write.low)
 496                                 _bufferevent_run_writecb(bev);
 497                         bev_async_consider_writing(bev_a);
 498                 } else if (!ok) {
 499                         what |= BEV_EVENT_ERROR;
 500                         bev_a->ok = 0;
 501                         _bufferevent_run_eventcb(bev, what);
 502                 } else if (!nbytes) {
 503                         what |= BEV_EVENT_EOF;
 504                         bev_a->ok = 0;
 505                         _bufferevent_run_eventcb(bev, what);
 506                 }
 507         }
 508
 509         _bufferevent_decref_and_unlock(bev);
 510 }
 511
 512 struct bufferevent *
 513 bufferevent_async_new(struct event_base *base,
 514     evutil_socket_t fd, int options)
 515 {
 516         struct bufferevent_async *bev_a;
 517         struct bufferevent *bev;
 518         struct event_iocp_port *iocp;
 519
 520         options |= BEV_OPT_THREADSAFE;
 521
 522         if (!(iocp = event_base_get_iocp(base)))
 523                 return NULL;
 524
 525         if (fd >= 0 && event_iocp_port_associate(iocp, fd, 1)<0) {
 526                 int err = GetLastError();
 527                 /* We may have alrady associated this fd with a port.
 528                  * Let's hope it's this port, and that the error code
 529                  * for doing this neer changes. */
 530                 if (err != ERROR_INVALID_PARAMETER)
 531                         return NULL;
 532         }
 533
 534         if (!(bev_a = mm_calloc(1, sizeof(struct bufferevent_async))))
 535                 return NULL;
 536
 537         bev = &bev_a->bev.bev;
 538         if (!(bev->input = evbuffer_overlapped_new(fd))) {
 539                 mm_free(bev_a);
 540                 return NULL;
 541         }
 542         if (!(bev->output = evbuffer_overlapped_new(fd))) {
 543                 evbuffer_free(bev->input);
 544                 mm_free(bev_a);
 545                 return NULL;
 546         }
 547
 548         if (bufferevent_init_common(&bev_a->bev, base, &bufferevent_ops_async,
 549                 options)<0)
 550                 goto err;
 551
 552         evbuffer_add_cb(bev->input, be_async_inbuf_callback, bev);
 553         evbuffer_add_cb(bev->output, be_async_outbuf_callback, bev);
 554
 555         event_overlapped_init(&bev_a->connect_overlapped, connect_complete);
 556         event_overlapped_init(&bev_a->read_overlapped, read_complete);
 557         event_overlapped_init(&bev_a->write_overlapped, write_complete);
 558
 559         bev_a->ok = fd >= 0;
 560         if (bev_a->ok)
 561                 _bufferevent_init_generic_timeout_cbs(bev);
 562
 563         return bev;
 564 err:
 565         bufferevent_free(&bev_a->bev.bev);
 566         return NULL;
 567 }
 568
 569 void
 570 bufferevent_async_set_connected(struct bufferevent *bev)
 571 {
 572         struct bufferevent_async *bev_async = upcast(bev);
 573         bev_async->ok = 1;
 574         _bufferevent_init_generic_timeout_cbs(bev);
 575         /* Now's a good time to consider reading/writing */
 576         be_async_enable(bev, bev->enabled);
 577 }
 578
 579 int
 580 bufferevent_async_can_connect(struct bufferevent *bev)
 581 {
 582         const struct win32_extension_fns *ext =
 583             event_get_win32_extension_fns();
 584
 585         if (BEV_IS_ASYNC(bev) &&
 586             event_base_get_iocp(bev->ev_base) &&
 587             ext && ext->ConnectEx)
 588                 return 1;
 589
 590         return 0;
 591 }
 592
 593 int
 594 bufferevent_async_connect(struct bufferevent *bev, evutil_socket_t fd,
 595         const struct sockaddr *sa, int socklen)
 596 {
 597         BOOL rc;
 598         struct bufferevent_async *bev_async = upcast(bev);
 599         struct sockaddr_storage ss;
 600         const struct win32_extension_fns *ext =
 601             event_get_win32_extension_fns();
 602
 603         EVUTIL_ASSERT(ext && ext->ConnectEx && fd >= 0 && sa != NULL);
 604
 605         /* ConnectEx() requires that the socket be bound to an address
 606          * with bind() before using, otherwise it will fail. We attempt
 607          * to issue a bind() here, taking into account that the error
 608          * code is set to WSAEINVAL when the socket is already bound. */
 609         memset(&ss, 0, sizeof(ss));
 610         if (sa->sa_family == AF_INET) {
 611                 struct sockaddr_in *sin = (struct sockaddr_in *)&ss;
 612                 sin->sin_family = AF_INET;
 613                 sin->sin_addr.s_addr = INADDR_ANY;
 614         } else if (sa->sa_family == AF_INET6) {
 615                 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&ss;
 616                 sin6->sin6_family = AF_INET6;
 617                 sin6->sin6_addr = in6addr_any;
 618         } else {
 619                 /* Well, the user will have to bind() */
 620                 return -1;
 621         }
 622         if (bind(fd, (struct sockaddr *)&ss, sizeof(ss)) < 0 &&
 623             WSAGetLastError() != WSAEINVAL)
 624                 return -1;
 625
 626         event_base_add_virtual(bev->ev_base);
 627         bufferevent_incref(bev);
 628         rc = ext->ConnectEx(fd, sa, socklen, NULL, 0, NULL,
 629                             &bev_async->connect_overlapped.overlapped);
 630         if (rc || WSAGetLastError() == ERROR_IO_PENDING)
 631                 return 0;
 632
 633         event_base_del_virtual(bev->ev_base);
 634         bufferevent_decref(bev);
 635
 636         return -1;
 637 }
 638
 639 static int
 640 be_async_ctrl(struct bufferevent *bev, enum bufferevent_ctrl_op op,
 641     union bufferevent_ctrl_data *data)
 642 {
 643         switch (op) {
 644         case BEV_CTRL_GET_FD:
 645                 data->fd = _evbuffer_overlapped_get_fd(bev->input);
 646                 return 0;
 647         case BEV_CTRL_SET_FD: {
 648                 struct event_iocp_port *iocp;
 649
 650                 if (data->fd == _evbuffer_overlapped_get_fd(bev->input))
 651                         return 0;
 652                 if (!(iocp = event_base_get_iocp(bev->ev_base)))
 653                         return -1;
 654                 if (event_iocp_port_associate(iocp, data->fd, 1) < 0)
 655                         return -1;
 656                 _evbuffer_overlapped_set_fd(bev->input, data->fd);
 657                 _evbuffer_overlapped_set_fd(bev->output, data->fd);
 658                 return 0;
 659         }
 660         case BEV_CTRL_GET_UNDERLYING:
 661         default:
 662                 return -1;
 663         }
 664 }