2 * Copyright (c) 2007-2012 Niels Provos and Nick Mathewson
3 * Copyright (c) 2002-2006 Niels Provos <provos@citi.umich.edu>
6 * Redistribution and use in source and binary forms, with or without
7 * modification, are permitted provided that the following conditions
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.
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.
29 #include "event2/event-config.h"
30 #include "evconfig-private.h"
32 #include <sys/types.h>
34 #ifdef EVENT__HAVE_SYS_TIME_H
42 #ifdef EVENT__HAVE_STDARG_H
45 #ifdef EVENT__HAVE_UNISTD_H
54 #ifdef EVENT__HAVE_SYS_SOCKET_H
55 #include <sys/socket.h>
57 #ifdef EVENT__HAVE_NETINET_IN_H
58 #include <netinet/in.h>
60 #ifdef EVENT__HAVE_NETINET_IN6_H
61 #include <netinet/in6.h>
64 #include "event2/util.h"
65 #include "event2/bufferevent.h"
66 #include "event2/buffer.h"
67 #include "event2/bufferevent_struct.h"
68 #include "event2/bufferevent_compat.h"
69 #include "event2/event.h"
70 #include "log-internal.h"
71 #include "mm-internal.h"
72 #include "bufferevent-internal.h"
73 #include "util-internal.h"
75 #include "iocp-internal.h"
79 static int be_socket_enable(struct bufferevent *, short);
80 static int be_socket_disable(struct bufferevent *, short);
81 static void be_socket_destruct(struct bufferevent *);
82 static int be_socket_flush(struct bufferevent *, short, enum bufferevent_flush_mode);
83 static int be_socket_ctrl(struct bufferevent *, enum bufferevent_ctrl_op, union bufferevent_ctrl_data *);
85 static void be_socket_setfd(struct bufferevent *, evutil_socket_t);
87 const struct bufferevent_ops bufferevent_ops_socket = {
89 evutil_offsetof(struct bufferevent_private, bev),
94 bufferevent_generic_adj_existing_timeouts_,
99 const struct sockaddr*
100 bufferevent_socket_get_conn_address_(struct bufferevent *bev)
102 struct bufferevent_private *bev_p =
103 EVUTIL_UPCAST(bev, struct bufferevent_private, bev);
105 return (struct sockaddr *)&bev_p->conn_address;
108 bufferevent_socket_set_conn_address_fd(struct bufferevent_private *bev_p, int fd)
110 socklen_t len = sizeof(bev_p->conn_address);
112 struct sockaddr *addr = (struct sockaddr *)&bev_p->conn_address;
113 if (addr->sa_family != AF_UNSPEC)
114 getpeername(fd, addr, &len);
117 bufferevent_socket_set_conn_address(struct bufferevent_private *bev_p,
118 struct sockaddr *addr, size_t addrlen)
120 EVUTIL_ASSERT(addrlen <= sizeof(bev_p->conn_address));
121 memcpy(&bev_p->conn_address, addr, addrlen);
125 bufferevent_socket_outbuf_cb(struct evbuffer *buf,
126 const struct evbuffer_cb_info *cbinfo,
129 struct bufferevent *bufev = arg;
130 struct bufferevent_private *bufev_p =
131 EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
133 if (cbinfo->n_added &&
134 (bufev->enabled & EV_WRITE) &&
135 !event_pending(&bufev->ev_write, EV_WRITE, NULL) &&
136 !bufev_p->write_suspended) {
137 /* Somebody added data to the buffer, and we would like to
138 * write, and we were not writing. So, start writing. */
139 if (bufferevent_add_event_(&bufev->ev_write, &bufev->timeout_write) == -1) {
140 /* Should we log this? */
146 bufferevent_readcb(evutil_socket_t fd, short event, void *arg)
148 struct bufferevent *bufev = arg;
149 struct bufferevent_private *bufev_p =
150 EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
151 struct evbuffer *input;
153 short what = BEV_EVENT_READING;
154 ev_ssize_t howmuch = -1, readmax=-1;
156 bufferevent_incref_and_lock_(bufev);
158 if (event == EV_TIMEOUT) {
159 /* Note that we only check for event==EV_TIMEOUT. If
160 * event==EV_TIMEOUT|EV_READ, we can safely ignore the
161 * timeout, since a read has occurred */
162 what |= BEV_EVENT_TIMEOUT;
166 input = bufev->input;
169 * If we have a high watermark configured then we don't want to
170 * read more data than would make us reach the watermark.
172 if (bufev->wm_read.high != 0) {
173 howmuch = bufev->wm_read.high - evbuffer_get_length(input);
174 /* we somehow lowered the watermark, stop reading */
176 bufferevent_wm_suspend_read(bufev);
180 readmax = bufferevent_get_read_max_(bufev_p);
181 if (howmuch < 0 || howmuch > readmax) /* The use of -1 for "unlimited"
182 * uglifies this code. XXXX */
184 if (bufev_p->read_suspended)
187 evbuffer_unfreeze(input, 0);
188 res = evbuffer_read(input, fd, (int)howmuch); /* XXXX evbuffer_read would do better to take and return ev_ssize_t */
189 evbuffer_freeze(input, 0);
192 int err = evutil_socket_geterror(fd);
193 if (EVUTIL_ERR_RW_RETRIABLE(err))
195 if (EVUTIL_ERR_CONNECT_REFUSED(err)) {
196 bufev_p->connection_refused = 1;
200 what |= BEV_EVENT_ERROR;
201 } else if (res == 0) {
203 what |= BEV_EVENT_EOF;
209 bufferevent_decrement_read_buckets_(bufev_p, res);
211 /* Invoke the user callback - must always be called last */
212 bufferevent_trigger_nolock_(bufev, EV_READ, 0);
220 bufferevent_disable(bufev, EV_READ);
221 bufferevent_run_eventcb_(bufev, what, 0);
224 bufferevent_decref_and_unlock_(bufev);
228 bufferevent_writecb(evutil_socket_t fd, short event, void *arg)
230 struct bufferevent *bufev = arg;
231 struct bufferevent_private *bufev_p =
232 EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
234 short what = BEV_EVENT_WRITING;
236 ev_ssize_t atmost = -1;
238 bufferevent_incref_and_lock_(bufev);
240 if (event == EV_TIMEOUT) {
241 /* Note that we only check for event==EV_TIMEOUT. If
242 * event==EV_TIMEOUT|EV_WRITE, we can safely ignore the
243 * timeout, since a read has occurred */
244 what |= BEV_EVENT_TIMEOUT;
247 if (bufev_p->connecting) {
248 int c = evutil_socket_finished_connecting_(fd);
249 /* we need to fake the error if the connection was refused
250 * immediately - usually connection to localhost on BSD */
251 if (bufev_p->connection_refused) {
252 bufev_p->connection_refused = 0;
259 bufev_p->connecting = 0;
261 event_del(&bufev->ev_write);
262 event_del(&bufev->ev_read);
263 bufferevent_run_eventcb_(bufev, BEV_EVENT_ERROR, 0);
267 bufferevent_socket_set_conn_address_fd(bufev_p, fd);
269 if (BEV_IS_ASYNC(bufev)) {
270 event_del(&bufev->ev_write);
271 bufferevent_async_set_connected_(bufev);
272 bufferevent_run_eventcb_(bufev,
273 BEV_EVENT_CONNECTED, 0);
277 bufferevent_run_eventcb_(bufev,
278 BEV_EVENT_CONNECTED, 0);
279 if (!(bufev->enabled & EV_WRITE) ||
280 bufev_p->write_suspended) {
281 event_del(&bufev->ev_write);
287 atmost = bufferevent_get_write_max_(bufev_p);
289 if (bufev_p->write_suspended)
292 if (evbuffer_get_length(bufev->output)) {
293 evbuffer_unfreeze(bufev->output, 1);
294 res = evbuffer_write_atmost(bufev->output, fd, atmost);
295 evbuffer_freeze(bufev->output, 1);
297 int err = evutil_socket_geterror(fd);
298 if (EVUTIL_ERR_RW_RETRIABLE(err))
300 what |= BEV_EVENT_ERROR;
301 } else if (res == 0) {
303 XXXX Actually, a 0 on write doesn't indicate
304 an EOF. An ECONNRESET might be more typical.
306 what |= BEV_EVENT_EOF;
311 bufferevent_decrement_write_buckets_(bufev_p, res);
314 if (evbuffer_get_length(bufev->output) == 0) {
315 event_del(&bufev->ev_write);
319 * Invoke the user callback if our buffer is drained or below the
322 if (res || !connected) {
323 bufferevent_trigger_nolock_(bufev, EV_WRITE, 0);
329 if (evbuffer_get_length(bufev->output) == 0) {
330 event_del(&bufev->ev_write);
335 bufferevent_disable(bufev, EV_WRITE);
336 bufferevent_run_eventcb_(bufev, what, 0);
339 bufferevent_decref_and_unlock_(bufev);
343 bufferevent_socket_new(struct event_base *base, evutil_socket_t fd,
346 struct bufferevent_private *bufev_p;
347 struct bufferevent *bufev;
350 if (base && event_base_get_iocp_(base))
351 return bufferevent_async_new_(base, fd, options);
354 if ((bufev_p = mm_calloc(1, sizeof(struct bufferevent_private)))== NULL)
357 if (bufferevent_init_common_(bufev_p, base, &bufferevent_ops_socket,
362 bufev = &bufev_p->bev;
363 evbuffer_set_flags(bufev->output, EVBUFFER_FLAG_DRAINS_TO_FD);
365 event_assign(&bufev->ev_read, bufev->ev_base, fd,
366 EV_READ|EV_PERSIST|EV_FINALIZE, bufferevent_readcb, bufev);
367 event_assign(&bufev->ev_write, bufev->ev_base, fd,
368 EV_WRITE|EV_PERSIST|EV_FINALIZE, bufferevent_writecb, bufev);
370 evbuffer_add_cb(bufev->output, bufferevent_socket_outbuf_cb, bufev);
372 evbuffer_freeze(bufev->input, 0);
373 evbuffer_freeze(bufev->output, 1);
379 bufferevent_socket_connect(struct bufferevent *bev,
380 const struct sockaddr *sa, int socklen)
382 struct bufferevent_private *bufev_p =
383 EVUTIL_UPCAST(bev, struct bufferevent_private, bev);
390 bufferevent_incref_and_lock_(bev);
395 fd = bufferevent_getfd(bev);
399 fd = evutil_socket_(sa->sa_family,
400 SOCK_STREAM|EVUTIL_SOCK_NONBLOCK, 0);
407 if (bufferevent_async_can_connect_(bev)) {
408 bufferevent_setfd(bev, fd);
409 r = bufferevent_async_connect_(bev, fd, sa, socklen);
412 bufev_p->connecting = 1;
417 r = evutil_socket_connect_(&fd, sa, socklen);
422 /* ConnectEx() isn't always around, even when IOCP is enabled.
423 * Here, we borrow the socket object's write handler to fall back
424 * on a non-blocking connect() when ConnectEx() is unavailable. */
425 if (BEV_IS_ASYNC(bev)) {
426 event_assign(&bev->ev_write, bev->ev_base, fd,
427 EV_WRITE|EV_PERSIST|EV_FINALIZE, bufferevent_writecb, bev);
430 bufferevent_setfd(bev, fd);
432 if (! be_socket_enable(bev, EV_WRITE)) {
433 bufev_p->connecting = 1;
438 /* The connect succeeded already. How very BSD of it. */
440 bufev_p->connecting = 1;
441 bufferevent_trigger_nolock_(bev, EV_WRITE, BEV_OPT_DEFER_CALLBACKS);
443 /* The connect failed already. How very BSD of it. */
445 bufferevent_run_eventcb_(bev, BEV_EVENT_ERROR, BEV_OPT_DEFER_CALLBACKS);
446 bufferevent_disable(bev, EV_WRITE|EV_READ);
452 bufferevent_run_eventcb_(bev, BEV_EVENT_ERROR, 0);
454 evutil_closesocket(fd);
455 /* do something about the error? */
457 bufferevent_decref_and_unlock_(bev);
462 bufferevent_connect_getaddrinfo_cb(int result, struct evutil_addrinfo *ai,
465 struct bufferevent *bev = arg;
466 struct bufferevent_private *bev_p =
467 EVUTIL_UPCAST(bev, struct bufferevent_private, bev);
471 bufferevent_unsuspend_write_(bev, BEV_SUSPEND_LOOKUP);
472 bufferevent_unsuspend_read_(bev, BEV_SUSPEND_LOOKUP);
474 bev_p->dns_request = NULL;
476 if (result == EVUTIL_EAI_CANCEL) {
477 bev_p->dns_error = result;
478 bufferevent_decref_and_unlock_(bev);
482 bev_p->dns_error = result;
483 bufferevent_run_eventcb_(bev, BEV_EVENT_ERROR, 0);
484 bufferevent_decref_and_unlock_(bev);
486 evutil_freeaddrinfo(ai);
490 /* XXX use the other addrinfos? */
491 /* XXX use this return value */
492 bufferevent_socket_set_conn_address(bev_p, ai->ai_addr, (int)ai->ai_addrlen);
493 r = bufferevent_socket_connect(bev, ai->ai_addr, (int)ai->ai_addrlen);
495 bufferevent_decref_and_unlock_(bev);
496 evutil_freeaddrinfo(ai);
500 bufferevent_socket_connect_hostname(struct bufferevent *bev,
501 struct evdns_base *evdns_base, int family, const char *hostname, int port)
504 struct evutil_addrinfo hint;
505 struct bufferevent_private *bev_p =
506 EVUTIL_UPCAST(bev, struct bufferevent_private, bev);
508 if (family != AF_INET && family != AF_INET6 && family != AF_UNSPEC)
510 if (port < 1 || port > 65535)
513 memset(&hint, 0, sizeof(hint));
514 hint.ai_family = family;
515 hint.ai_protocol = IPPROTO_TCP;
516 hint.ai_socktype = SOCK_STREAM;
518 evutil_snprintf(portbuf, sizeof(portbuf), "%d", port);
521 bev_p->dns_error = 0;
523 bufferevent_suspend_write_(bev, BEV_SUSPEND_LOOKUP);
524 bufferevent_suspend_read_(bev, BEV_SUSPEND_LOOKUP);
526 bufferevent_incref_(bev);
527 bev_p->dns_request = evutil_getaddrinfo_async_(evdns_base, hostname,
528 portbuf, &hint, bufferevent_connect_getaddrinfo_cb, bev);
535 bufferevent_socket_get_dns_error(struct bufferevent *bev)
538 struct bufferevent_private *bev_p =
539 EVUTIL_UPCAST(bev, struct bufferevent_private, bev);
542 rv = bev_p->dns_error;
549 * Create a new buffered event object.
551 * The read callback is invoked whenever we read new data.
552 * The write callback is invoked whenever the output buffer is drained.
553 * The error callback is invoked on a write/read error or on EOF.
555 * Both read and write callbacks maybe NULL. The error callback is not
556 * allowed to be NULL and have to be provided always.
560 bufferevent_new(evutil_socket_t fd,
561 bufferevent_data_cb readcb, bufferevent_data_cb writecb,
562 bufferevent_event_cb eventcb, void *cbarg)
564 struct bufferevent *bufev;
566 if (!(bufev = bufferevent_socket_new(NULL, fd, 0)))
569 bufferevent_setcb(bufev, readcb, writecb, eventcb, cbarg);
576 be_socket_enable(struct bufferevent *bufev, short event)
578 if (event & EV_READ &&
579 bufferevent_add_event_(&bufev->ev_read, &bufev->timeout_read) == -1)
581 if (event & EV_WRITE &&
582 bufferevent_add_event_(&bufev->ev_write, &bufev->timeout_write) == -1)
588 be_socket_disable(struct bufferevent *bufev, short event)
590 struct bufferevent_private *bufev_p =
591 EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
592 if (event & EV_READ) {
593 if (event_del(&bufev->ev_read) == -1)
596 /* Don't actually disable the write if we are trying to connect. */
597 if ((event & EV_WRITE) && ! bufev_p->connecting) {
598 if (event_del(&bufev->ev_write) == -1)
605 be_socket_destruct(struct bufferevent *bufev)
607 struct bufferevent_private *bufev_p =
608 EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
610 EVUTIL_ASSERT(bufev->be_ops == &bufferevent_ops_socket);
612 fd = event_get_fd(&bufev->ev_read);
614 if ((bufev_p->options & BEV_OPT_CLOSE_ON_FREE) && fd >= 0)
615 EVUTIL_CLOSESOCKET(fd);
617 evutil_getaddrinfo_cancel_async_(bufev_p->dns_request);
621 be_socket_flush(struct bufferevent *bev, short iotype,
622 enum bufferevent_flush_mode mode)
629 be_socket_setfd(struct bufferevent *bufev, evutil_socket_t fd)
631 struct bufferevent_private *bufev_p =
632 EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
635 EVUTIL_ASSERT(bufev->be_ops == &bufferevent_ops_socket);
637 event_del(&bufev->ev_read);
638 event_del(&bufev->ev_write);
640 evbuffer_unfreeze(bufev->input, 0);
641 evbuffer_unfreeze(bufev->output, 1);
643 event_assign(&bufev->ev_read, bufev->ev_base, fd,
644 EV_READ|EV_PERSIST|EV_FINALIZE, bufferevent_readcb, bufev);
645 event_assign(&bufev->ev_write, bufev->ev_base, fd,
646 EV_WRITE|EV_PERSIST|EV_FINALIZE, bufferevent_writecb, bufev);
649 bufferevent_enable(bufev, bufev->enabled);
651 evutil_getaddrinfo_cancel_async_(bufev_p->dns_request);
656 /* XXXX Should non-socket bufferevents support this? */
658 bufferevent_priority_set(struct bufferevent *bufev, int priority)
661 struct bufferevent_private *bufev_p =
662 EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
665 if (bufev->be_ops != &bufferevent_ops_socket)
668 if (event_priority_set(&bufev->ev_read, priority) == -1)
670 if (event_priority_set(&bufev->ev_write, priority) == -1)
673 event_deferred_cb_set_priority_(&bufev_p->deferred, priority);
681 /* XXXX Should non-socket bufferevents support this? */
683 bufferevent_base_set(struct event_base *base, struct bufferevent *bufev)
688 if (bufev->be_ops != &bufferevent_ops_socket)
691 bufev->ev_base = base;
693 res = event_base_set(base, &bufev->ev_read);
697 res = event_base_set(base, &bufev->ev_write);
704 be_socket_ctrl(struct bufferevent *bev, enum bufferevent_ctrl_op op,
705 union bufferevent_ctrl_data *data)
708 case BEV_CTRL_SET_FD:
709 be_socket_setfd(bev, data->fd);
711 case BEV_CTRL_GET_FD:
712 data->fd = event_get_fd(&bev->ev_read);
714 case BEV_CTRL_GET_UNDERLYING:
715 case BEV_CTRL_CANCEL_ALL: