The Tor Browser / file revision

 /*

  * Copyright (c) 2009-2012 Niels Provos and Nick Mathewson

  *

  * All rights reserved.

  *

  * Redistribution and use in source and binary forms, with or without

  * modification, are permitted provided that the following conditions

  * are met:

  * 1. Redistributions of source code must retain the above copyright

  *    notice, this list of conditions and the following disclaimer.

  * 2. Redistributions in binary form must reproduce the above copyright

  *    notice, this list of conditions and the following disclaimer in the

  *    documentation and/or other materials provided with the distribution.

  * 3. The name of the author may not be used to endorse or promote products

  *    derived from this software without specific prior written permission.

  *

  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR

  * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES

  * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.

  * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,

  * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT

  * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,

  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY

  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF

  * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

  */

 #include "event2/event-config.h"

 #ifdef _EVENT_HAVE_SYS_TIME_H

 #include <sys/time.h>

 #endif

 #include <errno.h>

 #include <stdio.h>

 #include <stdlib.h>

 #include <string.h>

 #ifdef _EVENT_HAVE_STDARG_H

 #include <stdarg.h>

 #endif

 #ifdef _EVENT_HAVE_UNISTD_H

 #include <unistd.h>

 #endif

 #ifdef WIN32

 #include <winsock2.h>

 #include <ws2tcpip.h>

 #endif

 #include <sys/queue.h>

 #include "event2/util.h"

 #include "event2/bufferevent.h"

 #include "event2/buffer.h"

 #include "event2/bufferevent_struct.h"

 #include "event2/event.h"

 #include "event2/util.h"

 #include "event-internal.h"

 #include "log-internal.h"

 #include "mm-internal.h"

 #include "bufferevent-internal.h"

 #include "util-internal.h"

 #include "iocp-internal.h"

 #ifndef SO_UPDATE_CONNECT_CONTEXT

 /* Mingw is sometimes missing this */

 #define SO_UPDATE_CONNECT_CONTEXT 0x7010

 #endif

 /* prototypes */

 static int be_async_enable(struct bufferevent *, short);

 static int be_async_disable(struct bufferevent *, short);

 static void be_async_destruct(struct bufferevent *);

 static int be_async_flush(struct bufferevent *, short, enum bufferevent_flush_mode);

 static int be_async_ctrl(struct bufferevent *, enum bufferevent_ctrl_op, union bufferevent_ctrl_data *);

 struct bufferevent_async {

 	struct bufferevent_private bev;

 	struct event_overlapped connect_overlapped;

 	struct event_overlapped read_overlapped;

 	struct event_overlapped write_overlapped;

 	size_t read_in_progress;

 	size_t write_in_progress;

 	unsigned ok : 1;

 	unsigned read_added : 1;

 	unsigned write_added : 1;

 };

 const struct bufferevent_ops bufferevent_ops_async = {

 	"socket_async",

 	evutil_offsetof(struct bufferevent_async, bev.bev),

 	be_async_enable,

 	be_async_disable,

 	be_async_destruct,

 	_bufferevent_generic_adj_timeouts,

 	be_async_flush,

 	be_async_ctrl,

 };

 static inline struct bufferevent_async *

 upcast(struct bufferevent *bev)

 {

 	struct bufferevent_async *bev_a;

 	if (bev->be_ops != &bufferevent_ops_async)

 		return NULL;

 	bev_a = EVUTIL_UPCAST(bev, struct bufferevent_async, bev.bev);

 	return bev_a;

 }

 static inline struct bufferevent_async *

 upcast_connect(struct event_overlapped *eo)

 {

 	struct bufferevent_async *bev_a;

 	bev_a = EVUTIL_UPCAST(eo, struct bufferevent_async, connect_overlapped);

 	EVUTIL_ASSERT(BEV_IS_ASYNC(&bev_a->bev.bev));

 	return bev_a;

 }

 static inline struct bufferevent_async *

 upcast_read(struct event_overlapped *eo)

 {

 	struct bufferevent_async *bev_a;

 	bev_a = EVUTIL_UPCAST(eo, struct bufferevent_async, read_overlapped);

 	EVUTIL_ASSERT(BEV_IS_ASYNC(&bev_a->bev.bev));

 	return bev_a;

 }

 static inline struct bufferevent_async *

 upcast_write(struct event_overlapped *eo)

 {

 	struct bufferevent_async *bev_a;

 	bev_a = EVUTIL_UPCAST(eo, struct bufferevent_async, write_overlapped);

 	EVUTIL_ASSERT(BEV_IS_ASYNC(&bev_a->bev.bev));

 	return bev_a;

 }

 static void

 bev_async_del_write(struct bufferevent_async *beva)

 {

 	struct bufferevent *bev = &beva->bev.bev;

 	if (beva->write_added) {

 		beva->write_added = 0;

 		event_base_del_virtual(bev->ev_base);

 	}

 }

 static void

 bev_async_del_read(struct bufferevent_async *beva)

 {

 	struct bufferevent *bev = &beva->bev.bev;

 	if (beva->read_added) {

 		beva->read_added = 0;

 		event_base_del_virtual(bev->ev_base);

 	}

 }

 static void

 bev_async_add_write(struct bufferevent_async *beva)

 {

 	struct bufferevent *bev = &beva->bev.bev;

 	if (!beva->write_added) {

 		beva->write_added = 1;

 		event_base_add_virtual(bev->ev_base);

 	}

 }

 static void

 bev_async_add_read(struct bufferevent_async *beva)

 {

 	struct bufferevent *bev = &beva->bev.bev;

 	if (!beva->read_added) {

 		beva->read_added = 1;

 		event_base_add_virtual(bev->ev_base);

 	}

 }

 static void

 bev_async_consider_writing(struct bufferevent_async *beva)

 {

 	size_t at_most;

 	int limit;

 	struct bufferevent *bev = &beva->bev.bev;

 	/* Don't write if there's a write in progress, or we do not

 	 * want to write, or when there's nothing left to write. */

 	if (beva->write_in_progress || beva->bev.connecting)

 		return;

 	if (!beva->ok || !(bev->enabled&EV_WRITE) ||

 	    !evbuffer_get_length(bev->output)) {

 		bev_async_del_write(beva);

 		return;

 	}

 	at_most = evbuffer_get_length(bev->output);

 	/* This is safe so long as bufferevent_get_write_max never returns

 	 * more than INT_MAX.  That's true for now. XXXX */

 	limit = (int)_bufferevent_get_write_max(&beva->bev);

 	if (at_most >= (size_t)limit && limit >= 0)

 		at_most = limit;

 	if (beva->bev.write_suspended) {

 		bev_async_del_write(beva);

 		return;

 	}

 	/*  XXXX doesn't respect low-water mark very well. */

 	bufferevent_incref(bev);

 	if (evbuffer_launch_write(bev->output, at_most,

 	    &beva->write_overlapped)) {

 		bufferevent_decref(bev);

 		beva->ok = 0;

 		_bufferevent_run_eventcb(bev, BEV_EVENT_ERROR);

 	} else {

 		beva->write_in_progress = at_most;

 		_bufferevent_decrement_write_buckets(&beva->bev, at_most);

 		bev_async_add_write(beva);

 	}

 }

 static void

 bev_async_consider_reading(struct bufferevent_async *beva)

 {

 	size_t cur_size;

 	size_t read_high;

 	size_t at_most;

 	int limit;

 	struct bufferevent *bev = &beva->bev.bev;

 	/* Don't read if there is a read in progress, or we do not

 	 * want to read. */

 	if (beva->read_in_progress || beva->bev.connecting)

 		return;

 	if (!beva->ok || !(bev->enabled&EV_READ)) {

 		bev_async_del_read(beva);

 		return;

 	}

 	/* Don't read if we're full */

 	cur_size = evbuffer_get_length(bev->input);

 	read_high = bev->wm_read.high;

 	if (read_high) {

 		if (cur_size >= read_high) {

 			bev_async_del_read(beva);

 			return;

 		}

 		at_most = read_high - cur_size;

 	} else {

 		at_most = 16384; /* FIXME totally magic. */

 	}

 	/* XXXX This over-commits. */

 	/* XXXX see also not above on cast on _bufferevent_get_write_max() */

 	limit = (int)_bufferevent_get_read_max(&beva->bev);

 	if (at_most >= (size_t)limit && limit >= 0)

 		at_most = limit;

 	if (beva->bev.read_suspended) {

 		bev_async_del_read(beva);

 		return;

 	}

 	bufferevent_incref(bev);

 	if (evbuffer_launch_read(bev->input, at_most, &beva->read_overlapped)) {

 		beva->ok = 0;

 		_bufferevent_run_eventcb(bev, BEV_EVENT_ERROR);

 		bufferevent_decref(bev);

 	} else {

 		beva->read_in_progress = at_most;

 		_bufferevent_decrement_read_buckets(&beva->bev, at_most);

 		bev_async_add_read(beva);

 	}

 	return;

 }

 static void

 be_async_outbuf_callback(struct evbuffer *buf,

     const struct evbuffer_cb_info *cbinfo,

     void *arg)

 {

 	struct bufferevent *bev = arg;

 	struct bufferevent_async *bev_async = upcast(bev);

 	/* If we added data to the outbuf and were not writing before,

 	 * we may want to write now. */

 	_bufferevent_incref_and_lock(bev);

 	if (cbinfo->n_added)

 		bev_async_consider_writing(bev_async);

 	_bufferevent_decref_and_unlock(bev);

 }

 static void

 be_async_inbuf_callback(struct evbuffer *buf,

     const struct evbuffer_cb_info *cbinfo,

     void *arg)

 {

 	struct bufferevent *bev = arg;

 	struct bufferevent_async *bev_async = upcast(bev);

 	/* If we drained data from the inbuf and were not reading before,

 	 * we may want to read now */

 	_bufferevent_incref_and_lock(bev);

 	if (cbinfo->n_deleted)

 		bev_async_consider_reading(bev_async);

 	_bufferevent_decref_and_unlock(bev);

 }

 static int

 be_async_enable(struct bufferevent *buf, short what)

 {

 	struct bufferevent_async *bev_async = upcast(buf);

 	if (!bev_async->ok)

 		return -1;

 	if (bev_async->bev.connecting) {

 		/* Don't launch anything during connection attempts. */

 		return 0;

 	}

 	if (what & EV_READ)

 		BEV_RESET_GENERIC_READ_TIMEOUT(buf);

 	if (what & EV_WRITE)

 		BEV_RESET_GENERIC_WRITE_TIMEOUT(buf);

 	/* If we newly enable reading or writing, and we aren't reading or

 	   writing already, consider launching a new read or write. */

 	if (what & EV_READ)

 		bev_async_consider_reading(bev_async);

 	if (what & EV_WRITE)

 		bev_async_consider_writing(bev_async);

 	return 0;

 }

 static int

 be_async_disable(struct bufferevent *bev, short what)

 {

 	struct bufferevent_async *bev_async = upcast(bev);

 	/* XXXX If we disable reading or writing, we may want to consider

 	 * canceling any in-progress read or write operation, though it might

 	 * not work. */

 	if (what & EV_READ) {

 		BEV_DEL_GENERIC_READ_TIMEOUT(bev);

 		bev_async_del_read(bev_async);

 	}

 	if (what & EV_WRITE) {

 		BEV_DEL_GENERIC_WRITE_TIMEOUT(bev);

 		bev_async_del_write(bev_async);

 	}

 	return 0;

 }

 static void

 be_async_destruct(struct bufferevent *bev)

 {

 	struct bufferevent_async *bev_async = upcast(bev);

 	struct bufferevent_private *bev_p = BEV_UPCAST(bev);

 	evutil_socket_t fd;

 	EVUTIL_ASSERT(!upcast(bev)->write_in_progress &&

 			!upcast(bev)->read_in_progress);

 	bev_async_del_read(bev_async);

 	bev_async_del_write(bev_async);

 	fd = _evbuffer_overlapped_get_fd(bev->input);

 	if (bev_p->options & BEV_OPT_CLOSE_ON_FREE) {

 		/* XXXX possible double-close */

 		evutil_closesocket(fd);

 	}

 	/* delete this in case non-blocking connect was used */

 	if (event_initialized(&bev->ev_write)) {

 		event_del(&bev->ev_write);

 		_bufferevent_del_generic_timeout_cbs(bev);

 	}

 }

 /* GetQueuedCompletionStatus doesn't reliably yield WSA error codes, so

  * we use WSAGetOverlappedResult to translate. */

 static void

 bev_async_set_wsa_error(struct bufferevent *bev, struct event_overlapped *eo)

 {

 	DWORD bytes, flags;

 	evutil_socket_t fd;

 	fd = _evbuffer_overlapped_get_fd(bev->input);

 	WSAGetOverlappedResult(fd, &eo->overlapped, &bytes, FALSE, &flags);

 }

 static int

 be_async_flush(struct bufferevent *bev, short what,

     enum bufferevent_flush_mode mode)

 {

 	return 0;

 }

 static void

 connect_complete(struct event_overlapped *eo, ev_uintptr_t key,

     ev_ssize_t nbytes, int ok)

 {

 	struct bufferevent_async *bev_a = upcast_connect(eo);

 	struct bufferevent *bev = &bev_a->bev.bev;

 	evutil_socket_t sock;

 	BEV_LOCK(bev);

 	EVUTIL_ASSERT(bev_a->bev.connecting);

 	bev_a->bev.connecting = 0;

 	sock = _evbuffer_overlapped_get_fd(bev_a->bev.bev.input);

 	/* XXXX Handle error? */

 	setsockopt(sock, SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT, NULL, 0);

 	if (ok)

 		bufferevent_async_set_connected(bev);

 	else

 		bev_async_set_wsa_error(bev, eo);

 	_bufferevent_run_eventcb(bev,

 			ok? BEV_EVENT_CONNECTED : BEV_EVENT_ERROR);

 	event_base_del_virtual(bev->ev_base);

 	_bufferevent_decref_and_unlock(bev);

 }

 static void

 read_complete(struct event_overlapped *eo, ev_uintptr_t key,

     ev_ssize_t nbytes, int ok)

 {

 	struct bufferevent_async *bev_a = upcast_read(eo);

 	struct bufferevent *bev = &bev_a->bev.bev;

 	short what = BEV_EVENT_READING;

 	ev_ssize_t amount_unread;

 	BEV_LOCK(bev);

 	EVUTIL_ASSERT(bev_a->read_in_progress);

 	amount_unread = bev_a->read_in_progress - nbytes;

 	evbuffer_commit_read(bev->input, nbytes);

 	bev_a->read_in_progress = 0;

 	if (amount_unread)

 		_bufferevent_decrement_read_buckets(&bev_a->bev, -amount_unread);

 	if (!ok)

 		bev_async_set_wsa_error(bev, eo);

 	if (bev_a->ok) {

 		if (ok && nbytes) {

 			BEV_RESET_GENERIC_READ_TIMEOUT(bev);

 			if (evbuffer_get_length(bev->input) >= bev->wm_read.low)

 				_bufferevent_run_readcb(bev);

 			bev_async_consider_reading(bev_a);

 		} else if (!ok) {

 			what |= BEV_EVENT_ERROR;

 			bev_a->ok = 0;

 			_bufferevent_run_eventcb(bev, what);

 		} else if (!nbytes) {

 			what |= BEV_EVENT_EOF;

 			bev_a->ok = 0;

 			_bufferevent_run_eventcb(bev, what);

 		}

 	}

 	_bufferevent_decref_and_unlock(bev);

 }

 static void

 write_complete(struct event_overlapped *eo, ev_uintptr_t key,

     ev_ssize_t nbytes, int ok)

 {

 	struct bufferevent_async *bev_a = upcast_write(eo);

 	struct bufferevent *bev = &bev_a->bev.bev;

 	short what = BEV_EVENT_WRITING;

 	ev_ssize_t amount_unwritten;

 	BEV_LOCK(bev);

 	EVUTIL_ASSERT(bev_a->write_in_progress);

 	amount_unwritten = bev_a->write_in_progress - nbytes;

 	evbuffer_commit_write(bev->output, nbytes);

 	bev_a->write_in_progress = 0;

 	if (amount_unwritten)

 		_bufferevent_decrement_write_buckets(&bev_a->bev,

 		                                     -amount_unwritten);

 	if (!ok)

 		bev_async_set_wsa_error(bev, eo);

 	if (bev_a->ok) {

 		if (ok && nbytes) {

 			BEV_RESET_GENERIC_WRITE_TIMEOUT(bev);

 			if (evbuffer_get_length(bev->output) <=

 			    bev->wm_write.low)

 				_bufferevent_run_writecb(bev);

 			bev_async_consider_writing(bev_a);

 		} else if (!ok) {

 			what |= BEV_EVENT_ERROR;

 			bev_a->ok = 0;

 			_bufferevent_run_eventcb(bev, what);

 		} else if (!nbytes) {

 			what |= BEV_EVENT_EOF;

 			bev_a->ok = 0;

 			_bufferevent_run_eventcb(bev, what);

 		}

 	}

 	_bufferevent_decref_and_unlock(bev);

 }

 struct bufferevent *

 bufferevent_async_new(struct event_base *base,

     evutil_socket_t fd, int options)

 {

 	struct bufferevent_async *bev_a;

 	struct bufferevent *bev;

 	struct event_iocp_port *iocp;

 	options |= BEV_OPT_THREADSAFE;

 	if (!(iocp = event_base_get_iocp(base)))

 		return NULL;

 	if (fd >= 0 && event_iocp_port_associate(iocp, fd, 1)<0) {

 		int err = GetLastError();

 		/* We may have alrady associated this fd with a port.

 		 * Let's hope it's this port, and that the error code

 		 * for doing this neer changes. */

 		if (err != ERROR_INVALID_PARAMETER)

 			return NULL;

 	}

 	if (!(bev_a = mm_calloc(1, sizeof(struct bufferevent_async))))

 		return NULL;

 	bev = &bev_a->bev.bev;

 	if (!(bev->input = evbuffer_overlapped_new(fd))) {

 		mm_free(bev_a);

 		return NULL;

 	}

 	if (!(bev->output = evbuffer_overlapped_new(fd))) {

 		evbuffer_free(bev->input);

 		mm_free(bev_a);

 		return NULL;

 	}

 	if (bufferevent_init_common(&bev_a->bev, base, &bufferevent_ops_async,

 		options)<0)

 		goto err;

 	evbuffer_add_cb(bev->input, be_async_inbuf_callback, bev);

 	evbuffer_add_cb(bev->output, be_async_outbuf_callback, bev);

 	event_overlapped_init(&bev_a->connect_overlapped, connect_complete);

 	event_overlapped_init(&bev_a->read_overlapped, read_complete);

 	event_overlapped_init(&bev_a->write_overlapped, write_complete);

 	bev_a->ok = fd >= 0;

 	if (bev_a->ok)

 		_bufferevent_init_generic_timeout_cbs(bev);

 	return bev;

 err:

 	bufferevent_free(&bev_a->bev.bev);

 	return NULL;

 }

 void

 bufferevent_async_set_connected(struct bufferevent *bev)

 {

 	struct bufferevent_async *bev_async = upcast(bev);

 	bev_async->ok = 1;

 	_bufferevent_init_generic_timeout_cbs(bev);

 	/* Now's a good time to consider reading/writing */

 	be_async_enable(bev, bev->enabled);

 }

 int

 bufferevent_async_can_connect(struct bufferevent *bev)

 {

 	const struct win32_extension_fns *ext =

 	    event_get_win32_extension_fns();

 	if (BEV_IS_ASYNC(bev) &&

 	    event_base_get_iocp(bev->ev_base) &&

 	    ext && ext->ConnectEx)

 		return 1;

 	return 0;

 }

 int

 bufferevent_async_connect(struct bufferevent *bev, evutil_socket_t fd,

 	const struct sockaddr *sa, int socklen)

 {

 	BOOL rc;

 	struct bufferevent_async *bev_async = upcast(bev);

 	struct sockaddr_storage ss;

 	const struct win32_extension_fns *ext =

 	    event_get_win32_extension_fns();

 	EVUTIL_ASSERT(ext && ext->ConnectEx && fd >= 0 && sa != NULL);

 	/* ConnectEx() requires that the socket be bound to an address

 	 * with bind() before using, otherwise it will fail. We attempt

 	 * to issue a bind() here, taking into account that the error

 	 * code is set to WSAEINVAL when the socket is already bound. */

 	memset(&ss, 0, sizeof(ss));

 	if (sa->sa_family == AF_INET) {

 		struct sockaddr_in *sin = (struct sockaddr_in *)&ss;

 		sin->sin_family = AF_INET;

 		sin->sin_addr.s_addr = INADDR_ANY;

 	} else if (sa->sa_family == AF_INET6) {

 		struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&ss;

 		sin6->sin6_family = AF_INET6;

 		sin6->sin6_addr = in6addr_any;

 	} else {

 		/* Well, the user will have to bind() */

 		return -1;

 	}

 	if (bind(fd, (struct sockaddr *)&ss, sizeof(ss)) < 0 &&

 	    WSAGetLastError() != WSAEINVAL)

 		return -1;

 	event_base_add_virtual(bev->ev_base);

 	bufferevent_incref(bev);

 	rc = ext->ConnectEx(fd, sa, socklen, NULL, 0, NULL,

 			    &bev_async->connect_overlapped.overlapped);

 	if (rc || WSAGetLastError() == ERROR_IO_PENDING)

 		return 0;

 	event_base_del_virtual(bev->ev_base);

 	bufferevent_decref(bev);

 	return -1;

 }

 static int

 be_async_ctrl(struct bufferevent *bev, enum bufferevent_ctrl_op op,

     union bufferevent_ctrl_data *data)

 {

 	switch (op) {

 	case BEV_CTRL_GET_FD:

 		data->fd = _evbuffer_overlapped_get_fd(bev->input);

 		return 0;

 	case BEV_CTRL_SET_FD: {

 		struct event_iocp_port *iocp;

 		if (data->fd == _evbuffer_overlapped_get_fd(bev->input))

 			return 0;

 		if (!(iocp = event_base_get_iocp(bev->ev_base)))

 			return -1;

 		if (event_iocp_port_associate(iocp, data->fd, 1) < 0)

 			return -1;

 		_evbuffer_overlapped_set_fd(bev->input, data->fd);

 		_evbuffer_overlapped_set_fd(bev->output, data->fd);

 		return 0;

 	}

 	case BEV_CTRL_CANCEL_ALL: {

 		struct bufferevent_async *bev_a = upcast(bev);

 		evutil_socket_t fd = _evbuffer_overlapped_get_fd(bev->input);

 		if (fd != (evutil_socket_t)INVALID_SOCKET &&

 		    (bev_a->bev.options & BEV_OPT_CLOSE_ON_FREE)) {

 			closesocket(fd);

 		}

 		bev_a->ok = 0;

 		return 0;

 	}

 	case BEV_CTRL_GET_UNDERLYING:

 	default:

 		return -1;

 	}

 }