Add MAX_READ_BUFSIZE, MAX_WRITE_BUFSIZE limit

event/hevent.c

@@ -107,7 +107,7 @@ void hio_ready(hio_t* io) {
     io->read_until_length = 0;
     io->small_readbytes_cnt = 0;
     // write_queue
-    io->write_queue_bytes = 0;
+    io->write_bufsize = 0;
     // callbacks
     io->read_cb = NULL;
     io->write_cb = NULL;
@@ -248,7 +248,7 @@ size_t hio_read_bufsize(hio_t* io) {
 }
 
 size_t hio_write_bufsize(hio_t* io) {
-    return io->write_queue_bytes;
+    return io->write_bufsize;
 }
 
 haccept_cb hio_getcb_accept(hio_t* io) {
@@ -669,6 +669,11 @@ void hio_set_heartbeat(hio_t* io, int interval_ms, hio_send_heartbeat_fn fn) {
 }
 
 void hio_alloc_readbuf(hio_t* io, int len) {
+    if (len > MAX_READ_BUFSIZE) {
+        hloge("read bufsize > %u, close it!", (unsigned int)MAX_READ_BUFSIZE);
+        hio_close_async(io);
+        return;
+    }
     if (hio_is_alloced_readbuf(io)) {
         io->readbuf.base = (char*)safe_realloc(io->readbuf.base, len, io->readbuf.len);
     } else {

event/hevent.h

@@ -15,7 +15,9 @@
 
 #define HLOOP_READ_BUFSIZE          8192        // 8K
 #define READ_BUFSIZE_HIGH_WATER     65536       // 64K
-#define WRITE_QUEUE_HIGH_WATER      (1U << 23)  // 8M
+#define WRITE_BUFSIZE_HIGH_WATER    (1U << 23)  // 8M
+#define MAX_READ_BUFSIZE            (1U << 24)  // 16M
+#define MAX_WRITE_BUFSIZE           (1U << 26)  // 64M
 
 // hio_read_flags
 #define HIO_READ_ONCE           0x1
@@ -133,7 +135,7 @@ struct hio_s {
     // write
     struct write_queue  write_queue;
     hrecursive_mutex_t  write_mutex; // lock write and write_queue
-    uint32_t            write_queue_bytes;
+    uint32_t            write_bufsize;
     // callbacks
     hread_cb    read_cb;
     hwrite_cb   write_cb;

event/nio.c

@@ -334,7 +334,7 @@ write:
     }
     __write_cb(io, buf, nwrite);
     pbuf->offset += nwrite;
-    io->write_queue_bytes -= nwrite;
+    io->write_bufsize -= nwrite;
     if (nwrite == len) {
         HV_FREE(pbuf->base);
         write_queue_pop_front(&io->write_queue);
@@ -460,22 +460,29 @@ enqueue:
         hio_add(io, hio_handle_events, HV_WRITE);
     }
     if (nwrite < len) {
+        if (io->write_bufsize + len - nwrite > MAX_WRITE_BUFSIZE) {
+            if (io->write_bufsize > MAX_WRITE_BUFSIZE) {
+                hloge("write bufsize > %u, close it!", (unsigned int)MAX_WRITE_BUFSIZE);
+                hio_close_async(io);
+                return -1;
+            }
+        }
         offset_buf_t remain;
-        remain.len = len;
-        remain.offset = nwrite;
+        remain.len = len - nwrite;
+        remain.offset = 0;
         // NOTE: free in nio_write
         HV_ALLOC(remain.base, remain.len);
-        memcpy(remain.base, buf, remain.len);
+        memcpy(remain.base, ((char*)buf) + nwrite, remain.len);
         if (io->write_queue.maxsize == 0) {
             write_queue_init(&io->write_queue, 4);
         }
         write_queue_push_back(&io->write_queue, &remain);
-        io->write_queue_bytes += remain.len - remain.offset;
-        if (io->write_queue_bytes > WRITE_QUEUE_HIGH_WATER) {
-            hlogw("write queue %u, total %u, over high water %u",
-                (unsigned int)(remain.len - remain.offset),
-                (unsigned int)io->write_queue_bytes,
-                (unsigned int)WRITE_QUEUE_HIGH_WATER);
+        io->write_bufsize += remain.len;
+        if (io->write_bufsize > WRITE_BUFSIZE_HIGH_WATER) {
+            hlogw("write len=%d enqueue %u, bufsize=%u over high water %u",
+                len, (unsigned int)(remain.len - remain.offset),
+                (unsigned int)io->write_bufsize,
+                (unsigned int)WRITE_BUFSIZE_HIGH_WATER);
         }
     }
     hrecursive_mutex_unlock(&io->write_mutex);