Avoid event loop starvation.

Author: rolftimmermans

CHANGELOG.md

@@ -1,3 +1,7 @@
+### Unreleased
+
+* Break out of busy loops automatically when the number of synchronous I/O operations moves beyond a built-in threshold. This avoids the ZeroMQ background I/O process(es) starving the Node.js event loop when it can process messages faster than the application, potentially causing decreased responsiveness and/or high memory usage.
+
 ### v6.0.0-beta.3
 
 * Error details have been added to the "handshake:error:protocol" and "handshake:error:auth" events.

src/incoming_msg.cc

@@ -21,7 +21,7 @@ Napi::Value IncomingMsg::IntoBuffer(const Napi::Env& env) {
         return env.Undefined();
     }
 
-    static auto constexpr zero_copy_threshold = 32;
+    static auto constexpr zero_copy_threshold = 1 << 7;
 
     auto data = reinterpret_cast<uint8_t*>(zmq_msg_data(*ref));
     auto length = zmq_msg_size(*ref);

src/outgoing_msg.cc

@@ -6,7 +6,8 @@
 
 namespace zmq {
 OutgoingMsg::OutgoingMsg(Napi::Value value, Module& module) {
-    static auto constexpr zero_copy_threshold = 32;
+    static auto constexpr zero_copy_threshold = 1 << 7;
+
     auto buffer_send = [&](uint8_t* data, size_t length) {
         /* Zero-copy heuristic. There's an overhead in releasing the buffer with an
            async call to the main thread (v8 is not threadsafe), so copying small

src/poller.h

@@ -26,10 +26,6 @@ class Poller {
         int32_t err;
         auto loop = UvLoop(env);
 
-        /* Initialize uv pollers and timers, but unreference them immediately.
-           The poller will be referenced once the socket connects or binds.
-           The timers are weak references and will never prevent an exit. */
-
         poll->data = this;
         err = uv_poll_init_socket(loop, poll, fd);
         if (err != 0) return err;
@@ -77,20 +73,20 @@ class Poller {
         assert((events & UV_READABLE) == 0);
 
         if (timeout > 0) {
-            auto result = uv_timer_start(readable_timer,
+            auto err = uv_timer_start(readable_timer,
                 [](uv_timer_t* timer) {
                     auto& poller = *reinterpret_cast<Poller*>(timer->data);
                     poller.Trigger(UV_READABLE);
                 },
                 timeout, 0);
 
-            assert(result == 0);
+            assert(err == 0);
         }
 
         if (!events) {
             /* Only start polling if we were not polling already. */
-            auto result = uv_poll_start(poll, UV_READABLE, Callback);
-            assert(result == 0);
+            auto err = uv_poll_start(poll, UV_READABLE, Callback);
+            assert(err == 0);
         }
 
         events |= UV_READABLE;
@@ -100,22 +96,22 @@ class Poller {
         assert((events & UV_WRITABLE) == 0);
 
         if (timeout > 0) {
-            auto result = uv_timer_start(writable_timer,
+            auto err = uv_timer_start(writable_timer,
                 [](uv_timer_t* timer) {
                     auto& poller = *reinterpret_cast<Poller*>(timer->data);
                     poller.Trigger(UV_WRITABLE);
                 },
                 timeout, 0);
 
-            assert(result == 0);
+            assert(err == 0);
         }
 
         /* Note: We poll for READS only! "ZMQ shall signal ANY pending
            events on the socket in an edge-triggered fashion by making the
            file descriptor become ready for READING." */
         if (!events) {
-            auto result = uv_poll_start(poll, UV_READABLE, Callback);
-            assert(result == 0);
+            auto err = uv_poll_start(poll, UV_READABLE, Callback);
+            assert(err == 0);
         }
 
         events |= UV_WRITABLE;
@@ -140,19 +136,19 @@ class Poller {
     inline void Trigger(int32_t triggered) {
         events &= ~triggered;
         if (!events) {
-            auto result = uv_poll_stop(poll);
-            assert(result == 0);
+            auto err = uv_poll_stop(poll);
+            assert(err == 0);
         }
 
         if (triggered & UV_READABLE) {
-            auto result = uv_timer_stop(readable_timer);
-            assert(result == 0);
+            auto err = uv_timer_stop(readable_timer);
+            assert(err == 0);
             static_cast<T*>(this)->ReadableCallback();
         }
 
         if (triggered & UV_WRITABLE) {
-            auto result = uv_timer_stop(writable_timer);
-            assert(result == 0);
+            auto err = uv_timer_stop(writable_timer);
+            assert(err == 0);
             static_cast<T*>(this)->WritableCallback();
         }
     }

src/socket.cc

@@ -8,6 +8,7 @@
 #include "util/arguments.h"
 #include "util/async_scope.h"
 #include "util/error.h"
+#include "util/uvimmediate.h"
 #include "util/uvloop.h"
 #include "util/uvwork.h"
 
@@ -16,6 +17,10 @@
 #include <unordered_set>
 
 namespace zmq {
+/* The maximum number of sync I/O operations that are allowed before the I/O
+   methods will force the returned promise to be resolved in the next tick. */
+auto constexpr max_sync_operations = 1 << 9;
+
 /* Ordinary static cast for all available numeric types. */
 template <typename T>
 T NumberCast(const Napi::Number& num) {
@@ -524,56 +529,61 @@ Napi::Value Socket::Send(const Napi::CallbackInfo& info) {
 #endif
 
     if (send_timeout == 0 || HasEvents(ZMQ_POLLOUT)) {
-        /* We can send on the socket immediately. This is a separate code
-           path so we can avoid creating a lambda. */
-        auto res = Napi::Promise::Deferred::New(Env());
-        Send(res, parts);
-
-        /* This operation may have caused a state change, so we must update
-           the poller state manually! */
-        poller.Trigger();
-
-        return res.Promise();
-    } else {
-        /* Check if we are already polling for writes. If so that means
-           two async read operations are started; which we do not allow.
-           This is not laziness; we should not introduce additional queueing
-           because it would break ZMQ semantics. */
-        if (poller.PollingWritable()) {
-            ErrnoException(Env(), EAGAIN).ThrowAsJavaScriptException();
-            return Env().Undefined();
+        /* We can send on the socket immediately. This is a fast path. NOTE: We
+           must make sure to not keep returning synchronously resolved promises,
+           or we will starve the event loop. This can happen because ZMQ uses a
+           background I/O thread, which could mean that the Node.js process is
+           busy sending data to the I/O thread but is no longer able to respond
+           to other events. This operation may have caused a state change, so we
+           must also update the poller state manually! */
+        if (sync_operations++ < max_sync_operations) {
+            auto res = Napi::Promise::Deferred::New(Env());
+            Send(res, parts);
+            poller.Trigger();
+            return res.Promise();
         }
 
-        return poller.WritePromise(send_timeout, std::move(parts));
+        SetImmediate(Env(), [&]() { poller.Trigger(); });
     }
+
+    /* Check if we are already polling for writes. If so that means two async
+       read operations are started; which we do not allow. This is not laziness;
+       we should not introduce additional queueing because it would break ZMQ
+       semantics. */
+    if (poller.PollingWritable()) {
+        ErrnoException(Env(), EAGAIN).ThrowAsJavaScriptException();
+        return Env().Undefined();
+    }
+
+    return poller.WritePromise(send_timeout, std::move(parts));
 }
 
 Napi::Value Socket::Receive(const Napi::CallbackInfo& info) {
     if (!ValidateArguments(info, {})) return Env().Undefined();
     if (!ValidateOpen()) return Env().Undefined();
 
     if (receive_timeout == 0 || HasEvents(ZMQ_POLLIN)) {
-        /* We can read from the socket immediately. This is a separate code
-           path so we can avoid creating a lambda. */
-        auto res = Napi::Promise::Deferred::New(Env());
-        Receive(res);
-
-        /* This operation may have caused a state change, so we must update
-           the poller state manually! */
-        poller.Trigger();
-
-        return res.Promise();
-    } else {
-        /* Check if we are already polling for reads. Only one promise may
-           receive the next message, so we must ensure that receive
-           operations are in sequence. */
-        if (poller.PollingReadable()) {
-            ErrnoException(Env(), EAGAIN).ThrowAsJavaScriptException();
-            return Env().Undefined();
+        /* We can read from the socket immediately. This is a fast path.
+           Also see the related comments in Send(). */
+        if (sync_operations++ < max_sync_operations) {
+            auto res = Napi::Promise::Deferred::New(Env());
+            Receive(res);
+            poller.Trigger();
+            return res.Promise();
         }
 
-        return poller.ReadPromise(receive_timeout);
+        SetImmediate(Env(), [&]() { poller.Trigger(); });
     }
+
+    /* Check if we are already polling for reads. Only one promise may receive
+       the next message, so we must ensure that receive operations are in
+       sequence. */
+    if (poller.PollingReadable()) {
+        ErrnoException(Env(), EAGAIN).ThrowAsJavaScriptException();
+        return Env().Undefined();
+    }
+
+    return poller.ReadPromise(receive_timeout);
 }
 
 void Socket::Join(const Napi::CallbackInfo& info) {
@@ -848,11 +858,13 @@ void Socket::Initialize(Module& module, Napi::Object& exports) {
 }
 
 void Socket::Poller::ReadableCallback() {
+    socket.sync_operations = 0;
     AsyncScope scope(read_deferred.Env(), socket.async_context);
     socket.Receive(read_deferred);
 }
 
 void Socket::Poller::WritableCallback() {
+    socket.sync_operations = 0;
     AsyncScope scope(write_deferred.Env(), socket.async_context);
     socket.Send(write_deferred, write_value);
     write_value.Clear();
@@ -865,7 +877,7 @@ Napi::Value Socket::Poller::ReadPromise(int64_t timeout) {
 }
 
 Napi::Value Socket::Poller::WritePromise(int64_t timeout, OutgoingMsg::Parts&& value) {
-    write_deferred = Napi::Promise::Deferred(read_deferred.Env());
+    write_deferred = Napi::Promise::Deferred(write_deferred.Env());
     write_value = std::move(value);
     zmq::Poller<Poller>::PollWritable(timeout);
     return write_deferred.Promise();

src/socket.h

@@ -98,6 +98,7 @@ class Socket : public Napi::ObjectWrap<Socket>, public Closable {
 
     int64_t send_timeout = -1;
     int64_t receive_timeout = -1;
+    uint32_t sync_operations = 0;
     uint32_t endpoints = 0;
 
     State state = State::Open;

src/util/uvimmediate.h

@@ -0,0 +1,49 @@
+/* Copyright (c) 2017-2019 Rolf Timmermans */
+#pragma once
+
+#include "uvhandle.h"
+#include "uvloop.h"
+
+namespace zmq {
+template <typename C>
+class UvImmediate {
+    UvHandle<uv_check_t> check;
+    UvHandle<uv_idle_t> idle;
+    C delayed_callback;
+
+public:
+    UvImmediate(uv_loop_t* loop, C&& callback) : delayed_callback(std::move(callback)) {
+        int32_t err;
+
+        check->data = this;
+        err = uv_check_init(loop, check);
+        assert(err == 0);
+
+        idle->data = this;
+        err = uv_idle_init(loop, idle);
+        assert(err == 0);
+    }
+
+    inline void Schedule() {
+        int32_t err;
+
+        /* Idle handle is needed to stop the event loop from blocking in poll. */
+        err = uv_idle_start(idle, [](uv_idle_t* idle) {});
+        assert(err == 0);
+
+        err = uv_check_start(check, [](uv_check_t* check) {
+            auto& immediate = *reinterpret_cast<UvImmediate*>(check->data);
+            immediate.delayed_callback();
+            delete &immediate;
+        });
+
+        assert(err == 0);
+    }
+};
+
+template <typename C>
+static inline void SetImmediate(const Napi::Env& env, C callback) {
+    auto immediate = new UvImmediate<C>(UvLoop(env), std::move(callback));
+    return immediate->Schedule();
+}
+}

src/util/uvloop.h

@@ -2,7 +2,7 @@
 #pragma once
 
 namespace zmq {
-inline uv_loop_t* UvLoop(Napi::Env env) {
+inline uv_loop_t* UvLoop(const Napi::Env& env) {
     uv_loop_t* loop = nullptr;
     auto status = napi_get_uv_event_loop(env, &loop);
     assert(status == napi_ok);

src/util/uvwork.h

@@ -20,7 +20,7 @@ class UvWork {
         work->data = this;
     }
 
-    inline int32_t Exec(uv_loop_t* loop) {
+    inline int32_t Schedule(uv_loop_t* loop) {
         auto err = uv_queue_work(loop, work.get(),
             [](uv_work_t* req) {
                 auto& work = *reinterpret_cast<UvWork*>(req->data);
@@ -39,9 +39,9 @@ class UvWork {
 };
 
 template <typename E, typename C>
-static inline int32_t UvQueue(Napi::Env env, E execute, C complete) {
+static inline int32_t UvQueue(const Napi::Env& env, E execute, C complete) {
     auto loop = UvLoop(env);
     auto work = new UvWork<E, C>(std::move(execute), std::move(complete));
-    return work->Exec(loop);
+    return work->Schedule(loop);
 }
 }

test/bench/create-socket.js

@@ -13,7 +13,7 @@ if (zmq.cur) {
 }
 
 if (zmq.ng) {
-  zmq.ng.global.maxSockets = n
+  zmq.ng.context.maxSockets = n
   suite.add(`create socket n=${n} zmq=ng`, Object.assign({
     fn: deferred => {
       const sockets = []