unix, windows: don't read/recv if buf.len==0

Make it possible for the libuv user to handle out of memory conditions
gracefully. When alloc_cb() returns a buffer with len==0, call the read
or recv callback with nread=UV_ENOBUFS. It's up to the user to stop or
close the handle.

Fixes #752.

include/uv.h

@@ -342,6 +342,12 @@ UV_EXTERN int uv_backend_timeout(const uv_loop_t*);
  *
  * `suggested_size` is a hint. Returning a buffer that is smaller is perfectly
  * okay as long as `buf.len > 0`.
+ *
+ * If you return a buffer with `buf.len == 0`, libuv skips the read and calls
+ * your read or recv callback with nread=UV_ENOBUFS.
+ *
+ * Note that returning a zero-length buffer does not stop the handle, call
+ * uv_read_stop() or uv_udp_recv_stop() for that.
  */
 typedef uv_buf_t (*uv_alloc_cb)(uv_handle_t* handle, size_t suggested_size);
 

src/unix/stream.c

@@ -977,11 +977,16 @@ static void uv__read(uv_stream_t* stream) {
   while ((stream->read_cb || stream->read2_cb)
       && (stream->flags & UV_STREAM_READING)
       && (count-- > 0)) {
-    assert(stream->alloc_cb);
-    buf = stream->alloc_cb((uv_handle_t*)stream, 64 * 1024);
+    assert(stream->alloc_cb != NULL);
 
-    assert(buf.len > 0);
-    assert(buf.base);
+    buf = stream->alloc_cb((uv_handle_t*)stream, 64 * 1024);
+    if (buf.len == 0) {
+      /* User indicates it can't or won't handle the read. */
+      uv__stream_read_cb(stream, UV_ENOBUFS, buf, UV_UNKNOWN_HANDLE);
+      return;
+    }
+
+    assert(buf.base != NULL);
     assert(uv__stream_fd(stream) >= 0);
 
     if (stream->read_cb) {

src/unix/udp.c

@@ -205,8 +205,11 @@ static void uv__udp_recvmsg(uv_loop_t* loop,
   h.msg_name = &peer;
 
   do {
-    buf = handle->alloc_cb((uv_handle_t*)handle, 64 * 1024);
-    assert(buf.len > 0);
+    buf = handle->alloc_cb((uv_handle_t*) handle, 64 * 1024);
+    if (buf.len == 0) {
+      handle->recv_cb(handle, UV_ENOBUFS, buf, NULL, 0);
+      return;
+    }
     assert(buf.base != NULL);
 
     h.msg_namelen = sizeof(peer);

src/win/pipe.c

@@ -1429,7 +1429,15 @@ void uv_process_pipe_read_req(uv_loop_t* loop, uv_pipe_t* handle,
       }
 
       buf = handle->alloc_cb((uv_handle_t*) handle, avail);
-      assert(buf.len > 0);
+      if (buf.len == 0) {
+        if (handle->read2_cb) {
+          handle->read2_cb(handle, UV_ENOBUFS, buf, UV_UNKNOWN_HANDLE);
+        } else if (handle->read_cb) {
+          handle->read_cb((uv_stream_t*) handle, UV_ENOBUFS, buf);
+        }
+        break;
+      }
+      assert(buf.base != NULL);
 
       if (ReadFile(handle->handle,
                    buf.base,

src/win/tcp.c

@@ -434,7 +434,11 @@ static void uv_tcp_queue_read(uv_loop_t* loop, uv_tcp_t* handle) {
   if (loop->active_tcp_streams < uv_active_tcp_streams_threshold) {
     handle->flags &= ~UV_HANDLE_ZERO_READ;
     handle->read_buffer = handle->alloc_cb((uv_handle_t*) handle, 65536);
-    assert(handle->read_buffer.len > 0);
+    if (handle->read_buffer.len == 0) {
+      handle->read_cb((uv_stream_t*) handle, UV_ENOBUFS, buf);
+      return;
+    }
+    assert(handle->read_buffer.base != NULL);
     buf = handle->read_buffer;
   } else {
     handle->flags |= UV_HANDLE_ZERO_READ;
@@ -943,7 +947,12 @@ void uv_process_tcp_read_req(uv_loop_t* loop, uv_tcp_t* handle,
     /* Do nonblocking reads until the buffer is empty */
     while (handle->flags & UV_HANDLE_READING) {
       buf = handle->alloc_cb((uv_handle_t*) handle, 65536);
-      assert(buf.len > 0);
+      if (buf.len == 0) {
+        handle->read_cb(handle, UV_ENOBUFS, buf);
+        break;
+      }
+      assert(buf.base != NULL);
+
       flags = 0;
       if (WSARecv(handle->socket,
                   (WSABUF*)&buf,

src/win/tty.c

@@ -348,8 +348,11 @@ static void uv_tty_queue_read_line(uv_loop_t* loop, uv_tty_t* handle) {
   memset(&req->overlapped, 0, sizeof(req->overlapped));
 
   handle->read_line_buffer = handle->alloc_cb((uv_handle_t*) handle, 8192);
+  if (handle->read_line_buffer.len == 0) {
+    handle->read_cb(handle, UV_ENOBUFS, handle->read_line_buffer);
+    return;
+  }
   assert(handle->read_line_buffer.base != NULL);
-  assert(handle->read_line_buffer.len > 0);
 
   /* Duplicate the console handle, so if we want to cancel the read, we can */
   /* just close this handle duplicate. */
@@ -682,6 +685,11 @@ void uv_process_tty_read_raw_req(uv_loop_t* loop, uv_tty_t* handle,
         /* Allocate a buffer if needed */
         if (buf_used == 0) {
           buf = handle->alloc_cb((uv_handle_t*) handle, 1024);
+          if (buf.len == 0) {
+            handle->read_cb((uv_stream_t*) handle, UV_ENOBUFS, buf);
+            goto out;
+          }
+          assert(buf.base != NULL);
         }
 
         buf.base[buf_used++] = handle->last_key[handle->last_key_offset++];

src/win/udp.c

@@ -272,7 +272,11 @@ static void uv_udp_queue_recv(uv_loop_t* loop, uv_udp_t* handle) {
     handle->flags &= ~UV_HANDLE_ZERO_READ;
 
     handle->recv_buffer = handle->alloc_cb((uv_handle_t*) handle, 65536);
-    assert(handle->recv_buffer.len > 0);
+    if (handle->recv_buffer.len == 0) {
+      handle->recv_cb(handle, UV_ENOBUFS, handle->recv_buffer, NULL, 0);
+      return;
+    }
+    assert(handle->recv_buffer.base != NULL);
 
     buf = handle->recv_buffer;
     memset(&handle->recv_from, 0, sizeof handle->recv_from);
@@ -515,7 +519,11 @@ void uv_process_udp_recv_req(uv_loop_t* loop, uv_udp_t* handle,
     /* Do a nonblocking receive */
     /* TODO: try to read multiple datagrams at once. FIONREAD maybe? */
     buf = handle->alloc_cb((uv_handle_t*) handle, 65536);
-    assert(buf.len > 0);
+    if (buf.len == 0) {
+      handle->recv_cb(handle, UV_ENOBUFS, buf, NULL, 0);
+      goto done;
+    }
+    assert(buf.base != NULL);
 
     memset(&from, 0, sizeof from);
     from_len = sizeof from;