libevent/event_iocp.c

   1 /*
   2  * Copyright (c) 2009-2010 Niels Provos, Nick Mathewson
   3  *
   4  * Redistribution and use in source and binary forms, with or without
   5  * modification, are permitted provided that the following conditions
   6  * are met:
   7  * 1. Redistributions of source code must retain the above copyright
   8  *    notice, this list of conditions and the following disclaimer.
   9  * 2. Redistributions in binary form must reproduce the above copyright
  10  *    notice, this list of conditions and the following disclaimer in the
  11  *    documentation and/or other materials provided with the distribution.
  12  * 3. The name of the author may not be used to endorse or promote products
  13  *    derived from this software without specific prior written permission.
  14  *
  15  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
  16  * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
  17  * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
  18  * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
  19  * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
  20  * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  21  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  22  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  23  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
  24  * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  25  */
  26
  27 #include <winsock2.h>
  28 #include <windows.h>
  29 #include <process.h>
  30 #include <stdio.h>
  31 #include <mswsock.h>
  32
  33 #include "event2/util.h"
  34 #include "util-internal.h"
  35 #include "iocp-internal.h"
  36 #include "log-internal.h"
  37 #include "mm-internal.h"
  38 #include "event-internal.h"
  39 #include "evthread-internal.h"
  40
  41 #define NOTIFICATION_KEY ((ULONG_PTR)-1)
  42
  43 void
  44 event_overlapped_init(struct event_overlapped *o, iocp_callback cb)
  45 {
  46         memset(o, 0, sizeof(struct event_overlapped));
  47         o->cb = cb;
  48 }
  49
  50 static void
  51 handle_entry(OVERLAPPED *o, ULONG_PTR completion_key, DWORD nBytes, int ok)
  52 {
  53         struct event_overlapped *eo =
  54             EVUTIL_UPCAST(o, struct event_overlapped, overlapped);
  55         eo->cb(eo, completion_key, nBytes, ok);
  56 }
  57
  58 static void
  59 loop(void *_port)
  60 {
  61         struct event_iocp_port *port = _port;
  62         long ms = port->ms;
  63         HANDLE p = port->port;
  64
  65         if (ms <= 0)
  66                 ms = INFINITE;
  67
  68         while (1) {
  69                 OVERLAPPED *overlapped=NULL;
  70                 ULONG_PTR key=0;
  71                 DWORD bytes=0;
  72                 int ok = GetQueuedCompletionStatus(p, &bytes, &key,
  73                         &overlapped, ms);
  74                 EnterCriticalSection(&port->lock);
  75                 if (port->shutdown) {
  76                         if (--port->n_live_threads == 0)
  77                                 ReleaseSemaphore(port->shutdownSemaphore, 1,
  78                                                 NULL);
  79                         LeaveCriticalSection(&port->lock);
  80                         return;
  81                 }
  82                 LeaveCriticalSection(&port->lock);
  83
  84                 if (key != NOTIFICATION_KEY && overlapped)
  85                         handle_entry(overlapped, key, bytes, ok);
  86                 else if (!overlapped)
  87                         break;
  88         }
  89         event_warnx("GetQueuedCompletionStatus exited with no event.");
  90         EnterCriticalSection(&port->lock);
  91         if (--port->n_live_threads == 0)
  92                 ReleaseSemaphore(port->shutdownSemaphore, 1, NULL);
  93         LeaveCriticalSection(&port->lock);
  94 }
  95
  96 int
  97 event_iocp_port_associate(struct event_iocp_port *port, evutil_socket_t fd,
  98     ev_uintptr_t key)
  99 {
 100         HANDLE h;
 101         h = CreateIoCompletionPort((HANDLE)fd, port->port, key, port->n_threads);
 102         if (!h)
 103                 return -1;
 104         return 0;
 105 }
 106
 107 static void *
 108 get_extension_function(SOCKET s, const GUID *which_fn)
 109 {
 110         void *ptr = NULL;
 111         DWORD bytes=0;
 112         WSAIoctl(s, SIO_GET_EXTENSION_FUNCTION_POINTER,
 113             (GUID*)which_fn, sizeof(*which_fn),
 114             &ptr, sizeof(ptr),
 115             &bytes, NULL, NULL);
 116
 117         /* No need to detect errors here: if ptr is set, then we have a good
 118            function pointer.  Otherwise, we should behave as if we had no
 119            function pointer.
 120         */
 121         return ptr;
 122 }
 123
 124 /* Mingw doesn't have these in its mswsock.h.  The values are copied from
 125    wine.h.   Perhaps if we copy them exactly, the cargo will come again.
 126 */
 127 #ifndef WSAID_ACCEPTEX
 128 #define WSAID_ACCEPTEX \
 129         {0xb5367df1,0xcbac,0x11cf,{0x95,0xca,0x00,0x80,0x5f,0x48,0xa1,0x92}}
 130 #endif
 131 #ifndef WSAID_CONNECTEX
 132 #define WSAID_CONNECTEX \
 133         {0x25a207b9,0xddf3,0x4660,{0x8e,0xe9,0x76,0xe5,0x8c,0x74,0x06,0x3e}}
 134 #endif
 135 #ifndef WSAID_GETACCEPTEXSOCKADDRS
 136 #define WSAID_GETACCEPTEXSOCKADDRS \
 137         {0xb5367df2,0xcbac,0x11cf,{0x95,0xca,0x00,0x80,0x5f,0x48,0xa1,0x92}}
 138 #endif
 139
 140 static void
 141 init_extension_functions(struct win32_extension_fns *ext)
 142 {
 143         const GUID acceptex = WSAID_ACCEPTEX;
 144         const GUID connectex = WSAID_CONNECTEX;
 145         const GUID getacceptexsockaddrs = WSAID_GETACCEPTEXSOCKADDRS;
 146         SOCKET s = socket(AF_INET, SOCK_STREAM, 0);
 147         if (s == INVALID_SOCKET)
 148                 return;
 149         ext->AcceptEx = get_extension_function(s, &acceptex);
 150         ext->ConnectEx = get_extension_function(s, &connectex);
 151         ext->GetAcceptExSockaddrs = get_extension_function(s,
 152             &getacceptexsockaddrs);
 153         closesocket(s);
 154 }
 155
 156 static struct win32_extension_fns the_extension_fns;
 157 static int extension_fns_initialized = 0;
 158
 159 const struct win32_extension_fns *
 160 event_get_win32_extension_fns(void)
 161 {
 162         return &the_extension_fns;
 163 }
 164
 165 #define N_CPUS_DEFAULT 2
 166
 167 struct event_iocp_port *
 168 event_iocp_port_launch(int n_cpus)
 169 {
 170         struct event_iocp_port *port;
 171         int i;
 172
 173         if (!extension_fns_initialized)
 174                 init_extension_functions(&the_extension_fns);
 175
 176         if (!(port = mm_calloc(1, sizeof(struct event_iocp_port))))
 177                 return NULL;
 178
 179         if (n_cpus <= 0)
 180                 n_cpus = N_CPUS_DEFAULT;
 181         port->n_threads = n_cpus * 2;
 182         port->threads = calloc(port->n_threads, sizeof(HANDLE));
 183         if (!port->threads)
 184                 goto err;
 185
 186         port->port = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0,
 187                         n_cpus);
 188         port->ms = -1;
 189         if (!port->port)
 190                 goto err;
 191
 192         port->shutdownSemaphore = CreateSemaphore(NULL, 0, 1, NULL);
 193         if (!port->shutdownSemaphore)
 194                 goto err;
 195
 196         for (i=0; i<port->n_threads; ++i) {
 197                 ev_uintptr_t th = _beginthread(loop, 0, port);
 198                 if (th == (ev_uintptr_t)-1)
 199                         goto err;
 200                 port->threads[i] = (HANDLE)th;
 201                 ++port->n_live_threads;
 202         }
 203
 204         InitializeCriticalSectionAndSpinCount(&port->lock, 1000);
 205
 206         return port;
 207 err:
 208         if (port->port)
 209                 CloseHandle(port->port);
 210         if (port->threads)
 211                 mm_free(port->threads);
 212         if (port->shutdownSemaphore)
 213                 CloseHandle(port->shutdownSemaphore);
 214         mm_free(port);
 215         return NULL;
 216 }
 217
 218 static void
 219 _event_iocp_port_unlock_and_free(struct event_iocp_port *port)
 220 {
 221         DeleteCriticalSection(&port->lock);
 222         CloseHandle(port->port);
 223         CloseHandle(port->shutdownSemaphore);
 224         mm_free(port->threads);
 225         mm_free(port);
 226 }
 227
 228 static int
 229 event_iocp_notify_all(struct event_iocp_port *port)
 230 {
 231         int i, r, ok=1;
 232         for (i=0; i<port->n_threads; ++i) {
 233                 r = PostQueuedCompletionStatus(port->port, 0, NOTIFICATION_KEY,
 234                     NULL);
 235                 if (!r)
 236                         ok = 0;
 237         }
 238         return ok ? 0 : -1;
 239 }
 240
 241 int
 242 event_iocp_shutdown(struct event_iocp_port *port, long waitMsec)
 243 {
 244         DWORD ms = INFINITE;
 245         int n;
 246
 247         EnterCriticalSection(&port->lock);
 248         port->shutdown = 1;
 249         LeaveCriticalSection(&port->lock);
 250         event_iocp_notify_all(port);
 251
 252         if (waitMsec >= 0)
 253                 ms = waitMsec;
 254
 255         WaitForSingleObject(port->shutdownSemaphore, ms);
 256         EnterCriticalSection(&port->lock);
 257         n = port->n_live_threads;
 258         LeaveCriticalSection(&port->lock);
 259         if (n == 0) {
 260                 _event_iocp_port_unlock_and_free(port);
 261                 return 0;
 262         } else {
 263                 return -1;
 264         }
 265 }
 266
 267 int
 268 event_iocp_activate_overlapped(
 269     struct event_iocp_port *port, struct event_overlapped *o,
 270     ev_uintptr_t key, ev_uint32_t n)
 271 {
 272         BOOL r;
 273
 274         r = PostQueuedCompletionStatus(port->port, n, key, &o->overlapped);
 275         return (r==0) ? -1 : 0;
 276 }
 277
 278 struct event_iocp_port *
 279 event_base_get_iocp(struct event_base *base)
 280 {
 281 #ifdef WIN32
 282         return base->iocp;
 283 #else
 284         return NULL;
 285 #endif
 286 }