1 /*
2 * Copyright (c) 2009-2012 Niels Provos and 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 /**
28 @file buffer_iocp.c
29
30 This module implements overlapped read and write functions for evbuffer
31 objects on Windows.
32 */
33 #include "event2/event-config.h"
34 #include "evconfig-private.h"
35
36 #include "event2/buffer.h"
37 #include "event2/buffer_compat.h"
38 #include "event2/util.h"
39 #include "event2/thread.h"
40 #include "util-internal.h"
41 #include "evthread-internal.h"
42 #include "evbuffer-internal.h"
43 #include "iocp-internal.h"
44 #include "mm-internal.h"
45
46 #include <winsock2.h>
47 #include <windows.h>
48 #include <stdio.h>
49
50 #define MAX_WSABUFS 16
51
52 /** An evbuffer that can handle overlapped IO. */
53 struct evbuffer_overlapped {
54 struct evbuffer buffer;
55 /** The socket that we're doing overlapped IO on. */
56 evutil_socket_t fd;
57
58 /** pending I/O type */
59 unsigned read_in_progress : 1;
60 unsigned write_in_progress : 1;
61
62 /** The first pinned chain in the buffer. */
63 struct evbuffer_chain *first_pinned;
64
65 /** How many chains are pinned; how many of the fields in buffers
66 * are we using. */
67 int n_buffers;
68 WSABUF buffers[MAX_WSABUFS];
69 };
70
71 /** Given an evbuffer, return the correponding evbuffer structure, or NULL if
72 * the evbuffer isn't overlapped. */
73 static inline struct evbuffer_overlapped *
upcast_evbuffer(struct evbuffer * buf)74 upcast_evbuffer(struct evbuffer *buf)
75 {
76 if (!buf || !buf->is_overlapped)
77 return NULL;
78 return EVUTIL_UPCAST(buf, struct evbuffer_overlapped, buffer);
79 }
80
81 /** Unpin all the chains noted as pinned in 'eo'. */
82 static void
pin_release(struct evbuffer_overlapped * eo,unsigned flag)83 pin_release(struct evbuffer_overlapped *eo, unsigned flag)
84 {
85 int i;
86 struct evbuffer_chain *next, *chain = eo->first_pinned;
87
88 for (i = 0; i < eo->n_buffers; ++i) {
89 EVUTIL_ASSERT(chain);
90 next = chain->next;
91 evbuffer_chain_unpin_(chain, flag);
92 chain = next;
93 }
94 }
95
96 void
evbuffer_commit_read_(struct evbuffer * evbuf,ev_ssize_t nBytes)97 evbuffer_commit_read_(struct evbuffer *evbuf, ev_ssize_t nBytes)
98 {
99 struct evbuffer_overlapped *buf = upcast_evbuffer(evbuf);
100 struct evbuffer_chain **chainp;
101 size_t remaining, len;
102 unsigned i;
103
104 EVBUFFER_LOCK(evbuf);
105 EVUTIL_ASSERT(buf->read_in_progress && !buf->write_in_progress);
106 EVUTIL_ASSERT(nBytes >= 0); /* XXXX Can this be false? */
107
108 evbuffer_unfreeze(evbuf, 0);
109
110 chainp = evbuf->last_with_datap;
111 if (!((*chainp)->flags & EVBUFFER_MEM_PINNED_R))
112 chainp = &(*chainp)->next;
113 remaining = nBytes;
114 for (i = 0; remaining > 0 && i < (unsigned)buf->n_buffers; ++i) {
115 EVUTIL_ASSERT(*chainp);
116 len = buf->buffers[i].len;
117 if (remaining < len)
118 len = remaining;
119 (*chainp)->off += len;
120 evbuf->last_with_datap = chainp;
121 remaining -= len;
122 chainp = &(*chainp)->next;
123 }
124
125 pin_release(buf, EVBUFFER_MEM_PINNED_R);
126
127 buf->read_in_progress = 0;
128
129 evbuf->total_len += nBytes;
130 evbuf->n_add_for_cb += nBytes;
131
132 evbuffer_invoke_callbacks_(evbuf);
133
134 evbuffer_decref_and_unlock_(evbuf);
135 }
136
137 void
evbuffer_commit_write_(struct evbuffer * evbuf,ev_ssize_t nBytes)138 evbuffer_commit_write_(struct evbuffer *evbuf, ev_ssize_t nBytes)
139 {
140 struct evbuffer_overlapped *buf = upcast_evbuffer(evbuf);
141
142 EVBUFFER_LOCK(evbuf);
143 EVUTIL_ASSERT(buf->write_in_progress && !buf->read_in_progress);
144 evbuffer_unfreeze(evbuf, 1);
145 evbuffer_drain(evbuf, nBytes);
146 pin_release(buf,EVBUFFER_MEM_PINNED_W);
147 buf->write_in_progress = 0;
148 evbuffer_decref_and_unlock_(evbuf);
149 }
150
151 struct evbuffer *
evbuffer_overlapped_new_(evutil_socket_t fd)152 evbuffer_overlapped_new_(evutil_socket_t fd)
153 {
154 struct evbuffer_overlapped *evo;
155
156 evo = mm_calloc(1, sizeof(struct evbuffer_overlapped));
157 if (!evo)
158 return NULL;
159
160 LIST_INIT(&evo->buffer.callbacks);
161 evo->buffer.refcnt = 1;
162 evo->buffer.last_with_datap = &evo->buffer.first;
163
164 evo->buffer.is_overlapped = 1;
165 evo->fd = fd;
166
167 return &evo->buffer;
168 }
169
170 int
evbuffer_launch_write_(struct evbuffer * buf,ev_ssize_t at_most,struct event_overlapped * ol)171 evbuffer_launch_write_(struct evbuffer *buf, ev_ssize_t at_most,
172 struct event_overlapped *ol)
173 {
174 struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf);
175 int r = -1;
176 int i;
177 struct evbuffer_chain *chain;
178 DWORD bytesSent;
179
180 if (!buf) {
181 /* No buffer, or it isn't overlapped */
182 return -1;
183 }
184
185 EVBUFFER_LOCK(buf);
186 EVUTIL_ASSERT(!buf_o->read_in_progress);
187 if (buf->freeze_start || buf_o->write_in_progress)
188 goto done;
189 if (!buf->total_len) {
190 /* Nothing to write */
191 r = 0;
192 goto done;
193 } else if (at_most < 0 || (size_t)at_most > buf->total_len) {
194 at_most = buf->total_len;
195 }
196 evbuffer_freeze(buf, 1);
197
198 buf_o->first_pinned = NULL;
199 buf_o->n_buffers = 0;
200 memset(buf_o->buffers, 0, sizeof(buf_o->buffers));
201
202 chain = buf_o->first_pinned = buf->first;
203
204 for (i=0; i < MAX_WSABUFS && chain; ++i, chain=chain->next) {
205 WSABUF *b = &buf_o->buffers[i];
206 b->buf = (char*)( chain->buffer + chain->misalign );
207 evbuffer_chain_pin_(chain, EVBUFFER_MEM_PINNED_W);
208
209 if ((size_t)at_most > chain->off) {
210 /* XXXX Cast is safe for now, since win32 has no
211 mmaped chains. But later, we need to have this
212 add more WSAbufs if chain->off is greater than
213 ULONG_MAX */
214 b->len = (unsigned long)chain->off;
215 at_most -= chain->off;
216 } else {
217 b->len = (unsigned long)at_most;
218 ++i;
219 break;
220 }
221 }
222
223 buf_o->n_buffers = i;
224 evbuffer_incref_(buf);
225 if (WSASend(buf_o->fd, buf_o->buffers, i, &bytesSent, 0,
226 &ol->overlapped, NULL)) {
227 int error = WSAGetLastError();
228 if (error != WSA_IO_PENDING) {
229 /* An actual error. */
230 pin_release(buf_o, EVBUFFER_MEM_PINNED_W);
231 evbuffer_unfreeze(buf, 1);
232 evbuffer_free(buf); /* decref */
233 goto done;
234 }
235 }
236
237 buf_o->write_in_progress = 1;
238 r = 0;
239 done:
240 EVBUFFER_UNLOCK(buf);
241 return r;
242 }
243
244 int
evbuffer_launch_read_(struct evbuffer * buf,size_t at_most,struct event_overlapped * ol)245 evbuffer_launch_read_(struct evbuffer *buf, size_t at_most,
246 struct event_overlapped *ol)
247 {
248 struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf);
249 int r = -1, i;
250 int nvecs;
251 int npin=0;
252 struct evbuffer_chain *chain=NULL, **chainp;
253 DWORD bytesRead;
254 DWORD flags = 0;
255 struct evbuffer_iovec vecs[MAX_WSABUFS];
256
257 if (!buf_o)
258 return -1;
259 EVBUFFER_LOCK(buf);
260 EVUTIL_ASSERT(!buf_o->write_in_progress);
261 if (buf->freeze_end || buf_o->read_in_progress)
262 goto done;
263
264 buf_o->first_pinned = NULL;
265 buf_o->n_buffers = 0;
266 memset(buf_o->buffers, 0, sizeof(buf_o->buffers));
267
268 if (evbuffer_expand_fast_(buf, at_most, MAX_WSABUFS) == -1)
269 goto done;
270 evbuffer_freeze(buf, 0);
271
272 nvecs = evbuffer_read_setup_vecs_(buf, at_most,
273 vecs, MAX_WSABUFS, &chainp, 1);
274 for (i=0;i<nvecs;++i) {
275 WSABUF_FROM_EVBUFFER_IOV(
276 &buf_o->buffers[i],
277 &vecs[i]);
278 }
279
280 buf_o->n_buffers = nvecs;
281 buf_o->first_pinned = chain = *chainp;
282
283 npin=0;
284 for ( ; chain; chain = chain->next) {
285 evbuffer_chain_pin_(chain, EVBUFFER_MEM_PINNED_R);
286 ++npin;
287 }
288 EVUTIL_ASSERT(npin == nvecs);
289
290 evbuffer_incref_(buf);
291 if (WSARecv(buf_o->fd, buf_o->buffers, nvecs, &bytesRead, &flags,
292 &ol->overlapped, NULL)) {
293 int error = WSAGetLastError();
294 if (error != WSA_IO_PENDING) {
295 /* An actual error. */
296 pin_release(buf_o, EVBUFFER_MEM_PINNED_R);
297 evbuffer_unfreeze(buf, 0);
298 evbuffer_free(buf); /* decref */
299 goto done;
300 }
301 }
302
303 buf_o->read_in_progress = 1;
304 r = 0;
305 done:
306 EVBUFFER_UNLOCK(buf);
307 return r;
308 }
309
310 evutil_socket_t
evbuffer_overlapped_get_fd_(struct evbuffer * buf)311 evbuffer_overlapped_get_fd_(struct evbuffer *buf)
312 {
313 struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf);
314 return buf_o ? buf_o->fd : -1;
315 }
316
317 void
evbuffer_overlapped_set_fd_(struct evbuffer * buf,evutil_socket_t fd)318 evbuffer_overlapped_set_fd_(struct evbuffer *buf, evutil_socket_t fd)
319 {
320 struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf);
321 EVBUFFER_LOCK(buf);
322 /* XXX is this right?, should it cancel current I/O operations? */
323 if (buf_o)
324 buf_o->fd = fd;
325 EVBUFFER_UNLOCK(buf);
326 }
327