forked from nodejs/node-addon-api
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Removes the `unique_ptr` from `ThreadSafeFunction`, thereby allowing copies. Ref: nodejs#524
- Loading branch information
Showing
7 changed files
with
243 additions
and
23 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,152 @@ | ||
| #include "napi.h" | ||
| #include <thread> | ||
| #include <future> | ||
|
|
||
| #if (NAPI_VERSION > 3) | ||
|
|
||
| using namespace Napi; | ||
| using namespace std; | ||
|
|
||
| namespace { | ||
|
|
||
| struct TestData { | ||
| // Native Promise returned to JavaScript | ||
| Promise::Deferred deferred; | ||
|
|
||
| // List of threads created for test. This list only ever accessed via main | ||
| // thread. | ||
| vector<thread> threads = {}; | ||
|
|
||
| ThreadSafeFunction tsfn = ThreadSafeFunction(); | ||
| }; | ||
|
|
||
| // The no-context with finalizer + finalizer data overload is | ||
| // ambiguous. Workaround: use a lambda... | ||
|
|
||
| #define FinalizerCallback [](Napi::Env env, TestData* finalizeData){ \ | ||
| for (size_t i = 0; i < finalizeData->threads.size(); ++i) { \ | ||
| finalizeData->threads[i].join(); \ | ||
| } \ | ||
| finalizeData->deferred.Resolve(Boolean::New(env,true)); \ | ||
| delete finalizeData; \ | ||
| } | ||
|
|
||
| /** | ||
| * See threadsafe_function_sum.js for descriptions of the tests in this file | ||
| */ | ||
|
|
||
| void entryWithTSFN(ThreadSafeFunction tsfn, int threadId) { | ||
| std::this_thread::sleep_for(std::chrono::milliseconds(std::rand() % 100 + 1)); | ||
| tsfn.BlockingCall( [=](Napi::Env env, Function callback) { | ||
| callback.Call( { Number::New(env, static_cast<double>(threadId))}); | ||
| }); | ||
| tsfn.Release(); | ||
| } | ||
|
|
||
| static Value TestWithTSFN(const CallbackInfo& info) { | ||
| int threadCount = info[0].As<Number>().Int32Value(); | ||
| Function cb = info[1].As<Function>(); | ||
|
|
||
| // We pass the test data to the Finalizer for cleanup. The finalizer is | ||
| // responsible for deleting this data as well. | ||
| TestData *testData = new TestData({ | ||
| Promise::Deferred::New(info.Env()) | ||
| }); | ||
|
|
||
| ThreadSafeFunction tsfn = ThreadSafeFunction::New(info.Env(), cb, "Test", 0, threadCount, FinalizerCallback, testData); | ||
|
|
||
| for (int i = 0; i < threadCount; ++i) { | ||
| testData->threads.push_back( thread(entryWithTSFN, tsfn, i) ); | ||
| } | ||
|
|
||
| return testData->deferred.Promise(); | ||
| } | ||
|
|
||
|
|
||
| void entryDelayedTSFN(std::future<ThreadSafeFunction> tsfnFuture, int threadId) { | ||
| std::this_thread::sleep_for(std::chrono::milliseconds(std::rand() % 100 + 1)); | ||
| ThreadSafeFunction tsfn = tsfnFuture.get(); | ||
| tsfn.BlockingCall( [=](Napi::Env env, Function callback) { | ||
| callback.Call( { Number::New(env, static_cast<double>(threadId))}); | ||
| }); | ||
| tsfn.Release(); | ||
| } | ||
|
|
||
| static Value TestDelayedTSFN(const CallbackInfo& info) { | ||
| int threadCount = info[0].As<Number>().Int32Value(); | ||
| Function cb = info[1].As<Function>(); | ||
|
|
||
| TestData *testData = new TestData({ | ||
| Promise::Deferred::New(info.Env()) | ||
| }); | ||
|
|
||
| vector< std::promise<ThreadSafeFunction> > tsfnPromises; | ||
|
|
||
| for (int i = 0; i < threadCount; ++i) { | ||
| tsfnPromises.emplace_back(); | ||
| testData->threads.push_back( thread(entryDelayedTSFN, tsfnPromises[i].get_future(), i) ); | ||
| } | ||
|
|
||
| testData->tsfn = ThreadSafeFunction::New(info.Env(), cb, "Test", 0, threadCount, FinalizerCallback, testData); | ||
|
|
||
| for (int i = 0; i < threadCount; ++i) { | ||
| tsfnPromises[i].set_value(testData->tsfn); | ||
| } | ||
|
|
||
| return testData->deferred.Promise(); | ||
| } | ||
|
|
||
| void entryAcquire(ThreadSafeFunction tsfn, int threadId) { | ||
| tsfn.Acquire(); | ||
| std::this_thread::sleep_for(std::chrono::milliseconds(std::rand() % 100 + 1)); | ||
| tsfn.BlockingCall( [=](Napi::Env env, Function callback) { | ||
| callback.Call( { Number::New(env, static_cast<double>(threadId))}); | ||
| }); | ||
| tsfn.Release(); | ||
| } | ||
|
|
||
| static Value CreateThread(const CallbackInfo& info) { | ||
| TestData* testData = static_cast<TestData*>(info.Data()); | ||
| ThreadSafeFunction tsfn = testData->tsfn; | ||
| int threadId = testData->threads.size(); | ||
| testData->threads.push_back( thread(entryAcquire, tsfn, threadId) ); | ||
| return Number::New(info.Env(), threadId); | ||
| } | ||
|
|
||
| static Value StopThreads(const CallbackInfo& info) { | ||
| TestData* testData = static_cast<TestData*>(info.Data()); | ||
| ThreadSafeFunction tsfn = testData->tsfn; | ||
| tsfn.Release(); | ||
| return info.Env().Undefined(); | ||
| } | ||
|
|
||
| static Value TestAcquire(const CallbackInfo& info) { | ||
| Function cb = info[0].As<Function>(); | ||
| Napi::Env env = info.Env(); | ||
|
|
||
| // We pass the test data to the Finalizer for cleanup. The finalizer is | ||
| // responsible for deleting this data as well. | ||
| TestData *testData = new TestData({ | ||
| Promise::Deferred::New(env) | ||
| }); | ||
|
|
||
| testData->tsfn = ThreadSafeFunction::New(env, cb, "Test", 0, 1, FinalizerCallback, testData); | ||
|
|
||
| Object result = Object::New(env); | ||
| result["createThread"] = Function::New( env, CreateThread, "createThread", testData); | ||
| result["stopThreads"] = Function::New( env, StopThreads, "stopThreads", testData); | ||
| result["promise"] = testData->deferred.Promise(); | ||
|
|
||
| return result; | ||
| } | ||
| } | ||
|
|
||
| Object InitThreadSafeFunctionSum(Env env) { | ||
| Object exports = Object::New(env); | ||
| exports["testDelayedTSFN"] = Function::New(env, TestDelayedTSFN); | ||
| exports["testWithTSFN"] = Function::New(env, TestWithTSFN); | ||
| exports["testAcquire"] = Function::New(env, TestAcquire); | ||
| return exports; | ||
| } | ||
|
|
||
| #endif |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| 'use strict'; | ||
| const assert = require('assert'); | ||
| const buildType = process.config.target_defaults.default_configuration; | ||
|
|
||
| /** | ||
| * | ||
| * ThreadSafeFunction Tests: Thread Id Sums | ||
| * | ||
| * Every native C++ function that utilizes the TSFN will call the registered | ||
| * callback with the thread id. Passing Array.prototype.push with a bound array | ||
| * will push the thread id to the array. Therefore, starting `N` threads, we | ||
| * will expect the sum of all elements in the array to be `(N-1) * (N) / 2` (as | ||
| * thread IDs are 0-based) | ||
| * | ||
| * We check different methods of passing a ThreadSafeFunction around multiple | ||
| * threads: | ||
| * - `testWithTSFN`: The main thread creates the TSFN. Then, it creates | ||
| * threads, passing the TSFN at thread construction. The number of threads is | ||
| * static (known at TSFN creation). | ||
| * - `testDelayedTSFN`: The main thread creates threads, passing a promise to a | ||
| * TSFN at construction. Then, it creates the TSFN, and resolves each | ||
| * threads' promise. The number of threads is static. | ||
| * - `testAcquire`: The native binding returns a function to start a new. A | ||
| * call to this function will return `false` once `N` calls have been made. | ||
| * Each thread will acquire its own use of the TSFN, call it, and then | ||
| * release. | ||
| */ | ||
|
|
||
| const THREAD_COUNT = 5; | ||
| const EXPECTED_SUM = (THREAD_COUNT - 1) * (THREAD_COUNT) / 2; | ||
|
|
||
| module.exports = Promise.all([ | ||
| test(require(`../build/${buildType}/binding.node`)), | ||
| test(require(`../build/${buildType}/binding_noexcept.node`)) | ||
| ]); | ||
|
|
||
| /** @param {number[]} N */ | ||
| const sum = (N) => N.reduce((sum, n) => sum + n, 0); | ||
|
|
||
| function test(binding) { | ||
| async function check(bindingFunction) { | ||
| const calls = []; | ||
| const result = await bindingFunction(THREAD_COUNT, Array.prototype.push.bind(calls)); | ||
| assert.ok(result); | ||
| assert.equal(sum(calls), EXPECTED_SUM); | ||
| } | ||
|
|
||
| async function checkAcquire() { | ||
| const calls = []; | ||
| const { promise, createThread, stopThreads } = binding.threadsafe_function_sum.testAcquire(Array.prototype.push.bind(calls)); | ||
| for (let i = 0; i < THREAD_COUNT; i++) { | ||
| createThread(); | ||
| } | ||
| stopThreads(); | ||
| const result = await promise; | ||
| assert.ok(result); | ||
| assert.equal(sum(calls), EXPECTED_SUM); | ||
| } | ||
|
|
||
| return Promise.all([ | ||
| check(binding.threadsafe_function_sum.testDelayedTSFN), | ||
| check(binding.threadsafe_function_sum.testWithTSFN), | ||
| checkAcquire() | ||
| ]); | ||
| } |