buffer_iocp.c (8661B)
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 }