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