Getting queue stats from libuv with N-API (Node.js v10)

binding.gyp

{
  "targets": [
    {
      "target_name": "queuestats",
      "sources": ["napi-queue-stats.cc"]
    }
  ]
}

index.js

const fs = require('fs');
const dns = require('dns');
const zlib = require('zlib');
const https = require('https');

const queueStats = require('./build/Release/queuestats').queueStats;

function currentDiffInMs(startTime) {
  const diff = process.hrtime(startTime);
  const ms = diff[0] * 1e3 + diff[1] / 1e6;
  return ms;
}

function getPosts(i, callback) {
  const req = https.request({
    hostname: 'jsonplaceholder.typicode.com',
    path: '/posts',
    headers: {
      'Accept-Encoding': 'gzip, deflate',
    },
  }, (res) => {
    console.log('response arrives', i, queueStats());
    const unzip = zlib.createUnzip();
    const decompressed = res.pipe(unzip);
    const chunks = [];
    decompressed.on('data', (chunk) => {
      console.log('response data', i, queueStats());
      chunks.push(chunk);
    });
    decompressed.on('end', () => {
      console.log('response end', i, queueStats());
      const body = Buffer.concat(chunks).toString('utf-8');
      const posts = JSON.parse(body);
      callback(null, posts);
    });
  });
  req.on('socket', (socket) => {
    console.log('socket', i, queueStats());
    socket.once('lookup', () => {
      console.log('lookup', i, queueStats());
    });
  });
  req.on('error', (err) => {
    callback(err);
  });
  req.end();
}

const start = process.hrtime();

Array.from(Array(10)).forEach((_, i) => {
  getPosts(i, (err, posts) => {
    if (err) {
      console.error('failed to get posts', err);
    } else {
      console.log('done', i, queueStats());
    }
  });
  /*
  dns.lookup('shuheikagawa.com', (err, address) => {
    const ms = currentDiffInMs(start);
    console.log('Lookup done', i, `${ms}ms`, queueStats());
  });
  */
  console.log('Started', i, queueStats());
});
setInterval(() => {
  console.log('timer', queueStats());
}, 10).unref();

Makefile

.PHONY: build

configure:
	@node-gyp configure

build:
	@node-gyp build

run:
	@node index.js

clean:
	@node-gyp clean

napi-queue-stats.cc

#include <node_api.h>
#include <uv.h>

#define ARRAY_SIZE(a) (sizeof(a) / sizeof((a)[0]))
#define CHECK_NAPI_RESULT(condition) (assert((condition) == napi_ok))
#define CHECK_NAPI_RESULT_RETURN_NULL(condition) do { if((condition) != napi_ok) { return nullptr; } } while(0)

// Queue macros
typedef void *QUEUE[2];
#define QUEUE_NEXT(q)       (*(QUEUE **) &((*(q))[0]))
#define QUEUE_PREV(q)       (*(QUEUE **) &((*(q))[1]))
#define QUEUE_PREV_NEXT(q)  (QUEUE_NEXT(QUEUE_PREV(q)))
#define QUEUE_NEXT_PREV(q)  (QUEUE_PREV(QUEUE_NEXT(q)))
#define QUEUE_DATA(ptr, type, field)                                          \
  ((type *) ((char *) (ptr) - offsetof(type, field)))
#define QUEUE_FOREACH(q, h)                                                   \
  for ((q) = QUEUE_NEXT(h); (q) != (h); (q) = QUEUE_NEXT(q))
#define QUEUE_EMPTY(q)                                                        \
  ((const QUEUE *) (q) == (const QUEUE *) QUEUE_NEXT(q))

namespace demodemo {
  napi_value Hello(napi_env env, napi_callback_info args) {
    napi_value greeting;

    CHECK_NAPI_RESULT_RETURN_NULL(napi_create_string_utf8(env, "hello", NAPI_AUTO_LENGTH, &greeting));
    return greeting;
  }

  napi_value QueueStats(napi_env env, napi_callback_info args) {
    // https://github.com/libuv/libuv/blob/v1.x/include/uv.h#L1555-L1568
    uv_loop_t* loop = nullptr;

    CHECK_NAPI_RESULT_RETURN_NULL(napi_get_uv_event_loop(env, &loop));

    // O(n)
    napi_value wq_size;
    int n = 0;
    QUEUE* q;
    QUEUE_FOREACH(q, &loop->wq) {
      n++;
    }
    CHECK_NAPI_RESULT_RETURN_NULL(napi_create_int32(env, n, &wq_size));

    // O(n)
    napi_value handle_queue_size;
    int m = 0;
    QUEUE* qq;
    QUEUE_FOREACH(qq, &loop->handle_queue) {
      m++;
    }
    CHECK_NAPI_RESULT_RETURN_NULL(napi_create_int32(env, m, &handle_queue_size));

    napi_value active_reqs_count;
    // This also includes async requests?
    // At least not the count of tasks in running threads.
    unsigned int count = loop->active_reqs.count;
    CHECK_NAPI_RESULT_RETURN_NULL(napi_create_int32(env, count, &active_reqs_count));

    napi_value result;
    CHECK_NAPI_RESULT_RETURN_NULL(napi_create_object(env, &result));

    CHECK_NAPI_RESULT_RETURN_NULL(napi_set_named_property(env, result, "wq_size", wq_size));
    CHECK_NAPI_RESULT_RETURN_NULL(napi_set_named_property(env, result, "handle_queue_size", handle_queue_size));
    CHECK_NAPI_RESULT_RETURN_NULL(napi_set_named_property(env, result, "active_reqs_count", active_reqs_count));

    return result;
  }

  napi_value Init(napi_env env, napi_value exports) {
    napi_value fn;

    CHECK_NAPI_RESULT_RETURN_NULL(napi_create_function(env, nullptr, 0, Hello, nullptr, &fn));
    CHECK_NAPI_RESULT_RETURN_NULL(napi_set_named_property(env, exports, "hello", fn));

    CHECK_NAPI_RESULT_RETURN_NULL(napi_create_function(env, nullptr, 0, QueueStats, nullptr, &fn));
    CHECK_NAPI_RESULT_RETURN_NULL(napi_set_named_property(env, exports, "queueStats", fn));

    return exports;
  }

  NAPI_MODULE(NODE_GYP_MODULE_NAME, Init)
}