Warn when messages are queued for more than 500ms on exit.

Author: rolftimmermans

CHANGELOG.md

@@ -6,6 +6,8 @@
 
 * Compatibility mode will now provide an implementation for bindSync()/unbindSync() if the 'deasync' package is available.
 
+* Produce a warning when messages are still queued at process exit and context termination takes more than 500ms.
+
 * Bump version requirement to Node.js 10.2, but reduce N-API version to 3 to support more 10.x Node.js versions.
 
 ### v6.0.0-beta.5

src/module.h

@@ -12,6 +12,11 @@
 #include "util/to_string.h"
 #include "util/trash.h"
 
+#include <chrono>
+#include <condition_variable>
+#include <cstdio>
+#include <thread>
+
 namespace zmq {
 class Context;
 class Socket;
@@ -20,8 +25,34 @@ struct Terminator {
     constexpr Terminator() noexcept = default;
     void operator()(void* context) {
         assert(context != nullptr);
-        auto err = zmq_ctx_term(context);
-        assert(err == 0);
+
+#ifdef ZMQ_BLOCKY
+        int32_t blocky = zmq_ctx_get(context, ZMQ_BLOCKY);
+#else
+        /* If the option cannot be set, don't suggest to set it. */
+        bool blocky = false;
+#endif
+
+        using namespace std::chrono_literals;
+        std::mutex mut;
+        std::condition_variable cv;
+
+        std::thread thread([&] {
+            auto err = zmq_ctx_term(context);
+            assert(err == 0);
+            cv.notify_all();
+        });
+
+        std::unique_lock<std::mutex> lock(mut);
+        if (cv.wait_for(lock, 500ms) == std::cv_status::timeout) {
+            fprintf(stderr,
+                "(node:%d) WARNING: Waiting for queued ZeroMQ messages to be "
+                "delivered.%s\n",
+                uv_os_getpid(),
+                blocky ? " Set 'context.blocky = false' to change this behaviour." : "");
+        }
+
+        thread.join();
     }
 };
 

test/unit/context-process-exit-test.ts

@@ -1,3 +1,4 @@
+import * as semver from "semver"
 import * as zmq from "../../src"
 
 import {assert} from "chai"
@@ -7,7 +8,7 @@ describe("context process exit", function() {
   describe("with default context", function() {
     it("should occur when sockets are closed", async function() {
       this.slow(200)
-      const code = await createProcess(() => {
+      const {code} = await createProcess(() => {
         const socket1 = new zmq.Dealer()
         socket1.close()
         const socket2 = new zmq.Router()
@@ -19,7 +20,7 @@ describe("context process exit", function() {
 
     it("should occur when sockets are not closed", async function() {
       this.slow(200)
-      const code = await createProcess(() => {
+      const {code} = await createProcess(() => {
         const socket1 = new zmq.Dealer()
         const socket2 = new zmq.Router()
       })
@@ -29,20 +30,69 @@ describe("context process exit", function() {
 
     it("should not occur when sockets are open and polling", async function() {
       this.slow(1000)
-      const code = await createProcess(() => {
+      const {code} = await createProcess(() => {
         const socket1 = new zmq.Dealer()
         socket1.connect("inproc://foo")
         socket1.receive()
       })
 
       assert.equal(code, -1)
     })
+
+    it("should produce warning when messages are queued with blocky", async function() {
+      this.slow(1000)
+      const {stderr} = await createProcess(() => {
+        zmq.context.blocky = true
+        const socket1 = new zmq.Dealer({linger: 600})
+        socket1.connect("tcp://127.0.0.1:6789")
+        socket1.send(null)
+      })
+
+      if (semver.satisfies(zmq.version, ">= 4.2")) {
+        assert.match(
+          stderr.toString(),
+          /\(node:\d+\) WARNING: Waiting for queued ZeroMQ messages to be delivered\. Set 'context\.blocky = false' to change this behaviour\.\n/,
+        )
+      } else {
+        assert.match(
+          stderr.toString(),
+          /\(node:\d+\) WARNING: Waiting for queued ZeroMQ messages to be delivered\.\n/,
+        )
+      }
+    })
+
+    it("should produce warning when messages are queued without blocky", async function() {
+      this.slow(1000)
+      const {stderr} = await createProcess(() => {
+        zmq.context.blocky = false
+        const socket1 = new zmq.Dealer({linger: 600})
+        socket1.connect("tcp://127.0.0.1:6789")
+        socket1.send(null)
+      })
+
+      assert.match(
+        stderr.toString(),
+        /\(node:\d+\) WARNING: Waiting for queued ZeroMQ messages to be delivered\.\n/,
+      )
+    })
+
+    it("should not produce warning when messages are queued for a short time", async function() {
+      this.slow(1000)
+      const {stderr} = await createProcess(() => {
+        zmq.context.blocky = true
+        const socket1 = new zmq.Dealer({linger: 100})
+        socket1.connect("tcp://127.0.0.1:6789")
+        socket1.send(null)
+      })
+
+      assert.equal(stderr.toString(), "")
+    })
   })
 
   describe("with custom context", function() {
     it("should occur when sockets are closed", async function() {
       this.slow(200)
-      const code = await createProcess(() => {
+      const {code} = await createProcess(() => {
         const context = new zmq.Context()
         const socket1 = new zmq.Dealer({context})
         socket1.close()
@@ -55,7 +105,7 @@ describe("context process exit", function() {
 
     it("should occur when sockets are closed and context is gced", async function() {
       this.slow(200)
-      const code = await createProcess(() => {
+      const {code} = await createProcess(() => {
         function run() {
           const context = new zmq.Context()
           const socket1 = new zmq.Dealer({context})
@@ -73,7 +123,7 @@ describe("context process exit", function() {
 
     it("should occur when sockets are not closed", async function() {
       this.slow(200)
-      const code = await createProcess(() => {
+      const {code} = await createProcess(() => {
         const context = new zmq.Context()
         const socket1 = new zmq.Dealer({context})
         const socket2 = new zmq.Router({context})
@@ -84,7 +134,7 @@ describe("context process exit", function() {
 
     it("should not occur when sockets are open and polling", async function() {
       this.slow(1000)
-      const code = await createProcess(() => {
+      const {code} = await createProcess(() => {
         const context = new zmq.Context()
         const socket1 = new zmq.Dealer({context})
         socket1.connect("inproc://foo")

test/unit/helpers.ts

@@ -85,7 +85,13 @@ export async function createWorker<T, D extends {}>(
   })
 }
 
-export function createProcess(fn: () => void): Promise<number> {
+interface Result {
+  code: number
+  stdout: Buffer
+  stderr: Buffer
+}
+
+export function createProcess(fn: () => void): Promise<Result> {
   const src = `
     const zmq = require(${JSON.stringify(path.resolve(__dirname, "../.."))})
     const fn = ${fn.toString()}
@@ -96,20 +102,26 @@ export function createProcess(fn: () => void): Promise<number> {
   child.stdin.write(src)
   child.stdin.end()
 
-  child.stdout.on("data", (data: Buffer) => console.log(data.toString()))
-  child.stderr.on("data", (data: Buffer) => console.error(data.toString()))
+  let stdout: Buffer = Buffer.alloc(0)
+  let stderr: Buffer = Buffer.alloc(0)
+  child.stdout.on("data", (data: Buffer) => {
+    stdout = Buffer.concat([stdout, data])
+  })
+  child.stderr.on("data", (data: Buffer) => {
+    stderr = Buffer.concat([stderr, data])
+  })
 
   return new Promise((resolve, reject) => {
     child.on("close", (code: number, signal: string) => {
       if (signal != null) {
         reject(new Error(`Child exited with ${signal}`))
       } else {
-        resolve(code)
+        resolve({code, stdout, stderr})
       }
     })
 
     setTimeout(() => {
-      resolve(-1)
+      resolve({code: -1, stdout, stderr})
       child.kill()
     }, 750)
   })

test/unit/socket-process-exit-test.ts

@@ -7,7 +7,7 @@ describe("socket process exit", function() {
   /* Reported: https://github.com/nodejs/node-addon-api/issues/591 */
   it.skip("should occur cleanly when sending in exit hook", async function() {
     this.slow(200)
-    const code = await createProcess(async () => {
+    const {code} = await createProcess(async () => {
       const sockA = new zmq.Pair()
       const sockB = new zmq.Pair()
       await sockA.bind("inproc://test-1")
@@ -25,7 +25,7 @@ describe("socket process exit", function() {
 
   it("should occur cleanly when sending on unbound socket", async function() {
     this.slow(200)
-    const code = await createProcess(async () => {
+    const {code} = await createProcess(async () => {
       const sock = new zmq.Publisher()
       await sock.send("test")
     })
@@ -35,7 +35,7 @@ describe("socket process exit", function() {
 
   it("should not occur when sending and blocked on unbound socket", async function() {
     this.slow(1000)
-    const code = await createProcess(async () => {
+    const {code} = await createProcess(async () => {
       const sock = new zmq.Dealer()
       await sock.send("test")
     })
@@ -45,7 +45,7 @@ describe("socket process exit", function() {
 
   it("should occur cleanly on socket close when reading events", async function() {
     this.slow(200)
-    const code = await createProcess(() => {
+    const {code} = await createProcess(() => {
       const sock = new zmq.Dealer()
 
       async function readEvents() {
@@ -64,7 +64,7 @@ describe("socket process exit", function() {
 
   it("should not occur while reading events", async function() {
     this.slow(1000)
-    const code = await createProcess(async () => {
+    const {code} = await createProcess(async () => {
       const sock = new zmq.Dealer()
 
       const events = []