IOCP-related evbuffer fixes.

- Prevent evbuffer_{add,prepend}_buffer from moving read-pinned chains.
- Fix evbuffer_drain to handle read-pinned chains better.
- Raise the limit on WSABUFs from two to MAX_WSABUFS for overlapped reads.

diff --git a/buffer.c b/buffer.c
index 098afec360501a534f4ac1dbb8528ceb4bf57792..8d0038c7b584b2cf75e2f02feca5a9959f21297b 100644 (file)
--- a/buffer.c
+++ b/buffer.c
@@ -575,7 +575,8 @@ evbuffer_reserve_space(struct evbuffer *buf, ev_ssize_t size,
        } else {
                if (_evbuffer_expand_fast(buf, size, n_vecs)<0)
                        goto done;
-               n = _evbuffer_read_setup_vecs(buf, size, vec, n_vecs, &chainp, 0);
+               n = _evbuffer_read_setup_vecs(buf, size, vec, n_vecs,
+                               &chainp, 0);
        }
 
 done:
@@ -670,6 +671,12 @@ done:
        return result;
 }
 
+static inline int
+HAS_PINNED_R(struct evbuffer *buf)
+{
+       return (buf->last && CHAIN_PINNED_R(buf->last));
+}
+
 static inline void
 ZERO_CHAIN(struct evbuffer *dst)
 {
@@ -680,6 +687,71 @@ ZERO_CHAIN(struct evbuffer *dst)
        dst->total_len = 0;
 }
 
+/* Prepares the contents of src to be moved to another buffer by removing
+ * read-pinned chains. The first pinned chain is saved in first, and the
+ * last in last. If src has no read-pinned chains, first and last are set
+ * to NULL. */
+static int
+PRESERVE_PINNED(struct evbuffer *src, struct evbuffer_chain **first,
+               struct evbuffer_chain **last)
+{
+       struct evbuffer_chain *chain, **pinned;
+
+       ASSERT_EVBUFFER_LOCKED(src);
+
+       if (!HAS_PINNED_R(src)) {
+               *first = *last = NULL;
+               return 0;
+       }
+
+       pinned = src->last_with_datap;
+       if (!CHAIN_PINNED_R(*pinned))
+               pinned = &(*pinned)->next;
+       EVUTIL_ASSERT(CHAIN_PINNED_R(*pinned));
+       chain = *first = *pinned;
+       *last = src->last;
+
+       /* If there's data in the first pinned chain, we need to allocate
+        * a new chain and copy the data over. */
+       if (chain->off) {
+               struct evbuffer_chain *tmp;
+
+               EVUTIL_ASSERT(pinned == src->last_with_datap);
+               tmp = evbuffer_chain_new(chain->off);
+               if (!tmp)
+                       return -1;
+               memcpy(tmp->buffer, chain->buffer + chain->misalign,
+                       chain->off);
+               tmp->off = chain->off;
+               *src->last_with_datap = tmp;
+               src->last = tmp;
+               chain->misalign += chain->off;
+               chain->off = 0;
+       } else {
+               src->last = *src->last_with_datap;
+               *pinned = NULL;
+       }
+
+       return 0;
+}
+
+static inline void
+RESTORE_PINNED(struct evbuffer *src, struct evbuffer_chain *pinned,
+               struct evbuffer_chain *last)
+{
+       ASSERT_EVBUFFER_LOCKED(src);
+
+       if (!pinned) {
+               ZERO_CHAIN(src);
+               return;
+       }
+
+       src->first = pinned;
+       src->last = last;
+       src->last_with_datap = &src->first;
+       src->total_len = 0;
+}
+
 static inline void
 COPY_CHAIN(struct evbuffer *dst, struct evbuffer *src)
 {
@@ -729,6 +801,7 @@ PREPEND_CHAIN(struct evbuffer *dst, struct evbuffer *src)
 int
 evbuffer_add_buffer(struct evbuffer *outbuf, struct evbuffer *inbuf)
 {
+       struct evbuffer_chain *pinned, *last;
        size_t in_total_len, out_total_len;
        int result = 0;
 
@@ -744,6 +817,11 @@ evbuffer_add_buffer(struct evbuffer *outbuf, struct evbuffer *inbuf)
                goto done;
        }
 
+       if (PRESERVE_PINNED(inbuf, &pinned, &last) < 0) {
+               result = -1;
+               goto done;
+       }
+
        if (out_total_len == 0) {
                /* There might be an empty chain at the start of outbuf; free
                 * it. */
@@ -753,8 +831,8 @@ evbuffer_add_buffer(struct evbuffer *outbuf, struct evbuffer *inbuf)
                APPEND_CHAIN(outbuf, inbuf);
        }
 
-       /* remove everything from inbuf */
-       ZERO_CHAIN(inbuf);
+       RESTORE_PINNED(inbuf, pinned, last);
+
        inbuf->n_del_for_cb += in_total_len;
        outbuf->n_add_for_cb += in_total_len;
 
@@ -769,6 +847,7 @@ done:
 int
 evbuffer_prepend_buffer(struct evbuffer *outbuf, struct evbuffer *inbuf)
 {
+       struct evbuffer_chain *pinned, *last;
        size_t in_total_len, out_total_len;
        int result = 0;
 
@@ -785,6 +864,11 @@ evbuffer_prepend_buffer(struct evbuffer *outbuf, struct evbuffer *inbuf)
                goto done;
        }
 
+       if (PRESERVE_PINNED(inbuf, &pinned, &last) < 0) {
+               result = -1;
+               goto done;
+       }
+
        if (out_total_len == 0) {
                /* There might be an empty chain at the start of outbuf; free
                 * it. */
@@ -794,8 +878,8 @@ evbuffer_prepend_buffer(struct evbuffer *outbuf, struct evbuffer *inbuf)
                PREPEND_CHAIN(outbuf, inbuf);
        }
 
-       /* remove everything from inbuf */
-       ZERO_CHAIN(inbuf);
+       RESTORE_PINNED(inbuf, pinned, last);
+
        inbuf->n_del_for_cb += in_total_len;
        outbuf->n_add_for_cb += in_total_len;
 
@@ -810,7 +894,7 @@ int
 evbuffer_drain(struct evbuffer *buf, size_t len)
 {
        struct evbuffer_chain *chain, *next;
-       size_t old_len;
+       size_t remaining, old_len;
        int result = 0;
 
        EVBUFFER_LOCK(buf);
@@ -824,12 +908,10 @@ evbuffer_drain(struct evbuffer *buf, size_t len)
                goto done;
        }
 
-
-       if (len >= old_len && !(buf->last && CHAIN_PINNED_R(buf->last))) {
+       if (len >= old_len && !HAS_PINNED_R(buf)) {
                len = old_len;
                for (chain = buf->first; chain != NULL; chain = next) {
                        next = chain->next;
-
                        evbuffer_chain_free(chain);
                }
 
@@ -839,10 +921,12 @@ evbuffer_drain(struct evbuffer *buf, size_t len)
                        len = old_len;
 
                buf->total_len -= len;
-
-               for (chain = buf->first; len >= chain->off; chain = next) {
+               remaining = len;
+               for (chain = buf->first;
+                    remaining >= chain->off;
+                    chain = next) {
                        next = chain->next;
-                       len -= chain->off;
+                       remaining -= chain->off;
 
                        if (chain == *buf->last_with_datap) {
                                buf->last_with_datap = &buf->first;
@@ -850,14 +934,20 @@ evbuffer_drain(struct evbuffer *buf, size_t len)
                        if (&chain->next == buf->last_with_datap)
                                buf->last_with_datap = &buf->first;
 
-                       if (len == 0 && CHAIN_PINNED_R(chain))
+                       if (CHAIN_PINNED_R(chain)) {
+                               EVUTIL_ASSERT(remaining == 0);
+                               chain->misalign += chain->off;
+                               chain->off = 0;
                                break;
-                       evbuffer_chain_free(chain);
+                       } else
+                               evbuffer_chain_free(chain);
                }
 
                buf->first = chain;
-               chain->misalign += len;
-               chain->off -= len;
+               if (chain) {
+                       chain->misalign += remaining;
+                       chain->off -= remaining;
+               }
        }
 
        buf->n_del_for_cb += len;

diff --git a/buffer_iocp.c b/buffer_iocp.c
index 8255599c22b3a65949e9f7c2285824d337bca957..e211e980c774885b3f0858bdbf502f1b39cb02dd 100644 (file)
--- a/buffer_iocp.c
+++ b/buffer_iocp.c
@@ -82,12 +82,13 @@ static void
 pin_release(struct evbuffer_overlapped *eo, unsigned flag)
 {
        int i;
-       struct evbuffer_chain *chain = eo->first_pinned;
+       struct evbuffer_chain *next, *chain = eo->first_pinned;
 
        for (i = 0; i < eo->n_buffers; ++i) {
                EVUTIL_ASSERT(chain);
+               next = chain->next;
                _evbuffer_chain_unpin(chain, flag);
-               chain = chain->next;
+               chain = next;
        }
 }
 
@@ -95,8 +96,9 @@ void
 evbuffer_commit_read(struct evbuffer *evbuf, ev_ssize_t nBytes)
 {
        struct evbuffer_overlapped *buf = upcast_evbuffer(evbuf);
-       struct evbuffer_iovec iov[2];
-       int n_vec;
+       struct evbuffer_chain **chainp;
+       size_t remaining, len;
+       unsigned i;
 
        EVBUFFER_LOCK(evbuf);
        EVUTIL_ASSERT(buf->read_in_progress && !buf->write_in_progress);
@@ -104,24 +106,27 @@ evbuffer_commit_read(struct evbuffer *evbuf, ev_ssize_t nBytes)
 
        evbuffer_unfreeze(evbuf, 0);
 
-       iov[0].iov_base = buf->buffers[0].buf;
-       if ((size_t)nBytes <= buf->buffers[0].len) {
-               iov[0].iov_len = nBytes;
-               n_vec = 1;
-       } else {
-               iov[0].iov_len = buf->buffers[0].len;
-               iov[1].iov_base = buf->buffers[1].buf;
-               iov[1].iov_len = nBytes - iov[0].iov_len;
-               n_vec = 2;
+       chainp = evbuf->last_with_datap;
+       if (!((*chainp)->flags & EVBUFFER_MEM_PINNED_R))
+               chainp = &(*chainp)->next;
+       remaining = nBytes;
+       for (i = 0; remaining > 0 && i < buf->n_buffers; ++i) {
+               EVUTIL_ASSERT(*chainp);
+               len = buf->buffers[i].len;
+               if (remaining < len)
+                       len = remaining;
+               (*chainp)->off += len;
+               evbuf->last_with_datap = chainp;
+               remaining -= len;
+               chainp = &(*chainp)->next;
        }
 
-       if (evbuffer_commit_space(evbuf, iov, n_vec) < 0)
-               EVUTIL_ASSERT(0); /* XXXX fail nicer. */
-
        pin_release(buf, EVBUFFER_MEM_PINNED_R);
 
        buf->read_in_progress = 0;
 
+       evbuf->total_len += nBytes;
+
        _evbuffer_decref_and_unlock(evbuf);
 }
 
@@ -184,7 +189,7 @@ evbuffer_launch_write(struct evbuffer *buf, ev_ssize_t at_most,
        }
        evbuffer_freeze(buf, 1);
 
-       buf_o->first_pinned = 0;
+       buf_o->first_pinned = NULL;
        buf_o->n_buffers = 0;
        memset(buf_o->buffers, 0, sizeof(buf_o->buffers));
 
@@ -246,19 +251,16 @@ evbuffer_launch_read(struct evbuffer *buf, size_t at_most,
        if (buf->freeze_end || buf_o->read_in_progress)
                goto done;
 
-       buf_o->first_pinned = 0;
+       buf_o->first_pinned = NULL;
        buf_o->n_buffers = 0;
        memset(buf_o->buffers, 0, sizeof(buf_o->buffers));
 
-       if (_evbuffer_expand_fast(buf, at_most, 2) == -1)
+       if (_evbuffer_expand_fast(buf, at_most, MAX_WSABUFS) == -1)
                goto done;
        evbuffer_freeze(buf, 0);
 
-       /* XXX This and evbuffer_read_setup_vecs() should say MAX_WSABUFS,
-        * not "2".  But commit_read() above can't handle more than two
-        * buffers yet. */
        nvecs = _evbuffer_read_setup_vecs(buf, at_most,
-           vecs, 2, &chainp, 1);
+           vecs, MAX_WSABUFS, &chainp, 1);
        for (i=0;i<nvecs;++i) {
                WSABUF_FROM_EVBUFFER_IOV(
                        &buf_o->buffers[i],
@@ -266,7 +268,8 @@ evbuffer_launch_read(struct evbuffer *buf, size_t at_most,
        }
 
        buf_o->n_buffers = nvecs;
-       buf_o->first_pinned = chain= *chainp;
+       buf_o->first_pinned = chain = *chainp;
+
        npin=0;
        for ( ; chain; chain = chain->next) {
                _evbuffer_chain_pin(chain, EVBUFFER_MEM_PINNED_R);

diff --git a/evbuffer-internal.h b/evbuffer-internal.h
index 90fd1b55f9be0185a3f5b7792b11d9b6f1b996fb..719e85dd0df67dda4d0bca5b54ce355e947fffae 100644 (file)
--- a/evbuffer-internal.h
+++ b/evbuffer-internal.h
@@ -256,7 +256,8 @@ int _evbuffer_expand_fast(struct evbuffer *, size_t, int);
  * Returns the number of vecs used.
  */
 int _evbuffer_read_setup_vecs(struct evbuffer *buf, ev_ssize_t howmuch,
-    struct evbuffer_iovec *vecs, int n_vecs, struct evbuffer_chain ***chainp, int exact);
+    struct evbuffer_iovec *vecs, int n_vecs, struct evbuffer_chain ***chainp,
+    int exact);
 
 /* Helper macro: copies an evbuffer_iovec in ei to a win32 WSABUF in i. */
 #define WSABUF_FROM_EVBUFFER_IOV(i,ei) do {            \