Integrate #1579 - properly clear c->is_readable and c->is_writable flags in mg_iotest()

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);

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);

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();