diff --git a/mongoose.c b/mongoose.c index f39efa7d..538943a2 100644 --- a/mongoose.c +++ b/mongoose.c @@ -4428,6 +4428,7 @@ static void mg_iotest(struct mg_mgr *mgr, int ms) { #if MG_ARCH == MG_ARCH_FREERTOS_TCP struct mg_connection *c; for (c = mgr->conns; c != NULL; c = c->next) { + c->is_readable = c->is_writable = 0; if (skip_iotest(c)) continue; if (can_read(c)) FreeRTOS_FD_SET(c->fd, mgr->ss, eSELECT_READ | eSELECT_EXCEPT); @@ -4449,6 +4450,7 @@ static void mg_iotest(struct mg_mgr *mgr, int ms) { memset(fds, 0, sizeof(fds)); n = 0; for (struct mg_connection *c = mgr->conns; c != NULL; c = c->next) { + c->is_readable = c->is_writable = 0; if (skip_iotest(c)) { // Socket not valid, ignore } else { @@ -4487,6 +4489,7 @@ static void mg_iotest(struct mg_mgr *mgr, int ms) { FD_ZERO(&wset); for (c = mgr->conns; c != NULL; c = c->next) { + c->is_readable = c->is_writable = 0; if (skip_iotest(c)) continue; if (can_read(c)) FD_SET(FD(c), &rset); if (can_write(c)) FD_SET(FD(c), &wset); diff --git a/src/sock.c b/src/sock.c index a862d57b..8e50b945 100644 --- a/src/sock.c +++ b/src/sock.c @@ -486,6 +486,7 @@ static void mg_iotest(struct mg_mgr *mgr, int ms) { #if MG_ARCH == MG_ARCH_FREERTOS_TCP struct mg_connection *c; for (c = mgr->conns; c != NULL; c = c->next) { + c->is_readable = c->is_writable = 0; if (skip_iotest(c)) continue; if (can_read(c)) FreeRTOS_FD_SET(c->fd, mgr->ss, eSELECT_READ | eSELECT_EXCEPT); @@ -507,6 +508,7 @@ static void mg_iotest(struct mg_mgr *mgr, int ms) { memset(fds, 0, sizeof(fds)); n = 0; for (struct mg_connection *c = mgr->conns; c != NULL; c = c->next) { + c->is_readable = c->is_writable = 0; if (skip_iotest(c)) { // Socket not valid, ignore } else { @@ -545,6 +547,7 @@ static void mg_iotest(struct mg_mgr *mgr, int ms) { FD_ZERO(&wset); for (c = mgr->conns; c != NULL; c = c->next) { + c->is_readable = c->is_writable = 0; if (skip_iotest(c)) continue; if (can_read(c)) FD_SET(FD(c), &rset); if (can_write(c)) FD_SET(FD(c), &wset); diff --git a/test/unit_test.c b/test/unit_test.c index 15921d80..156fa107 100644 --- a/test/unit_test.c +++ b/test/unit_test.c @@ -1796,6 +1796,102 @@ static void test_invalid_listen_addr(void) { ASSERT(mgr.conns == NULL); } +struct stream_status { + uint32_t polls; + size_t sent; + size_t received; + uint32_t send_crc; + uint32_t recv_crc; +}; + +// Consume recv buffer after letting it reach MG_MAX_RECV_SIZE +static void eh8(struct mg_connection *c, int ev, void *ev_data, void *fn_data) { + struct stream_status *status = (struct stream_status *) fn_data; + if (c->is_listening) + return; + + ASSERT(c->recv.len <= MG_MAX_RECV_SIZE); + + if (ev == MG_EV_ACCEPT) { + // Optimize recv buffer size near max to speed up test + mg_iobuf_resize(&c->recv, MG_MAX_RECV_SIZE - MG_IO_SIZE); + status->received = 0; + status->recv_crc = 0; + } + + if (ev == MG_EV_CLOSE) { + ASSERT(status->received == status->sent); + } + + // Let buffer fill up and start consuming after 10 full buffer poll events + if (status->polls >= 10 && ev == MG_EV_POLL) { + // consume at most a third of MG_MAX_RECV_SIZE on each poll + size_t consume; + if (MG_MAX_RECV_SIZE / 3 >= c->recv.len) + consume = c->recv.len; + else + consume = MG_MAX_RECV_SIZE / 3; + status->received += consume; + status->recv_crc = mg_crc32(status->recv_crc, (const char *) c->recv.buf, consume); + mg_iobuf_del(&c->recv, 0, consume); + } + + // count polls with full buffer to ensure c->is_full prevents reads + if (ev == MG_EV_POLL && c->recv.len == MG_MAX_RECV_SIZE) + status->polls += 1; + (void) ev_data; +} + +// Toggle c->is_full to prevent max_recv_buf_size reached read errors +static void eh10(struct mg_connection *c, int ev, void *ev_data, void *fn_data) { + if (c->recv.len >= MG_MAX_RECV_SIZE && ev == MG_EV_READ) + c->is_full = true; + + eh8(c, ev, ev_data, fn_data); + + if (c->recv.len < MG_MAX_RECV_SIZE && ev == MG_EV_POLL) + c->is_full = false; +} + +// Send buffer larger than MG_MAX_RECV_SIZE to server +static void eh11(struct mg_connection *c, int ev, void *ev_data, void *fn_data) { + struct stream_status *status = (struct stream_status *) fn_data; + if (ev == MG_EV_CONNECT) { + size_t len = MG_MAX_RECV_SIZE * 2; + struct mg_iobuf buf = {NULL, 0, 0}; + mg_iobuf_init(&buf, len); + mg_random(buf.buf, buf.size); + buf.len = buf.size; + mg_send(c, buf.buf, buf.len); + status->sent = buf.len; + status->send_crc = mg_crc32(0, (const char *) buf.buf, buf.len); + mg_iobuf_free(&buf); + } + (void) ev_data; + (void) fn_data; +} + +static void test_http_stream_buffer(void) { + struct mg_mgr mgr; + const char *url = "tcp://127.0.0.1:12344"; + uint32_t i; + struct stream_status status; + mg_mgr_init(&mgr); + mg_listen(&mgr, url, eh10, &status); + + status.polls = 0; + mg_connect(&mgr, url, eh11, &status); + for (i = 0; i < (MG_MAX_RECV_SIZE / MG_IO_SIZE) * 50; i++) { + mg_mgr_poll(&mgr, 1); + if (status.polls >= 10 && status.sent == status.received) break; + } + ASSERT(status.sent == status.received); + ASSERT(status.send_crc == status.recv_crc); + + mg_mgr_free(&mgr); + ASSERT(mgr.conns == NULL); +} + static void test_multipart(void) { struct mg_http_part part; size_t ofs; @@ -2072,6 +2168,7 @@ int main(void) { test_invalid_listen_addr(); test_http_chunked(); test_http_upload(); + test_http_stream_buffer(); test_http_parse(); test_util(); test_sntp();