test: add test for effect of UV_THREADPOOL_SIZE (#77)

packages/test/CMakeLists.txt

@@ -264,6 +264,7 @@ if((NOT IS_WASM) OR IS_EMSCRIPTEN OR IS_WASI_THREADS)
  endif()
  add_test("tsfn" "./tsfn/binding.c" OFF ON "")
  add_test("async_cleanup_hook" "./async_cleanup_hook/binding.c" OFF ON "")
+ add_test("uv_threadpool_size" "./uv_threadpool_size/binding.c" OFF ON "")
 endif()
 
 add_test("arg" "./arg/binding.c" ON OFF "")

packages/test/script/test.js

@@ -18,7 +18,8 @@ const pthread = [
  'pool/**/*',
  'tsfn/**/*',
  'async_cleanup_hook/**/*',
- 'string/string-pthread.test.js'
+ 'string/string-pthread.test.js',
+ 'uv_threadpool_size/**/*'
 ]
 
 if (process.env.EMNAPI_TEST_NATIVE) {

packages/test/uv_threadpool_size/binding.c

@@ -0,0 +1,190 @@
+#undef NDEBUG
+#include <assert.h>
+#include <node_api.h>
+#include <stdlib.h>
+#include <uv.h>
+#include <stdio.h>
+#include "../common.h"
+
+typedef struct {
+ uv_mutex_t mutex;
+ uint32_t threadpool_size;
+ uint32_t n_tasks_started;
+ uint32_t n_tasks_exited;
+ uint32_t n_tasks_finalized;
+ bool observed_saturation;
+} async_shared_data;
+
+typedef struct {
+ uint32_t task_id;
+ async_shared_data* shared_data;
+ napi_async_work request;
+} async_carrier;
+
+static inline bool all_tasks_started(async_shared_data* d) {
+ assert(d->n_tasks_started <= d->threadpool_size + 1);
+ return d->n_tasks_started == d->threadpool_size + 1;
+}
+
+static inline bool all_tasks_exited(async_shared_data* d) {
+ assert(d->n_tasks_exited <= d->n_tasks_started);
+ return all_tasks_started(d) && d->n_tasks_exited == d->n_tasks_started;
+}
+
+static inline bool all_tasks_finalized(async_shared_data* d) {
+ assert(d->n_tasks_finalized <= d->n_tasks_exited);
+ return all_tasks_exited(d) && d->n_tasks_finalized == d->n_tasks_exited;
+}
+
+static inline bool still_saturating(async_shared_data* d) {
+ return d->n_tasks_started < d->threadpool_size;
+}
+
+static inline bool threadpool_saturated(async_shared_data* d) {
+ return d->n_tasks_started == d->threadpool_size && d->n_tasks_exited == 0;
+}
+
+static inline bool threadpool_desaturating(async_shared_data* d) {
+ return d->n_tasks_started >= d->threadpool_size && d->n_tasks_exited != 0;
+}
+
+static inline void print_info(const char* label, async_carrier* c) {
+ async_shared_data* d = c->shared_data;
+ printf("%s task_id=%u n_tasks_started=%u n_tasks_exited=%u "
+ "n_tasks_finalized=%u observed_saturation=%d\n",
+ label,
+ c->task_id,
+ d->n_tasks_started,
+ d->n_tasks_exited,
+ d->n_tasks_finalized,
+ d->observed_saturation);
+}
+
+static void Execute(napi_env env, void* data) {
+ async_carrier* c = (async_carrier*)data;
+ async_shared_data* d = c->shared_data;
+
+ // As long as fewer than threadpool_size async tasks have been started, more
+ // should be started (eventually). Only once that happens should scheduled
+ // async tasks remain queued.
+ uv_mutex_lock(&d->mutex);
+ bool should_be_concurrent = still_saturating(d);
+ d->n_tasks_started++;
+ assert(d->n_tasks_started <= d->threadpool_size + 1);
+
+ print_info("start", c);
+
+ if (should_be_concurrent) {
+ // Wait for the thread pool to be saturated. This is not an elegant way of
+ // doing so, but it really does not matter much here.
+ while (still_saturating(d)) {
+ print_info("waiting", c);
+ uv_mutex_unlock(&d->mutex);
+ uv_sleep(100);
+ uv_mutex_lock(&d->mutex);
+ }
+
+ // One async task will observe that the threadpool is saturated, that is,
+ // that threadpool_size tasks have been started and none have exited yet.
+ // That task will be the first to exit.
+ if (!d->observed_saturation) {
+ assert(threadpool_saturated(d));
+ d->observed_saturation = true;
+ } else {
+ assert(threadpool_saturated(d) || threadpool_desaturating(d));
+ }
+ } else {
+ // If this task is not among the first threadpool_size tasks, it should not
+ // have been started unless other tasks have already finished.
+ assert(threadpool_desaturating(d));
+ }
+
+ print_info("exit", c);
+
+ // Allow other tasks to access the shared data. If the thread pool is actually
+ // larger than threadpool_size, this allows an extraneous task to start, which
+ // will lead to an assertion error.
+ uv_mutex_unlock(&d->mutex);
+ uv_sleep(1000);
+ uv_mutex_lock(&d->mutex);
+
+ d->n_tasks_exited++;
+ uv_mutex_unlock(&d->mutex);
+}
+
+static void Complete(napi_env env, napi_status status, void* data) {
+ async_carrier* c = (async_carrier*)data;
+ async_shared_data* d = c->shared_data;
+
+ if (status != napi_ok) {
+ napi_throw_type_error(env, NULL, "Execute callback failed.");
+ return;
+ }
+
+ uv_mutex_lock(&d->mutex);
+ assert(threadpool_desaturating(d));
+ d->n_tasks_finalized++;
+ print_info("finalize", c);
+ if (all_tasks_finalized(d)) {
+ uv_mutex_unlock(&d->mutex);
+ uv_mutex_destroy(&d->mutex);
+ free(d);
+ } else {
+ uv_mutex_unlock(&d->mutex);
+ }
+
+ NAPI_CALL_RETURN_VOID(env, napi_delete_async_work(env, c->request));
+ free(c);
+}
+
+static napi_value Test(napi_env env, napi_callback_info info) {
+ size_t argc = 1;
+ napi_value argv[1];
+ napi_value this;
+ void* data;
+ NAPI_CALL(env, napi_get_cb_info(env, info, &argc, argv, &this, &data));
+ NAPI_ASSERT(env, argc >= 1, "Not enough arguments, expected 1.");
+
+ async_shared_data* shared_data = calloc(1, sizeof(async_shared_data));
+ assert(shared_data != NULL);
+ int ret = uv_mutex_init(&shared_data->mutex);
+ assert(ret == 0);
+
+ napi_valuetype t;
+ NAPI_CALL(env, napi_typeof(env, argv[0], &t));
+ NAPI_ASSERT(
+ env, t == napi_number, "Wrong first argument, integer expected.");
+ NAPI_CALL(
+ env, napi_get_value_uint32(env, argv[0], &shared_data->threadpool_size));
+
+ napi_value resource_name;
+ NAPI_CALL(env,
+ napi_create_string_utf8(
+ env, "TestResource", NAPI_AUTO_LENGTH, &resource_name));
+
+ for (uint32_t i = 0; i <= shared_data->threadpool_size; i++) {
+ async_carrier* carrier = malloc(sizeof(async_carrier));
+ assert(carrier != NULL);
+ carrier->task_id = i;
+ carrier->shared_data = shared_data;
+ NAPI_CALL(env,
+ napi_create_async_work(env,
+ NULL,
+ resource_name,
+ Execute,
+ Complete,
+ carrier,
+ &carrier->request));
+ NAPI_CALL(env, napi_queue_async_work(env, carrier->request));
+ }
+
+ return NULL;
+}
+
+static napi_value Init(napi_env env, napi_value exports) {
+ napi_property_descriptor desc = DECLARE_NAPI_PROPERTY("test", Test);
+ NAPI_CALL(env, napi_define_properties(env, exports, 1, &desc));
+ return exports;
+}
+
+NAPI_MODULE(NODE_GYP_MODULE_NAME, Init)

packages/test/uv_threadpool_size/uv_threadpool_size.test.js

@@ -0,0 +1,8 @@
+'use strict'
+const { load } = require('../util')
+
+module.exports = load('uv_threadpool_size').then(({ test }) => {
+ const uvThreadpoolSize = parseInt(process.env.EXPECTED_UV_THREADPOOL_SIZE ||
+ process.env.UV_THREADPOOL_SIZE, 10) || 4
+ test(uvThreadpoolSize)
+})