Make IOCP rate-limiting group support stricter and less surprising.

Previously, we wouldn't decrement read/write buckets because of IOCP
reads and writes until those reads and writes were complete.  That's
not so bad on the per-connection front.  But for group limits, the
old approach makes us launch a huge amount of reads and writes
whenever the group limit becomes positive, and then decrement the
limit to a hugely negative number as they complete.

With this patch, we decrement our read buckets whenever we launch an
IOCP read or write, based on the maximum that tried to read or
write.  Later, when the operations finish, we re-increment the
bucket based on the portion of the request that couldn't finish.

diff --git a/bufferevent_async.c b/bufferevent_async.c
index 87ba404c014f680289fdeceed294728c3a948ad6..9416e31f2d75e3bb9e281ad91fd6a97c9f60fba1 100644 (file)
--- a/bufferevent_async.c
+++ b/bufferevent_async.c
@@ -80,8 +80,8 @@ struct bufferevent_async {
        struct event_overlapped connect_overlapped;
        struct event_overlapped read_overlapped;
        struct event_overlapped write_overlapped;
-       unsigned read_in_progress : 1;
-       unsigned write_in_progress : 1;
+       size_t read_in_progress;
+       size_t write_in_progress;
        unsigned ok : 1;
        unsigned read_added : 1;
        unsigned write_added : 1;
@@ -198,7 +198,6 @@ bev_async_consider_writing(struct bufferevent_async *beva)
 
        at_most = evbuffer_get_length(bev->output);
 
-       /* XXXX This over-commits. */
        /* This is safe so long as bufferevent_get_write_max never returns
         * more than INT_MAX.  That's true for now. XXXX */
        limit = (int)_bufferevent_get_write_max(&beva->bev);
@@ -218,7 +217,8 @@ bev_async_consider_writing(struct bufferevent_async *beva)
                beva->ok = 0;
                _bufferevent_run_eventcb(bev, BEV_EVENT_ERROR);
        } else {
-               beva->write_in_progress = 1;
+               beva->write_in_progress = at_most;
+               _bufferevent_decrement_write_buckets(&beva->bev, at_most);
                bev_async_add_write(beva);
        }
 }
@@ -271,7 +271,8 @@ bev_async_consider_reading(struct bufferevent_async *beva)
                bufferevent_decref(bev);
                _bufferevent_run_eventcb(bev, BEV_EVENT_ERROR);
        } else {
-               beva->read_in_progress = 1;
+               beva->read_in_progress = at_most;
+               _bufferevent_decrement_read_buckets(&beva->bev, at_most);
                bev_async_add_read(beva);
        }
 
@@ -442,12 +443,15 @@ read_complete(struct event_overlapped *eo, ev_uintptr_t key,
        struct bufferevent_async *bev_a = upcast_read(eo);
        struct bufferevent *bev = &bev_a->bev.bev;
        short what = BEV_EVENT_READING;
-
+       ev_ssize_t amount_unread;
        BEV_LOCK(bev);
        EVUTIL_ASSERT(bev_a->read_in_progress);
 
+       amount_unread = bev_a->read_in_progress - nbytes;
        evbuffer_commit_read(bev->input, nbytes);
        bev_a->read_in_progress = 0;
+       if (amount_unread)
+               _bufferevent_decrement_read_buckets(&bev_a->bev, -amount_unread);
 
        if (!ok)
                bev_async_set_wsa_error(bev, eo);
@@ -455,8 +459,6 @@ read_complete(struct event_overlapped *eo, ev_uintptr_t key,
        if (bev_a->ok) {
                if (ok && nbytes) {
                        BEV_RESET_GENERIC_READ_TIMEOUT(bev);
-                       _bufferevent_decrement_read_buckets(&bev_a->bev,
-                                       nbytes);
                        if (evbuffer_get_length(bev->input) >= bev->wm_read.low)
                                _bufferevent_run_readcb(bev);
                        bev_async_consider_reading(bev_a);
@@ -481,20 +483,26 @@ write_complete(struct event_overlapped *eo, ev_uintptr_t key,
        struct bufferevent_async *bev_a = upcast_write(eo);
        struct bufferevent *bev = &bev_a->bev.bev;
        short what = BEV_EVENT_WRITING;
+       ev_ssize_t amount_unwritten;
 
        BEV_LOCK(bev);
        EVUTIL_ASSERT(bev_a->write_in_progress);
+
+       amount_unwritten = bev_a->write_in_progress - nbytes;
        evbuffer_commit_write(bev->output, nbytes);
        bev_a->write_in_progress = 0;
 
+       if (amount_unwritten)
+               _bufferevent_decrement_write_buckets(&bev_a->bev,
+                                                    -amount_unwritten);
+
+
        if (!ok)
                bev_async_set_wsa_error(bev, eo);
 
        if (bev_a->ok) {
                if (ok && nbytes) {
                        BEV_RESET_GENERIC_WRITE_TIMEOUT(bev);
-                       _bufferevent_decrement_write_buckets(&bev_a->bev,
-                                       nbytes);
                        if (evbuffer_get_length(bev->output) <=
                            bev->wm_write.low)
                                _bufferevent_run_writecb(bev);