1 /* $NetBSD: buffer_iocp.c,v 1.1.1.1 2013/04/11 16:43:25 christos Exp $ */ 2 /* 3 * Copyright (c) 2009-2012 Niels Provos and Nick Mathewson 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions 7 * are met: 8 * 1. Redistributions of source code must retain the above copyright 9 * notice, this list of conditions and the following disclaimer. 10 * 2. Redistributions in binary form must reproduce the above copyright 11 * notice, this list of conditions and the following disclaimer in the 12 * documentation and/or other materials provided with the distribution. 13 * 3. The name of the author may not be used to endorse or promote products 14 * derived from this software without specific prior written permission. 15 * 16 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR 17 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES 18 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. 19 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, 20 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT 21 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, 22 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 23 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 24 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF 25 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 26 */ 27 28 /** 29 @file buffer_iocp.c 30 31 This module implements overlapped read and write functions for evbuffer 32 objects on Windows. 33 */ 34 35 #include "event2/buffer.h" 36 #include "event2/buffer_compat.h" 37 #include "event2/util.h" 38 #include "event2/thread.h" 39 #include "event2/event-config.h" 40 #include <sys/cdefs.h> 41 __RCSID("$NetBSD: buffer_iocp.c,v 1.1.1.1 2013/04/11 16:43:25 christos Exp $"); 42 #include "util-internal.h" 43 #include "evthread-internal.h" 44 #include "evbuffer-internal.h" 45 #include "iocp-internal.h" 46 #include "mm-internal.h" 47 48 #include <winsock2.h> 49 #include <windows.h> 50 #include <stdio.h> 51 52 #define MAX_WSABUFS 16 53 54 /** An evbuffer that can handle overlapped IO. */ 55 struct evbuffer_overlapped { 56 struct evbuffer buffer; 57 /** The socket that we're doing overlapped IO on. */ 58 evutil_socket_t fd; 59 60 /** pending I/O type */ 61 unsigned read_in_progress : 1; 62 unsigned write_in_progress : 1; 63 64 /** The first pinned chain in the buffer. */ 65 struct evbuffer_chain *first_pinned; 66 67 /** How many chains are pinned; how many of the fields in buffers 68 * are we using. */ 69 int n_buffers; 70 WSABUF buffers[MAX_WSABUFS]; 71 }; 72 73 /** Given an evbuffer, return the correponding evbuffer structure, or NULL if 74 * the evbuffer isn't overlapped. */ 75 static inline struct evbuffer_overlapped * 76 upcast_evbuffer(struct evbuffer *buf) 77 { 78 if (!buf || !buf->is_overlapped) 79 return NULL; 80 return EVUTIL_UPCAST(buf, struct evbuffer_overlapped, buffer); 81 } 82 83 /** Unpin all the chains noted as pinned in 'eo'. */ 84 static void 85 pin_release(struct evbuffer_overlapped *eo, unsigned flag) 86 { 87 int i; 88 struct evbuffer_chain *next, *chain = eo->first_pinned; 89 90 for (i = 0; i < eo->n_buffers; ++i) { 91 EVUTIL_ASSERT(chain); 92 next = chain->next; 93 _evbuffer_chain_unpin(chain, flag); 94 chain = next; 95 } 96 } 97 98 void 99 evbuffer_commit_read(struct evbuffer *evbuf, ev_ssize_t nBytes) 100 { 101 struct evbuffer_overlapped *buf = upcast_evbuffer(evbuf); 102 struct evbuffer_chain **chainp; 103 size_t remaining, len; 104 unsigned i; 105 106 EVBUFFER_LOCK(evbuf); 107 EVUTIL_ASSERT(buf->read_in_progress && !buf->write_in_progress); 108 EVUTIL_ASSERT(nBytes >= 0); /* XXXX Can this be false? */ 109 110 evbuffer_unfreeze(evbuf, 0); 111 112 chainp = evbuf->last_with_datap; 113 if (!((*chainp)->flags & EVBUFFER_MEM_PINNED_R)) 114 chainp = &(*chainp)->next; 115 remaining = nBytes; 116 for (i = 0; remaining > 0 && i < (unsigned)buf->n_buffers; ++i) { 117 EVUTIL_ASSERT(*chainp); 118 len = buf->buffers[i].len; 119 if (remaining < len) 120 len = remaining; 121 (*chainp)->off += len; 122 evbuf->last_with_datap = chainp; 123 remaining -= len; 124 chainp = &(*chainp)->next; 125 } 126 127 pin_release(buf, EVBUFFER_MEM_PINNED_R); 128 129 buf->read_in_progress = 0; 130 131 evbuf->total_len += nBytes; 132 evbuf->n_add_for_cb += nBytes; 133 134 evbuffer_invoke_callbacks(evbuf); 135 136 _evbuffer_decref_and_unlock(evbuf); 137 } 138 139 void 140 evbuffer_commit_write(struct evbuffer *evbuf, ev_ssize_t nBytes) 141 { 142 struct evbuffer_overlapped *buf = upcast_evbuffer(evbuf); 143 144 EVBUFFER_LOCK(evbuf); 145 EVUTIL_ASSERT(buf->write_in_progress && !buf->read_in_progress); 146 evbuffer_unfreeze(evbuf, 1); 147 evbuffer_drain(evbuf, nBytes); 148 pin_release(buf,EVBUFFER_MEM_PINNED_W); 149 buf->write_in_progress = 0; 150 _evbuffer_decref_and_unlock(evbuf); 151 } 152 153 struct evbuffer * 154 evbuffer_overlapped_new(evutil_socket_t fd) 155 { 156 struct evbuffer_overlapped *evo; 157 158 evo = mm_calloc(1, sizeof(struct evbuffer_overlapped)); 159 if (!evo) 160 return NULL; 161 162 TAILQ_INIT(&evo->buffer.callbacks); 163 evo->buffer.refcnt = 1; 164 evo->buffer.last_with_datap = &evo->buffer.first; 165 166 evo->buffer.is_overlapped = 1; 167 evo->fd = fd; 168 169 return &evo->buffer; 170 } 171 172 int 173 evbuffer_launch_write(struct evbuffer *buf, ev_ssize_t at_most, 174 struct event_overlapped *ol) 175 { 176 struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf); 177 int r = -1; 178 int i; 179 struct evbuffer_chain *chain; 180 DWORD bytesSent; 181 182 if (!buf) { 183 /* No buffer, or it isn't overlapped */ 184 return -1; 185 } 186 187 EVBUFFER_LOCK(buf); 188 EVUTIL_ASSERT(!buf_o->read_in_progress); 189 if (buf->freeze_start || buf_o->write_in_progress) 190 goto done; 191 if (!buf->total_len) { 192 /* Nothing to write */ 193 r = 0; 194 goto done; 195 } else if (at_most < 0 || (size_t)at_most > buf->total_len) { 196 at_most = buf->total_len; 197 } 198 evbuffer_freeze(buf, 1); 199 200 buf_o->first_pinned = NULL; 201 buf_o->n_buffers = 0; 202 memset(buf_o->buffers, 0, sizeof(buf_o->buffers)); 203 204 chain = buf_o->first_pinned = buf->first; 205 206 for (i=0; i < MAX_WSABUFS && chain; ++i, chain=chain->next) { 207 WSABUF *b = &buf_o->buffers[i]; 208 b->buf = (char*)( chain->buffer + chain->misalign ); 209 _evbuffer_chain_pin(chain, EVBUFFER_MEM_PINNED_W); 210 211 if ((size_t)at_most > chain->off) { 212 /* XXXX Cast is safe for now, since win32 has no 213 mmaped chains. But later, we need to have this 214 add more WSAbufs if chain->off is greater than 215 ULONG_MAX */ 216 b->len = (unsigned long)chain->off; 217 at_most -= chain->off; 218 } else { 219 b->len = (unsigned long)at_most; 220 ++i; 221 break; 222 } 223 } 224 225 buf_o->n_buffers = i; 226 _evbuffer_incref(buf); 227 if (WSASend(buf_o->fd, buf_o->buffers, i, &bytesSent, 0, 228 &ol->overlapped, NULL)) { 229 int error = WSAGetLastError(); 230 if (error != WSA_IO_PENDING) { 231 /* An actual error. */ 232 pin_release(buf_o, EVBUFFER_MEM_PINNED_W); 233 evbuffer_unfreeze(buf, 1); 234 evbuffer_free(buf); /* decref */ 235 goto done; 236 } 237 } 238 239 buf_o->write_in_progress = 1; 240 r = 0; 241 done: 242 EVBUFFER_UNLOCK(buf); 243 return r; 244 } 245 246 int 247 evbuffer_launch_read(struct evbuffer *buf, size_t at_most, 248 struct event_overlapped *ol) 249 { 250 struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf); 251 int r = -1, i; 252 int nvecs; 253 int npin=0; 254 struct evbuffer_chain *chain=NULL, **chainp; 255 DWORD bytesRead; 256 DWORD flags = 0; 257 struct evbuffer_iovec vecs[MAX_WSABUFS]; 258 259 if (!buf_o) 260 return -1; 261 EVBUFFER_LOCK(buf); 262 EVUTIL_ASSERT(!buf_o->write_in_progress); 263 if (buf->freeze_end || buf_o->read_in_progress) 264 goto done; 265 266 buf_o->first_pinned = NULL; 267 buf_o->n_buffers = 0; 268 memset(buf_o->buffers, 0, sizeof(buf_o->buffers)); 269 270 if (_evbuffer_expand_fast(buf, at_most, MAX_WSABUFS) == -1) 271 goto done; 272 evbuffer_freeze(buf, 0); 273 274 nvecs = _evbuffer_read_setup_vecs(buf, at_most, 275 vecs, MAX_WSABUFS, &chainp, 1); 276 for (i=0;i<nvecs;++i) { 277 WSABUF_FROM_EVBUFFER_IOV( 278 &buf_o->buffers[i], 279 &vecs[i]); 280 } 281 282 buf_o->n_buffers = nvecs; 283 buf_o->first_pinned = chain = *chainp; 284 285 npin=0; 286 for ( ; chain; chain = chain->next) { 287 _evbuffer_chain_pin(chain, EVBUFFER_MEM_PINNED_R); 288 ++npin; 289 } 290 EVUTIL_ASSERT(npin == nvecs); 291 292 _evbuffer_incref(buf); 293 if (WSARecv(buf_o->fd, buf_o->buffers, nvecs, &bytesRead, &flags, 294 &ol->overlapped, NULL)) { 295 int error = WSAGetLastError(); 296 if (error != WSA_IO_PENDING) { 297 /* An actual error. */ 298 pin_release(buf_o, EVBUFFER_MEM_PINNED_R); 299 evbuffer_unfreeze(buf, 0); 300 evbuffer_free(buf); /* decref */ 301 goto done; 302 } 303 } 304 305 buf_o->read_in_progress = 1; 306 r = 0; 307 done: 308 EVBUFFER_UNLOCK(buf); 309 return r; 310 } 311 312 evutil_socket_t 313 _evbuffer_overlapped_get_fd(struct evbuffer *buf) 314 { 315 struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf); 316 return buf_o ? buf_o->fd : -1; 317 } 318 319 void 320 _evbuffer_overlapped_set_fd(struct evbuffer *buf, evutil_socket_t fd) 321 { 322 struct evbuffer_overlapped *buf_o = upcast_evbuffer(buf); 323 EVBUFFER_LOCK(buf); 324 /* XXX is this right?, should it cancel current I/O operations? */ 325 if (buf_o) 326 buf_o->fd = fd; 327 EVBUFFER_UNLOCK(buf); 328 } 329