contrib/libevent/buffer_iocp.c

   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 }