Blame libevent/event_iocp.c

Packit e9ba0d
/*
Packit e9ba0d
 * Copyright (c) 2009-2012 Niels Provos, Nick Mathewson
Packit e9ba0d
 *
Packit e9ba0d
 * Redistribution and use in source and binary forms, with or without
Packit e9ba0d
 * modification, are permitted provided that the following conditions
Packit e9ba0d
 * are met:
Packit e9ba0d
 * 1. Redistributions of source code must retain the above copyright
Packit e9ba0d
 *    notice, this list of conditions and the following disclaimer.
Packit e9ba0d
 * 2. Redistributions in binary form must reproduce the above copyright
Packit e9ba0d
 *    notice, this list of conditions and the following disclaimer in the
Packit e9ba0d
 *    documentation and/or other materials provided with the distribution.
Packit e9ba0d
 * 3. The name of the author may not be used to endorse or promote products
Packit e9ba0d
 *    derived from this software without specific prior written permission.
Packit e9ba0d
 *
Packit e9ba0d
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
Packit e9ba0d
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
Packit e9ba0d
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
Packit e9ba0d
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
Packit e9ba0d
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
Packit e9ba0d
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
Packit e9ba0d
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
Packit e9ba0d
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
Packit e9ba0d
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
Packit e9ba0d
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
Packit e9ba0d
 */
Packit e9ba0d
Packit e9ba0d
#ifndef _WIN32_WINNT
Packit e9ba0d
/* Minimum required for InitializeCriticalSectionAndSpinCount */
Packit e9ba0d
#define _WIN32_WINNT 0x0403
Packit e9ba0d
#endif
Packit e9ba0d
#include <winsock2.h>
Packit e9ba0d
#include <windows.h>
Packit e9ba0d
#include <process.h>
Packit e9ba0d
#include <stdio.h>
Packit e9ba0d
#include <mswsock.h>
Packit e9ba0d
Packit e9ba0d
#include "event2/util.h"
Packit e9ba0d
#include "util-internal.h"
Packit e9ba0d
#include "iocp-internal.h"
Packit e9ba0d
#include "log-internal.h"
Packit e9ba0d
#include "mm-internal.h"
Packit e9ba0d
#include "event-internal.h"
Packit e9ba0d
#include "evthread-internal.h"
Packit e9ba0d
Packit e9ba0d
#define NOTIFICATION_KEY ((ULONG_PTR)-1)
Packit e9ba0d
Packit e9ba0d
void
Packit e9ba0d
event_overlapped_init(struct event_overlapped *o, iocp_callback cb)
Packit e9ba0d
{
Packit e9ba0d
	memset(o, 0, sizeof(struct event_overlapped));
Packit e9ba0d
	o->cb = cb;
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
static void
Packit e9ba0d
handle_entry(OVERLAPPED *o, ULONG_PTR completion_key, DWORD nBytes, int ok)
Packit e9ba0d
{
Packit e9ba0d
	struct event_overlapped *eo =
Packit e9ba0d
	    EVUTIL_UPCAST(o, struct event_overlapped, overlapped);
Packit e9ba0d
	eo->cb(eo, completion_key, nBytes, ok);
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
static void
Packit e9ba0d
loop(void *_port)
Packit e9ba0d
{
Packit e9ba0d
	struct event_iocp_port *port = _port;
Packit e9ba0d
	long ms = port->ms;
Packit e9ba0d
	HANDLE p = port->port;
Packit e9ba0d
Packit e9ba0d
	if (ms <= 0)
Packit e9ba0d
		ms = INFINITE;
Packit e9ba0d
Packit e9ba0d
	while (1) {
Packit e9ba0d
		OVERLAPPED *overlapped=NULL;
Packit e9ba0d
		ULONG_PTR key=0;
Packit e9ba0d
		DWORD bytes=0;
Packit e9ba0d
		int ok = GetQueuedCompletionStatus(p, &bytes, &key,
Packit e9ba0d
			&overlapped, ms);
Packit e9ba0d
		EnterCriticalSection(&port->lock);
Packit e9ba0d
		if (port->shutdown) {
Packit e9ba0d
			if (--port->n_live_threads == 0)
Packit e9ba0d
				ReleaseSemaphore(port->shutdownSemaphore, 1,
Packit e9ba0d
						NULL);
Packit e9ba0d
			LeaveCriticalSection(&port->lock);
Packit e9ba0d
			return;
Packit e9ba0d
		}
Packit e9ba0d
		LeaveCriticalSection(&port->lock);
Packit e9ba0d
Packit e9ba0d
		if (key != NOTIFICATION_KEY && overlapped)
Packit e9ba0d
			handle_entry(overlapped, key, bytes, ok);
Packit e9ba0d
		else if (!overlapped)
Packit e9ba0d
			break;
Packit e9ba0d
	}
Packit e9ba0d
	event_warnx("GetQueuedCompletionStatus exited with no event.");
Packit e9ba0d
	EnterCriticalSection(&port->lock);
Packit e9ba0d
	if (--port->n_live_threads == 0)
Packit e9ba0d
		ReleaseSemaphore(port->shutdownSemaphore, 1, NULL);
Packit e9ba0d
	LeaveCriticalSection(&port->lock);
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
int
Packit e9ba0d
event_iocp_port_associate(struct event_iocp_port *port, evutil_socket_t fd,
Packit e9ba0d
    ev_uintptr_t key)
Packit e9ba0d
{
Packit e9ba0d
	HANDLE h;
Packit e9ba0d
	h = CreateIoCompletionPort((HANDLE)fd, port->port, key, port->n_threads);
Packit e9ba0d
	if (!h)
Packit e9ba0d
		return -1;
Packit e9ba0d
	return 0;
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
static void *
Packit e9ba0d
get_extension_function(SOCKET s, const GUID *which_fn)
Packit e9ba0d
{
Packit e9ba0d
	void *ptr = NULL;
Packit e9ba0d
	DWORD bytes=0;
Packit e9ba0d
	WSAIoctl(s, SIO_GET_EXTENSION_FUNCTION_POINTER,
Packit e9ba0d
	    (GUID*)which_fn, sizeof(*which_fn),
Packit e9ba0d
	    &ptr, sizeof(ptr),
Packit e9ba0d
	    &bytes, NULL, NULL);
Packit e9ba0d
Packit e9ba0d
	/* No need to detect errors here: if ptr is set, then we have a good
Packit e9ba0d
	   function pointer.  Otherwise, we should behave as if we had no
Packit e9ba0d
	   function pointer.
Packit e9ba0d
	*/
Packit e9ba0d
	return ptr;
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
/* Mingw doesn't have these in its mswsock.h.  The values are copied from
Packit e9ba0d
   wine.h.   Perhaps if we copy them exactly, the cargo will come again.
Packit e9ba0d
*/
Packit e9ba0d
#ifndef WSAID_ACCEPTEX
Packit e9ba0d
#define WSAID_ACCEPTEX \
Packit e9ba0d
	{0xb5367df1,0xcbac,0x11cf,{0x95,0xca,0x00,0x80,0x5f,0x48,0xa1,0x92}}
Packit e9ba0d
#endif
Packit e9ba0d
#ifndef WSAID_CONNECTEX
Packit e9ba0d
#define WSAID_CONNECTEX \
Packit e9ba0d
	{0x25a207b9,0xddf3,0x4660,{0x8e,0xe9,0x76,0xe5,0x8c,0x74,0x06,0x3e}}
Packit e9ba0d
#endif
Packit e9ba0d
#ifndef WSAID_GETACCEPTEXSOCKADDRS
Packit e9ba0d
#define WSAID_GETACCEPTEXSOCKADDRS \
Packit e9ba0d
	{0xb5367df2,0xcbac,0x11cf,{0x95,0xca,0x00,0x80,0x5f,0x48,0xa1,0x92}}
Packit e9ba0d
#endif
Packit e9ba0d
Packit e9ba0d
static int extension_fns_initialized = 0;
Packit e9ba0d
Packit e9ba0d
static void
Packit e9ba0d
init_extension_functions(struct win32_extension_fns *ext)
Packit e9ba0d
{
Packit e9ba0d
	const GUID acceptex = WSAID_ACCEPTEX;
Packit e9ba0d
	const GUID connectex = WSAID_CONNECTEX;
Packit e9ba0d
	const GUID getacceptexsockaddrs = WSAID_GETACCEPTEXSOCKADDRS;
Packit e9ba0d
	SOCKET s = socket(AF_INET, SOCK_STREAM, 0);
Packit e9ba0d
	if (s == INVALID_SOCKET)
Packit e9ba0d
		return;
Packit e9ba0d
	ext->AcceptEx = get_extension_function(s, &acceptex);
Packit e9ba0d
	ext->ConnectEx = get_extension_function(s, &connectex);
Packit e9ba0d
	ext->GetAcceptExSockaddrs = get_extension_function(s,
Packit e9ba0d
	    &getacceptexsockaddrs);
Packit e9ba0d
	closesocket(s);
Packit e9ba0d
Packit e9ba0d
	extension_fns_initialized = 1;
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
static struct win32_extension_fns the_extension_fns;
Packit e9ba0d
Packit e9ba0d
const struct win32_extension_fns *
Packit e9ba0d
event_get_win32_extension_fns(void)
Packit e9ba0d
{
Packit e9ba0d
	return &the_extension_fns;
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
#define N_CPUS_DEFAULT 2
Packit e9ba0d
Packit e9ba0d
struct event_iocp_port *
Packit e9ba0d
event_iocp_port_launch(int n_cpus)
Packit e9ba0d
{
Packit e9ba0d
	struct event_iocp_port *port;
Packit e9ba0d
	int i;
Packit e9ba0d
Packit e9ba0d
	if (!extension_fns_initialized)
Packit e9ba0d
		init_extension_functions(&the_extension_fns);
Packit e9ba0d
Packit e9ba0d
	if (!(port = mm_calloc(1, sizeof(struct event_iocp_port))))
Packit e9ba0d
		return NULL;
Packit e9ba0d
Packit e9ba0d
	if (n_cpus <= 0)
Packit e9ba0d
		n_cpus = N_CPUS_DEFAULT;
Packit e9ba0d
	port->n_threads = n_cpus * 2;
Packit e9ba0d
	port->threads = mm_calloc(port->n_threads, sizeof(HANDLE));
Packit e9ba0d
	if (!port->threads)
Packit e9ba0d
		goto err;
Packit e9ba0d
Packit e9ba0d
	port->port = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0,
Packit e9ba0d
			n_cpus);
Packit e9ba0d
	port->ms = -1;
Packit e9ba0d
	if (!port->port)
Packit e9ba0d
		goto err;
Packit e9ba0d
Packit e9ba0d
	port->shutdownSemaphore = CreateSemaphore(NULL, 0, 1, NULL);
Packit e9ba0d
	if (!port->shutdownSemaphore)
Packit e9ba0d
		goto err;
Packit e9ba0d
Packit e9ba0d
	for (i=0; i<port->n_threads; ++i) {
Packit e9ba0d
		ev_uintptr_t th = _beginthread(loop, 0, port);
Packit e9ba0d
		if (th == (ev_uintptr_t)-1)
Packit e9ba0d
			goto err;
Packit e9ba0d
		port->threads[i] = (HANDLE)th;
Packit e9ba0d
		++port->n_live_threads;
Packit e9ba0d
	}
Packit e9ba0d
Packit e9ba0d
	InitializeCriticalSectionAndSpinCount(&port->lock, 1000);
Packit e9ba0d
Packit e9ba0d
	return port;
Packit e9ba0d
err:
Packit e9ba0d
	if (port->port)
Packit e9ba0d
		CloseHandle(port->port);
Packit e9ba0d
	if (port->threads)
Packit e9ba0d
		mm_free(port->threads);
Packit e9ba0d
	if (port->shutdownSemaphore)
Packit e9ba0d
		CloseHandle(port->shutdownSemaphore);
Packit e9ba0d
	mm_free(port);
Packit e9ba0d
	return NULL;
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
static void
Packit e9ba0d
_event_iocp_port_unlock_and_free(struct event_iocp_port *port)
Packit e9ba0d
{
Packit e9ba0d
	DeleteCriticalSection(&port->lock);
Packit e9ba0d
	CloseHandle(port->port);
Packit e9ba0d
	CloseHandle(port->shutdownSemaphore);
Packit e9ba0d
	mm_free(port->threads);
Packit e9ba0d
	mm_free(port);
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
static int
Packit e9ba0d
event_iocp_notify_all(struct event_iocp_port *port)
Packit e9ba0d
{
Packit e9ba0d
	int i, r, ok=1;
Packit e9ba0d
	for (i=0; i<port->n_threads; ++i) {
Packit e9ba0d
		r = PostQueuedCompletionStatus(port->port, 0, NOTIFICATION_KEY,
Packit e9ba0d
		    NULL);
Packit e9ba0d
		if (!r)
Packit e9ba0d
			ok = 0;
Packit e9ba0d
	}
Packit e9ba0d
	return ok ? 0 : -1;
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
int
Packit e9ba0d
event_iocp_shutdown(struct event_iocp_port *port, long waitMsec)
Packit e9ba0d
{
Packit e9ba0d
	DWORD ms = INFINITE;
Packit e9ba0d
	int n;
Packit e9ba0d
Packit e9ba0d
	EnterCriticalSection(&port->lock);
Packit e9ba0d
	port->shutdown = 1;
Packit e9ba0d
	LeaveCriticalSection(&port->lock);
Packit e9ba0d
	event_iocp_notify_all(port);
Packit e9ba0d
Packit e9ba0d
	if (waitMsec >= 0)
Packit e9ba0d
		ms = waitMsec;
Packit e9ba0d
Packit e9ba0d
	WaitForSingleObject(port->shutdownSemaphore, ms);
Packit e9ba0d
	EnterCriticalSection(&port->lock);
Packit e9ba0d
	n = port->n_live_threads;
Packit e9ba0d
	LeaveCriticalSection(&port->lock);
Packit e9ba0d
	if (n == 0) {
Packit e9ba0d
		_event_iocp_port_unlock_and_free(port);
Packit e9ba0d
		return 0;
Packit e9ba0d
	} else {
Packit e9ba0d
		return -1;
Packit e9ba0d
	}
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
int
Packit e9ba0d
event_iocp_activate_overlapped(
Packit e9ba0d
    struct event_iocp_port *port, struct event_overlapped *o,
Packit e9ba0d
    ev_uintptr_t key, ev_uint32_t n)
Packit e9ba0d
{
Packit e9ba0d
	BOOL r;
Packit e9ba0d
Packit e9ba0d
	r = PostQueuedCompletionStatus(port->port, n, key, &o->overlapped);
Packit e9ba0d
	return (r==0) ? -1 : 0;
Packit e9ba0d
}
Packit e9ba0d
Packit e9ba0d
struct event_iocp_port *
Packit e9ba0d
event_base_get_iocp(struct event_base *base)
Packit e9ba0d
{
Packit e9ba0d
#ifdef WIN32
Packit e9ba0d
	return base->iocp;
Packit e9ba0d
#else
Packit e9ba0d
	return NULL;
Packit e9ba0d
#endif
Packit e9ba0d
}