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