libevent/buffer_iocp.c

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