diff options
author | alexv-smirnov <alex@ydb.tech> | 2022-12-20 00:50:48 +0300 |
---|---|---|
committer | alexv-smirnov <alex@ydb.tech> | 2022-12-20 00:50:48 +0300 |
commit | 84f2cfa253cc618438ed6e9d68b33fa7c0d88cb9 (patch) | |
tree | f0cf2236e0aafb3e437199f1ac7b559e7fad554a /contrib/libs/libevent | |
parent | bde6febc1ad3b826e72746de21d7250803e8e0b5 (diff) | |
download | ydb-84f2cfa253cc618438ed6e9d68b33fa7c0d88cb9.tar.gz |
add windows platform to ydb github export
Diffstat (limited to 'contrib/libs/libevent')
-rw-r--r-- | contrib/libs/libevent/buffer_iocp.c | 327 | ||||
-rw-r--r-- | contrib/libs/libevent/bufferevent_async.c | 706 | ||||
-rw-r--r-- | contrib/libs/libevent/event_iocp.c | 294 | ||||
-rw-r--r-- | contrib/libs/libevent/evthread_win32.c | 341 | ||||
-rw-r--r-- | contrib/libs/libevent/win32select.c | 389 |
5 files changed, 2057 insertions, 0 deletions
diff --git a/contrib/libs/libevent/buffer_iocp.c b/contrib/libs/libevent/buffer_iocp.c new file mode 100644 index 00000000000..2af0c49cc64 --- /dev/null +++ b/contrib/libs/libevent/buffer_iocp.c @@ -0,0 +1,327 @@ +/* + * Copyright (c) 2009-2012 Niels Provos and Nick Mathewson + * + * 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. + */ + +/** + @file buffer_iocp.c + + This module implements overlapped read and write functions for evbuffer + objects on Windows. +*/ +#include "event2/event-config.h" +#include "evconfig-private.h" + +#include "event2/buffer.h" +#include "event2/buffer_compat.h" +#include "event2/util.h" +#include "event2/thread.h" +#include "util-internal.h" +#include "evthread-internal.h" +#include "evbuffer-internal.h" +#include "iocp-internal.h" +#include "mm-internal.h" + +#include <winsock2.h> +#include <winerror.h> +#include <windows.h> +#include <stdio.h> + +#define MAX_WSABUFS 16 + +/** An evbuffer that can handle overlapped IO. */ +struct evbuffer_overlapped { + struct evbuffer buffer; + /** The socket that we're doing overlapped IO on. */ + evutil_socket_t fd; + + /** pending I/O type */ + unsigned read_in_progress : 1; + unsigned write_in_progress : 1; + + /** The first pinned chain in the buffer. */ + struct evbuffer_chain *first_pinned; + + /** How many chains are pinned; how many of the fields in buffers + * are we using. */ + int n_buffers; + WSABUF buffers[MAX_WSABUFS]; +}; + +/** Given an evbuffer, return the correponding evbuffer structure, or NULL if + * the evbuffer isn't overlapped. */ +static inline struct evbuffer_overlapped * +upcast_evbuffer(struct evbuffer *buf) +{ + if (!buf || !buf->is_overlapped) + return NULL; + return EVUTIL_UPCAST(buf, struct evbuffer_overlapped, buffer); +} + +/** Unpin all the chains noted as pinned in 'eo'. */ +static void +pin_release(struct evbuffer_overlapped *eo, unsigned flag) +{ + int i; + struct evbuffer_chain *next, *chain = eo->first_pinned; + + for (i = 0; i < eo->n_buffers; ++i) { + EVUTIL_ASSERT(chain); + next = chain->next; + evbuffer_chain_unpin_(chain, flag); + chain = next; + } +} + +void +evbuffer_commit_read_(struct evbuffer *evbuf, ev_ssize_t nBytes) +{ + struct evbuffer_overlapped *buf = upcast_evbuffer(evbuf); + struct evbuffer_chain **chainp; + size_t remaining, len; + unsigned i; + + EVBUFFER_LOCK(evbuf); + EVUTIL_ASSERT(buf->read_in_progress && !buf->write_in_progress); + EVUTIL_ASSERT(nBytes >= 0); /* XXXX Can this be false? */ + + evbuffer_unfreeze(evbuf, 0); + + chainp = evbuf->last_with_datap; + if (!((*chainp)->flags & EVBUFFER_MEM_PINNED_R)) + chainp = &(*chainp)->next; + remaining = nBytes; + for (i = 0; remaining > 0 && i < (unsigned)buf->n_buffers; ++i) { + EVUTIL_ASSERT(*chainp); + len = buf->buffers[i].len; + if (remaining < len) + len = remaining; + (*chainp)->off += len; + evbuf->last_with_datap = chainp; + remaining -= len; + chainp = &(*chainp)->next; + } + + pin_release(buf, EVBUFFER_MEM_PINNED_R); + + buf->read_in_progress = 0; + + evbuf->total_len += nBytes; + evbuf->n_add_for_cb += nBytes; + + evbuffer_invoke_callbacks_(evbuf); + + evbuffer_decref_and_unlock_(evbuf); +} + +void +evbuffer_commit_write_(struct evbuffer *evbuf, ev_ssize_t nBytes) +{ + struct evbuffer_overlapped *buf = upcast_evbuffer(evbuf); + + EVBUFFER_LOCK(evbuf); + EVUTIL_ASSERT(buf->write_in_progress && !buf->read_in_progress); + evbuffer_unfreeze(evbuf, 1); + evbuffer_drain(evbuf, nBytes); + pin_release(buf,EVBUFFER_MEM_PINNED_W); + buf->write_in_progress = 0; + evbuffer_decref_and_unlock_(evbuf); +} + +struct evbuffer * +evbuffer_overlapped_new_(evutil_socket_t fd) +{ + struct evbuffer_overlapped *evo; + + evo = mm_calloc(1, sizeof(struct evbuffer_overlapped)); + if (!evo) + return NULL; + + LIST_INIT(&evo->buffer.callbacks); + evo->buffer.refcnt = 1; + evo->buffer.last_with_datap = &evo->buffer.first; + + evo->buffer.is_overlapped = 1; + evo->fd = fd; + + return &evo->buffer; +} + +int +evbuffer_launch_write_(struct evbuffer *buf, ev_ssize_t at_most, + struct event_overlapped *ol) +{ + struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf); + int r = -1; + int i; + struct evbuffer_chain *chain; + DWORD bytesSent; + + if (!buf) { + /* No buffer, or it isn't overlapped */ + return -1; + } + + EVBUFFER_LOCK(buf); + EVUTIL_ASSERT(!buf_o->read_in_progress); + if (buf->freeze_start || buf_o->write_in_progress) + goto done; + if (!buf->total_len) { + /* Nothing to write */ + r = 0; + goto done; + } else if (at_most < 0 || (size_t)at_most > buf->total_len) { + at_most = buf->total_len; + } + evbuffer_freeze(buf, 1); + + buf_o->first_pinned = NULL; + buf_o->n_buffers = 0; + memset(buf_o->buffers, 0, sizeof(buf_o->buffers)); + + chain = buf_o->first_pinned = buf->first; + + for (i=0; i < MAX_WSABUFS && chain; ++i, chain=chain->next) { + WSABUF *b = &buf_o->buffers[i]; + b->buf = (char*)( chain->buffer + chain->misalign ); + evbuffer_chain_pin_(chain, EVBUFFER_MEM_PINNED_W); + + if ((size_t)at_most > chain->off) { + /* XXXX Cast is safe for now, since win32 has no + mmaped chains. But later, we need to have this + add more WSAbufs if chain->off is greater than + ULONG_MAX */ + b->len = (unsigned long)chain->off; + at_most -= chain->off; + } else { + b->len = (unsigned long)at_most; + ++i; + break; + } + } + + buf_o->n_buffers = i; + evbuffer_incref_(buf); + if (WSASend(buf_o->fd, buf_o->buffers, i, &bytesSent, 0, + &ol->overlapped, NULL)) { + int error = WSAGetLastError(); + if (error != WSA_IO_PENDING) { + /* An actual error. */ + pin_release(buf_o, EVBUFFER_MEM_PINNED_W); + evbuffer_unfreeze(buf, 1); + evbuffer_free(buf); /* decref */ + goto done; + } + } + + buf_o->write_in_progress = 1; + r = 0; +done: + EVBUFFER_UNLOCK(buf); + return r; +} + +int +evbuffer_launch_read_(struct evbuffer *buf, size_t at_most, + struct event_overlapped *ol) +{ + struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf); + int r = -1, i; + int nvecs; + int npin=0; + struct evbuffer_chain *chain=NULL, **chainp; + DWORD bytesRead; + DWORD flags = 0; + struct evbuffer_iovec vecs[MAX_WSABUFS]; + + if (!buf_o) + return -1; + EVBUFFER_LOCK(buf); + EVUTIL_ASSERT(!buf_o->write_in_progress); + if (buf->freeze_end || buf_o->read_in_progress) + goto done; + + buf_o->first_pinned = NULL; + buf_o->n_buffers = 0; + memset(buf_o->buffers, 0, sizeof(buf_o->buffers)); + + if (evbuffer_expand_fast_(buf, at_most, MAX_WSABUFS) == -1) + goto done; + evbuffer_freeze(buf, 0); + + nvecs = evbuffer_read_setup_vecs_(buf, at_most, + vecs, MAX_WSABUFS, &chainp, 1); + for (i=0;i<nvecs;++i) { + WSABUF_FROM_EVBUFFER_IOV( + &buf_o->buffers[i], + &vecs[i]); + } + + buf_o->n_buffers = nvecs; + buf_o->first_pinned = chain = *chainp; + + npin=0; + for ( ; chain; chain = chain->next) { + evbuffer_chain_pin_(chain, EVBUFFER_MEM_PINNED_R); + ++npin; + } + EVUTIL_ASSERT(npin == nvecs); + + evbuffer_incref_(buf); + if (WSARecv(buf_o->fd, buf_o->buffers, nvecs, &bytesRead, &flags, + &ol->overlapped, NULL)) { + int error = WSAGetLastError(); + if (error != WSA_IO_PENDING) { + /* An actual error. */ + pin_release(buf_o, EVBUFFER_MEM_PINNED_R); + evbuffer_unfreeze(buf, 0); + evbuffer_free(buf); /* decref */ + goto done; + } + } + + buf_o->read_in_progress = 1; + r = 0; +done: + EVBUFFER_UNLOCK(buf); + return r; +} + +evutil_socket_t +evbuffer_overlapped_get_fd_(struct evbuffer *buf) +{ + struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf); + return buf_o ? buf_o->fd : -1; +} + +void +evbuffer_overlapped_set_fd_(struct evbuffer *buf, evutil_socket_t fd) +{ + struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf); + EVBUFFER_LOCK(buf); + /* XXX is this right?, should it cancel current I/O operations? */ + if (buf_o) + buf_o->fd = fd; + EVBUFFER_UNLOCK(buf); +} diff --git a/contrib/libs/libevent/bufferevent_async.c b/contrib/libs/libevent/bufferevent_async.c new file mode 100644 index 00000000000..40c7c5e8d0d --- /dev/null +++ b/contrib/libs/libevent/bufferevent_async.c @@ -0,0 +1,706 @@ +/* + * 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" +#include "evconfig-private.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 <winerror.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, + NULL, /* Unlink */ + be_async_destruct, + bufferevent_generic_adj_timeouts_, + be_async_flush, + be_async_ctrl, +}; + +static inline void +be_async_run_eventcb(struct bufferevent *bev, short what, int options) +{ bufferevent_run_eventcb_(bev, what, options|BEV_TRIG_DEFER_CALLBACKS); } + +static inline void +be_async_trigger_nolock(struct bufferevent *bev, short what, int options) +{ bufferevent_trigger_nolock_(bev, what, options|BEV_TRIG_DEFER_CALLBACKS); } + +static inline int +fatal_error(int err) +{ + switch (err) { + /* We may have already associated this fd with a port. + * Let's hope it's this port, and that the error code + * for doing this neer changes. */ + case ERROR_INVALID_PARAMETER: + return 0; + } + return 1; +} + +static inline struct bufferevent_async * +upcast(struct bufferevent *bev) +{ + struct bufferevent_async *bev_a; + if (!BEV_IS_ASYNC(bev)) + 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; + be_async_run_eventcb(bev, BEV_EVENT_ERROR, 0); + } 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; + be_async_run_eventcb(bev, BEV_EVENT_ERROR, 0); + 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 (fd != (evutil_socket_t)EVUTIL_INVALID_SOCKET && + (bev_p->options & BEV_OPT_CLOSE_ON_FREE)) { + evutil_closesocket(fd); + evbuffer_overlapped_set_fd_(bev->input, EVUTIL_INVALID_SOCKET); + } +} + +/* 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); + + be_async_run_eventcb(bev, ok ? BEV_EVENT_CONNECTED : BEV_EVENT_ERROR, 0); + + 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); + be_async_trigger_nolock(bev, EV_READ, 0); + bev_async_consider_reading(bev_a); + } else if (!ok) { + what |= BEV_EVENT_ERROR; + bev_a->ok = 0; + be_async_run_eventcb(bev, what, 0); + } else if (!nbytes) { + what |= BEV_EVENT_EOF; + bev_a->ok = 0; + be_async_run_eventcb(bev, what, 0); + } + } + + 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); + be_async_trigger_nolock(bev, EV_WRITE, 0); + bev_async_consider_writing(bev_a); + } else if (!ok) { + what |= BEV_EVENT_ERROR; + bev_a->ok = 0; + be_async_run_eventcb(bev, what, 0); + } else if (!nbytes) { + what |= BEV_EVENT_EOF; + bev_a->ok = 0; + be_async_run_eventcb(bev, what, 0); + } + } + + 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) { + if (fatal_error(GetLastError())) + 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); + + bufferevent_init_generic_timeout_cbs_(bev); + + bev_a->ok = fd >= 0; + + 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; + /* 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 bufferevent_async *bev_a = upcast(bev); + 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) { + if (fatal_error(GetLastError())) + return -1; + } + evbuffer_overlapped_set_fd_(bev->input, data->fd); + evbuffer_overlapped_set_fd_(bev->output, data->fd); + bev_a->ok = data->fd >= 0; + 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)EVUTIL_INVALID_SOCKET && + (bev_a->bev.options & BEV_OPT_CLOSE_ON_FREE)) { + closesocket(fd); + evbuffer_overlapped_set_fd_(bev->input, EVUTIL_INVALID_SOCKET); + } + bev_a->ok = 0; + return 0; + } + case BEV_CTRL_GET_UNDERLYING: + default: + return -1; + } +} + + diff --git a/contrib/libs/libevent/event_iocp.c b/contrib/libs/libevent/event_iocp.c new file mode 100644 index 00000000000..6b2a2e15ef8 --- /dev/null +++ b/contrib/libs/libevent/event_iocp.c @@ -0,0 +1,294 @@ +/* + * Copyright (c) 2009-2012 Niels Provos, Nick Mathewson + * + * 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 "evconfig-private.h" + +#ifndef _WIN32_WINNT +/* Minimum required for InitializeCriticalSectionAndSpinCount */ +#define _WIN32_WINNT 0x0403 +#endif +#include <winsock2.h> +#include <windows.h> +#include <process.h> +#include <stdio.h> +#include <mswsock.h> + +#include "event2/util.h" +#include "util-internal.h" +#include "iocp-internal.h" +#include "log-internal.h" +#include "mm-internal.h" +#include "event-internal.h" +#include "evthread-internal.h" + +#define NOTIFICATION_KEY ((ULONG_PTR)-1) + +void +event_overlapped_init_(struct event_overlapped *o, iocp_callback cb) +{ + memset(o, 0, sizeof(struct event_overlapped)); + o->cb = cb; +} + +static void +handle_entry(OVERLAPPED *o, ULONG_PTR completion_key, DWORD nBytes, int ok) +{ + struct event_overlapped *eo = + EVUTIL_UPCAST(o, struct event_overlapped, overlapped); + eo->cb(eo, completion_key, nBytes, ok); +} + +static void +loop(void *port_) +{ + struct event_iocp_port *port = port_; + long ms = port->ms; + HANDLE p = port->port; + + if (ms <= 0) + ms = INFINITE; + + while (1) { + OVERLAPPED *overlapped=NULL; + ULONG_PTR key=0; + DWORD bytes=0; + int ok = GetQueuedCompletionStatus(p, &bytes, &key, + &overlapped, ms); + EnterCriticalSection(&port->lock); + if (port->shutdown) { + if (--port->n_live_threads == 0) + ReleaseSemaphore(port->shutdownSemaphore, 1, + NULL); + LeaveCriticalSection(&port->lock); + return; + } + LeaveCriticalSection(&port->lock); + + if (key != NOTIFICATION_KEY && overlapped) + handle_entry(overlapped, key, bytes, ok); + else if (!overlapped) + break; + } + event_warnx("GetQueuedCompletionStatus exited with no event."); + EnterCriticalSection(&port->lock); + if (--port->n_live_threads == 0) + ReleaseSemaphore(port->shutdownSemaphore, 1, NULL); + LeaveCriticalSection(&port->lock); +} + +int +event_iocp_port_associate_(struct event_iocp_port *port, evutil_socket_t fd, + ev_uintptr_t key) +{ + HANDLE h; + h = CreateIoCompletionPort((HANDLE)fd, port->port, key, port->n_threads); + if (!h) + return -1; + return 0; +} + +static void * +get_extension_function(SOCKET s, const GUID *which_fn) +{ + void *ptr = NULL; + DWORD bytes=0; + WSAIoctl(s, SIO_GET_EXTENSION_FUNCTION_POINTER, + (GUID*)which_fn, sizeof(*which_fn), + &ptr, sizeof(ptr), + &bytes, NULL, NULL); + + /* No need to detect errors here: if ptr is set, then we have a good + function pointer. Otherwise, we should behave as if we had no + function pointer. + */ + return ptr; +} + +/* Mingw doesn't have these in its mswsock.h. The values are copied from + wine.h. Perhaps if we copy them exactly, the cargo will come again. +*/ +#ifndef WSAID_ACCEPTEX +#define WSAID_ACCEPTEX \ + {0xb5367df1,0xcbac,0x11cf,{0x95,0xca,0x00,0x80,0x5f,0x48,0xa1,0x92}} +#endif +#ifndef WSAID_CONNECTEX +#define WSAID_CONNECTEX \ + {0x25a207b9,0xddf3,0x4660,{0x8e,0xe9,0x76,0xe5,0x8c,0x74,0x06,0x3e}} +#endif +#ifndef WSAID_GETACCEPTEXSOCKADDRS +#define WSAID_GETACCEPTEXSOCKADDRS \ + {0xb5367df2,0xcbac,0x11cf,{0x95,0xca,0x00,0x80,0x5f,0x48,0xa1,0x92}} +#endif + +static int extension_fns_initialized = 0; + +static void +init_extension_functions(struct win32_extension_fns *ext) +{ + const GUID acceptex = WSAID_ACCEPTEX; + const GUID connectex = WSAID_CONNECTEX; + const GUID getacceptexsockaddrs = WSAID_GETACCEPTEXSOCKADDRS; + SOCKET s = socket(AF_INET, SOCK_STREAM, 0); + if (s == EVUTIL_INVALID_SOCKET) + return; + ext->AcceptEx = get_extension_function(s, &acceptex); + ext->ConnectEx = get_extension_function(s, &connectex); + ext->GetAcceptExSockaddrs = get_extension_function(s, + &getacceptexsockaddrs); + closesocket(s); + + extension_fns_initialized = 1; +} + +static struct win32_extension_fns the_extension_fns; + +const struct win32_extension_fns * +event_get_win32_extension_fns_(void) +{ + return &the_extension_fns; +} + +#define N_CPUS_DEFAULT 2 + +struct event_iocp_port * +event_iocp_port_launch_(int n_cpus) +{ + struct event_iocp_port *port; + int i; + + if (!extension_fns_initialized) + init_extension_functions(&the_extension_fns); + + if (!(port = mm_calloc(1, sizeof(struct event_iocp_port)))) + return NULL; + + if (n_cpus <= 0) + n_cpus = N_CPUS_DEFAULT; + port->n_threads = n_cpus * 2; + port->threads = mm_calloc(port->n_threads, sizeof(HANDLE)); + if (!port->threads) + goto err; + + port->port = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, + n_cpus); + port->ms = -1; + if (!port->port) + goto err; + + port->shutdownSemaphore = CreateSemaphore(NULL, 0, 1, NULL); + if (!port->shutdownSemaphore) + goto err; + + for (i=0; i<port->n_threads; ++i) { + ev_uintptr_t th = _beginthread(loop, 0, port); + if (th == (ev_uintptr_t)-1) + goto err; + port->threads[i] = (HANDLE)th; + ++port->n_live_threads; + } + + InitializeCriticalSectionAndSpinCount(&port->lock, 1000); + + return port; +err: + if (port->port) + CloseHandle(port->port); + if (port->threads) + mm_free(port->threads); + if (port->shutdownSemaphore) + CloseHandle(port->shutdownSemaphore); + mm_free(port); + return NULL; +} + +static void +event_iocp_port_unlock_and_free_(struct event_iocp_port *port) +{ + DeleteCriticalSection(&port->lock); + CloseHandle(port->port); + CloseHandle(port->shutdownSemaphore); + mm_free(port->threads); + mm_free(port); +} + +static int +event_iocp_notify_all(struct event_iocp_port *port) +{ + int i, r, ok=1; + for (i=0; i<port->n_threads; ++i) { + r = PostQueuedCompletionStatus(port->port, 0, NOTIFICATION_KEY, + NULL); + if (!r) + ok = 0; + } + return ok ? 0 : -1; +} + +int +event_iocp_shutdown_(struct event_iocp_port *port, long waitMsec) +{ + DWORD ms = INFINITE; + int n; + + EnterCriticalSection(&port->lock); + port->shutdown = 1; + LeaveCriticalSection(&port->lock); + event_iocp_notify_all(port); + + if (waitMsec >= 0) + ms = waitMsec; + + WaitForSingleObject(port->shutdownSemaphore, ms); + EnterCriticalSection(&port->lock); + n = port->n_live_threads; + LeaveCriticalSection(&port->lock); + if (n == 0) { + event_iocp_port_unlock_and_free_(port); + return 0; + } else { + return -1; + } +} + +int +event_iocp_activate_overlapped_( + struct event_iocp_port *port, struct event_overlapped *o, + ev_uintptr_t key, ev_uint32_t n) +{ + BOOL r; + + r = PostQueuedCompletionStatus(port->port, n, key, &o->overlapped); + return (r==0) ? -1 : 0; +} + +struct event_iocp_port * +event_base_get_iocp_(struct event_base *base) +{ +#ifdef _WIN32 + return base->iocp; +#else + return NULL; +#endif +} diff --git a/contrib/libs/libevent/evthread_win32.c b/contrib/libs/libevent/evthread_win32.c new file mode 100644 index 00000000000..2ec80560a5c --- /dev/null +++ b/contrib/libs/libevent/evthread_win32.c @@ -0,0 +1,341 @@ +/* + * Copyright 2009-2012 Niels Provos and Nick Mathewson + * + * 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" +#include "evconfig-private.h" + +#ifdef _WIN32 +#ifndef _WIN32_WINNT +/* Minimum required for InitializeCriticalSectionAndSpinCount */ +#define _WIN32_WINNT 0x0403 +#endif +#include <winsock2.h> +#define WIN32_LEAN_AND_MEAN +#include <windows.h> +#undef WIN32_LEAN_AND_MEAN +#include <sys/locking.h> +#endif + +struct event_base; +#include "event2/thread.h" + +#include "mm-internal.h" +#include "evthread-internal.h" +#include "time-internal.h" + +#define SPIN_COUNT 2000 + +static void * +evthread_win32_lock_create(unsigned locktype) +{ + CRITICAL_SECTION *lock = mm_malloc(sizeof(CRITICAL_SECTION)); + if (!lock) + return NULL; + if (InitializeCriticalSectionAndSpinCount(lock, SPIN_COUNT) == 0) { + mm_free(lock); + return NULL; + } + return lock; +} + +static void +evthread_win32_lock_free(void *lock_, unsigned locktype) +{ + CRITICAL_SECTION *lock = lock_; + DeleteCriticalSection(lock); + mm_free(lock); +} + +static int +evthread_win32_lock(unsigned mode, void *lock_) +{ + CRITICAL_SECTION *lock = lock_; + if ((mode & EVTHREAD_TRY)) { + return ! TryEnterCriticalSection(lock); + } else { + EnterCriticalSection(lock); + return 0; + } +} + +static int +evthread_win32_unlock(unsigned mode, void *lock_) +{ + CRITICAL_SECTION *lock = lock_; + LeaveCriticalSection(lock); + return 0; +} + +static unsigned long +evthread_win32_get_id(void) +{ + return (unsigned long) GetCurrentThreadId(); +} + +#ifdef WIN32_HAVE_CONDITION_VARIABLES +static void WINAPI (*InitializeConditionVariable_fn)(PCONDITION_VARIABLE) + = NULL; +static BOOL WINAPI (*SleepConditionVariableCS_fn)( + PCONDITION_VARIABLE, PCRITICAL_SECTION, DWORD) = NULL; +static void WINAPI (*WakeAllConditionVariable_fn)(PCONDITION_VARIABLE) = NULL; +static void WINAPI (*WakeConditionVariable_fn)(PCONDITION_VARIABLE) = NULL; + +static int +evthread_win32_condvar_init(void) +{ + HANDLE lib; + + lib = GetModuleHandle(TEXT("kernel32.dll")); + if (lib == NULL) + return 0; + +#define LOAD(name) \ + name##_fn = GetProcAddress(lib, #name) + LOAD(InitializeConditionVariable); + LOAD(SleepConditionVariableCS); + LOAD(WakeAllConditionVariable); + LOAD(WakeConditionVariable); + + return InitializeConditionVariable_fn && SleepConditionVariableCS_fn && + WakeAllConditionVariable_fn && WakeConditionVariable_fn; +} + +/* XXXX Even if we can build this, we don't necessarily want to: the functions + * in question didn't exist before Vista, so we'd better LoadProc them. */ +static void * +evthread_win32_condvar_alloc(unsigned condflags) +{ + CONDITION_VARIABLE *cond = mm_malloc(sizeof(CONDITION_VARIABLE)); + if (!cond) + return NULL; + InitializeConditionVariable_fn(cond); + return cond; +} + +static void +evthread_win32_condvar_free(void *cond_) +{ + CONDITION_VARIABLE *cond = cond_; + /* There doesn't _seem_ to be a cleaup fn here... */ + mm_free(cond); +} + +static int +evthread_win32_condvar_signal(void *cond, int broadcast) +{ + CONDITION_VARIABLE *cond = cond_; + if (broadcast) + WakeAllConditionVariable_fn(cond); + else + WakeConditionVariable_fn(cond); + return 0; +} + +static int +evthread_win32_condvar_wait(void *cond_, void *lock_, const struct timeval *tv) +{ + CONDITION_VARIABLE *cond = cond_; + CRITICAL_SECTION *lock = lock_; + DWORD ms, err; + BOOL result; + + if (tv) + ms = evutil_tv_to_msec_(tv); + else + ms = INFINITE; + result = SleepConditionVariableCS_fn(cond, lock, ms); + if (result) { + if (GetLastError() == WAIT_TIMEOUT) + return 1; + else + return -1; + } else { + return 0; + } +} +#endif + +struct evthread_win32_cond { + HANDLE event; + + CRITICAL_SECTION lock; + int n_waiting; + int n_to_wake; + int generation; +}; + +static void * +evthread_win32_cond_alloc(unsigned flags) +{ + struct evthread_win32_cond *cond; + if (!(cond = mm_malloc(sizeof(struct evthread_win32_cond)))) + return NULL; + if (InitializeCriticalSectionAndSpinCount(&cond->lock, SPIN_COUNT)==0) { + mm_free(cond); + return NULL; + } + if ((cond->event = CreateEvent(NULL,TRUE,FALSE,NULL)) == NULL) { + DeleteCriticalSection(&cond->lock); + mm_free(cond); + return NULL; + } + cond->n_waiting = cond->n_to_wake = cond->generation = 0; + return cond; +} + +static void +evthread_win32_cond_free(void *cond_) +{ + struct evthread_win32_cond *cond = cond_; + DeleteCriticalSection(&cond->lock); + CloseHandle(cond->event); + mm_free(cond); +} + +static int +evthread_win32_cond_signal(void *cond_, int broadcast) +{ + struct evthread_win32_cond *cond = cond_; + EnterCriticalSection(&cond->lock); + if (broadcast) + cond->n_to_wake = cond->n_waiting; + else + ++cond->n_to_wake; + cond->generation++; + SetEvent(cond->event); + LeaveCriticalSection(&cond->lock); + return 0; +} + +static int +evthread_win32_cond_wait(void *cond_, void *lock_, const struct timeval *tv) +{ + struct evthread_win32_cond *cond = cond_; + CRITICAL_SECTION *lock = lock_; + int generation_at_start; + int waiting = 1; + int result = -1; + DWORD ms = INFINITE, ms_orig = INFINITE, startTime, endTime; + if (tv) + ms_orig = ms = evutil_tv_to_msec_(tv); + + EnterCriticalSection(&cond->lock); + ++cond->n_waiting; + generation_at_start = cond->generation; + LeaveCriticalSection(&cond->lock); + + LeaveCriticalSection(lock); + + startTime = GetTickCount(); + do { + DWORD res; + res = WaitForSingleObject(cond->event, ms); + EnterCriticalSection(&cond->lock); + if (cond->n_to_wake && + cond->generation != generation_at_start) { + --cond->n_to_wake; + --cond->n_waiting; + result = 0; + waiting = 0; + goto out; + } else if (res != WAIT_OBJECT_0) { + result = (res==WAIT_TIMEOUT) ? 1 : -1; + --cond->n_waiting; + waiting = 0; + goto out; + } else if (ms != INFINITE) { + endTime = GetTickCount(); + if (startTime + ms_orig <= endTime) { + result = 1; /* Timeout */ + --cond->n_waiting; + waiting = 0; + goto out; + } else { + ms = startTime + ms_orig - endTime; + } + } + /* If we make it here, we are still waiting. */ + if (cond->n_to_wake == 0) { + /* There is nobody else who should wake up; reset + * the event. */ + ResetEvent(cond->event); + } + out: + LeaveCriticalSection(&cond->lock); + } while (waiting); + + EnterCriticalSection(lock); + + EnterCriticalSection(&cond->lock); + if (!cond->n_waiting) + ResetEvent(cond->event); + LeaveCriticalSection(&cond->lock); + + return result; +} + +int +evthread_use_windows_threads(void) +{ + struct evthread_lock_callbacks cbs = { + EVTHREAD_LOCK_API_VERSION, + EVTHREAD_LOCKTYPE_RECURSIVE, + evthread_win32_lock_create, + evthread_win32_lock_free, + evthread_win32_lock, + evthread_win32_unlock + }; + + + struct evthread_condition_callbacks cond_cbs = { + EVTHREAD_CONDITION_API_VERSION, + evthread_win32_cond_alloc, + evthread_win32_cond_free, + evthread_win32_cond_signal, + evthread_win32_cond_wait + }; +#ifdef WIN32_HAVE_CONDITION_VARIABLES + struct evthread_condition_callbacks condvar_cbs = { + EVTHREAD_CONDITION_API_VERSION, + evthread_win32_condvar_alloc, + evthread_win32_condvar_free, + evthread_win32_condvar_signal, + evthread_win32_condvar_wait + }; +#endif + + evthread_set_lock_callbacks(&cbs); + evthread_set_id_callback(evthread_win32_get_id); +#ifdef WIN32_HAVE_CONDITION_VARIABLES + if (evthread_win32_condvar_init()) { + evthread_set_condition_callbacks(&condvar_cbs); + return 0; + } +#endif + evthread_set_condition_callbacks(&cond_cbs); + + return 0; +} + diff --git a/contrib/libs/libevent/win32select.c b/contrib/libs/libevent/win32select.c new file mode 100644 index 00000000000..d005b587d45 --- /dev/null +++ b/contrib/libs/libevent/win32select.c @@ -0,0 +1,389 @@ +/* + * Copyright 2007-2012 Niels Provos and Nick Mathewson + * Copyright 2000-2007 Niels Provos <provos@citi.umich.edu> + * Copyright 2003 Michael A. Davis <mike@datanerds.net> + * + * 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" +#include "evconfig-private.h" + +#ifdef _WIN32 + +#include <winsock2.h> +#include <windows.h> +#include <sys/types.h> +#include <sys/queue.h> +#include <limits.h> +#include <signal.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <errno.h> + +#include "event2/util.h" +#include "util-internal.h" +#include "log-internal.h" +#include "event2/event.h" +#include "event-internal.h" +#include "evmap-internal.h" +#include "event2/thread.h" +#include "evthread-internal.h" +#include "time-internal.h" + +#define XFREE(ptr) do { if (ptr) mm_free(ptr); } while (0) + +extern struct event_list timequeue; +extern struct event_list addqueue; + +struct win_fd_set { + unsigned int fd_count; + SOCKET fd_array[1]; +}; + +/* MSDN says this is required to handle SIGFPE */ +volatile double SIGFPE_REQ = 0.0f; + +struct idx_info { + int read_pos_plus1; + int write_pos_plus1; +}; + +struct win32op { + unsigned num_fds_in_fd_sets; + int resize_out_sets; + struct win_fd_set *readset_in; + struct win_fd_set *writeset_in; + struct win_fd_set *readset_out; + struct win_fd_set *writeset_out; + struct win_fd_set *exset_out; + unsigned signals_are_broken : 1; +}; + +static void *win32_init(struct event_base *); +static int win32_add(struct event_base *, evutil_socket_t, short old, short events, void *idx_); +static int win32_del(struct event_base *, evutil_socket_t, short old, short events, void *idx_); +static int win32_dispatch(struct event_base *base, struct timeval *); +static void win32_dealloc(struct event_base *); + +struct eventop win32ops = { + "win32", + win32_init, + win32_add, + win32_del, + win32_dispatch, + win32_dealloc, + 0, /* doesn't need reinit */ + 0, /* No features supported. */ + sizeof(struct idx_info), +}; + +#define FD_SET_ALLOC_SIZE(n) ((sizeof(struct win_fd_set) + ((n)-1)*sizeof(SOCKET))) + +static int +grow_fd_sets(struct win32op *op, unsigned new_num_fds) +{ + size_t size; + + EVUTIL_ASSERT(new_num_fds >= op->readset_in->fd_count && + new_num_fds >= op->writeset_in->fd_count); + EVUTIL_ASSERT(new_num_fds >= 1); + + size = FD_SET_ALLOC_SIZE(new_num_fds); + if (!(op->readset_in = mm_realloc(op->readset_in, size))) + return (-1); + if (!(op->writeset_in = mm_realloc(op->writeset_in, size))) + return (-1); + op->resize_out_sets = 1; + op->num_fds_in_fd_sets = new_num_fds; + return (0); +} + +static int +do_fd_set(struct win32op *op, struct idx_info *ent, evutil_socket_t s, int read) +{ + struct win_fd_set *set = read ? op->readset_in : op->writeset_in; + if (read) { + if (ent->read_pos_plus1 > 0) + return (0); + } else { + if (ent->write_pos_plus1 > 0) + return (0); + } + if (set->fd_count == op->num_fds_in_fd_sets) { + if (grow_fd_sets(op, op->num_fds_in_fd_sets*2)) + return (-1); + /* set pointer will have changed and needs reiniting! */ + set = read ? op->readset_in : op->writeset_in; + } + set->fd_array[set->fd_count] = s; + if (read) + ent->read_pos_plus1 = set->fd_count+1; + else + ent->write_pos_plus1 = set->fd_count+1; + return (set->fd_count++); +} + +static int +do_fd_clear(struct event_base *base, + struct win32op *op, struct idx_info *ent, int read) +{ + int i; + struct win_fd_set *set = read ? op->readset_in : op->writeset_in; + if (read) { + i = ent->read_pos_plus1 - 1; + ent->read_pos_plus1 = 0; + } else { + i = ent->write_pos_plus1 - 1; + ent->write_pos_plus1 = 0; + } + if (i < 0) + return (0); + if (--set->fd_count != (unsigned)i) { + struct idx_info *ent2; + SOCKET s2; + s2 = set->fd_array[i] = set->fd_array[set->fd_count]; + + ent2 = evmap_io_get_fdinfo_(&base->io, s2); + + if (!ent2) /* This indicates a bug. */ + return (0); + if (read) + ent2->read_pos_plus1 = i+1; + else + ent2->write_pos_plus1 = i+1; + } + return (0); +} + +#define NEVENT 32 +void * +win32_init(struct event_base *base) +{ + struct win32op *winop; + size_t size; + if (!(winop = mm_calloc(1, sizeof(struct win32op)))) + return NULL; + winop->num_fds_in_fd_sets = NEVENT; + size = FD_SET_ALLOC_SIZE(NEVENT); + if (!(winop->readset_in = mm_malloc(size))) + goto err; + if (!(winop->writeset_in = mm_malloc(size))) + goto err; + if (!(winop->readset_out = mm_malloc(size))) + goto err; + if (!(winop->writeset_out = mm_malloc(size))) + goto err; + if (!(winop->exset_out = mm_malloc(size))) + goto err; + winop->readset_in->fd_count = winop->writeset_in->fd_count = 0; + winop->readset_out->fd_count = winop->writeset_out->fd_count + = winop->exset_out->fd_count = 0; + + if (evsig_init_(base) < 0) + winop->signals_are_broken = 1; + + evutil_weakrand_seed_(&base->weakrand_seed, 0); + + return (winop); + err: + XFREE(winop->readset_in); + XFREE(winop->writeset_in); + XFREE(winop->readset_out); + XFREE(winop->writeset_out); + XFREE(winop->exset_out); + XFREE(winop); + return (NULL); +} + +int +win32_add(struct event_base *base, evutil_socket_t fd, + short old, short events, void *idx_) +{ + struct win32op *win32op = base->evbase; + struct idx_info *idx = idx_; + + if ((events & EV_SIGNAL) && win32op->signals_are_broken) + return (-1); + + if (!(events & (EV_READ|EV_WRITE))) + return (0); + + event_debug(("%s: adding event for %d", __func__, (int)fd)); + if (events & EV_READ) { + if (do_fd_set(win32op, idx, fd, 1)<0) + return (-1); + } + if (events & EV_WRITE) { + if (do_fd_set(win32op, idx, fd, 0)<0) + return (-1); + } + return (0); +} + +int +win32_del(struct event_base *base, evutil_socket_t fd, short old, short events, + void *idx_) +{ + struct win32op *win32op = base->evbase; + struct idx_info *idx = idx_; + + event_debug(("%s: Removing event for "EV_SOCK_FMT, + __func__, EV_SOCK_ARG(fd))); + if (events & EV_READ) + do_fd_clear(base, win32op, idx, 1); + if (events & EV_WRITE) + do_fd_clear(base, win32op, idx, 0); + + return 0; +} + +static void +fd_set_copy(struct win_fd_set *out, const struct win_fd_set *in) +{ + out->fd_count = in->fd_count; + memcpy(out->fd_array, in->fd_array, in->fd_count * (sizeof(SOCKET))); +} + +/* + static void dump_fd_set(struct win_fd_set *s) + { + unsigned int i; + printf("[ "); + for(i=0;i<s->fd_count;++i) + printf("%d ",(int)s->fd_array[i]); + printf("]\n"); + } +*/ + +int +win32_dispatch(struct event_base *base, struct timeval *tv) +{ + struct win32op *win32op = base->evbase; + int res = 0; + unsigned j, i; + int fd_count; + SOCKET s; + + if (win32op->resize_out_sets) { + size_t size = FD_SET_ALLOC_SIZE(win32op->num_fds_in_fd_sets); + if (!(win32op->readset_out = mm_realloc(win32op->readset_out, size))) + return (-1); + if (!(win32op->exset_out = mm_realloc(win32op->exset_out, size))) + return (-1); + if (!(win32op->writeset_out = mm_realloc(win32op->writeset_out, size))) + return (-1); + win32op->resize_out_sets = 0; + } + + fd_set_copy(win32op->readset_out, win32op->readset_in); + fd_set_copy(win32op->exset_out, win32op->writeset_in); + fd_set_copy(win32op->writeset_out, win32op->writeset_in); + + fd_count = + (win32op->readset_out->fd_count > win32op->writeset_out->fd_count) ? + win32op->readset_out->fd_count : win32op->writeset_out->fd_count; + + if (!fd_count) { + long msec = tv ? evutil_tv_to_msec_(tv) : LONG_MAX; + /* Sleep's DWORD argument is unsigned long */ + if (msec < 0) + msec = LONG_MAX; + /* Windows doesn't like you to call select() with no sockets */ + Sleep(msec); + return (0); + } + + EVBASE_RELEASE_LOCK(base, th_base_lock); + + res = select(fd_count, + (struct fd_set*)win32op->readset_out, + (struct fd_set*)win32op->writeset_out, + (struct fd_set*)win32op->exset_out, tv); + + EVBASE_ACQUIRE_LOCK(base, th_base_lock); + + event_debug(("%s: select returned %d", __func__, res)); + + if (res <= 0) { + event_debug(("%s: %s", __func__, + evutil_socket_error_to_string(EVUTIL_SOCKET_ERROR()))); + return res; + } + + if (win32op->readset_out->fd_count) { + i = evutil_weakrand_range_(&base->weakrand_seed, + win32op->readset_out->fd_count); + for (j=0; j<win32op->readset_out->fd_count; ++j) { + if (++i >= win32op->readset_out->fd_count) + i = 0; + s = win32op->readset_out->fd_array[i]; + evmap_io_active_(base, s, EV_READ); + } + } + if (win32op->exset_out->fd_count) { + i = evutil_weakrand_range_(&base->weakrand_seed, + win32op->exset_out->fd_count); + for (j=0; j<win32op->exset_out->fd_count; ++j) { + if (++i >= win32op->exset_out->fd_count) + i = 0; + s = win32op->exset_out->fd_array[i]; + evmap_io_active_(base, s, EV_WRITE); + } + } + if (win32op->writeset_out->fd_count) { + i = evutil_weakrand_range_(&base->weakrand_seed, + win32op->writeset_out->fd_count); + for (j=0; j<win32op->writeset_out->fd_count; ++j) { + if (++i >= win32op->writeset_out->fd_count) + i = 0; + s = win32op->writeset_out->fd_array[i]; + evmap_io_active_(base, s, EV_WRITE); + } + } + return (0); +} + +void +win32_dealloc(struct event_base *base) +{ + struct win32op *win32op = base->evbase; + + evsig_dealloc_(base); + if (win32op->readset_in) + mm_free(win32op->readset_in); + if (win32op->writeset_in) + mm_free(win32op->writeset_in); + if (win32op->readset_out) + mm_free(win32op->readset_out); + if (win32op->writeset_out) + mm_free(win32op->writeset_out); + if (win32op->exset_out) + mm_free(win32op->exset_out); + /* XXXXX free the tree. */ + + memset(win32op, 0, sizeof(*win32op)); + mm_free(win32op); +} + +#endif |