diff --git a/generate/templates/manual/include/thread_pool.h b/generate/templates/manual/include/thread_pool.h index 7e6e3b605..4653e8079 100644 --- a/generate/templates/manual/include/thread_pool.h +++ b/generate/templates/manual/include/thread_pool.h @@ -8,6 +8,20 @@ #include "async_worker.h" +// Temporary workaround for LFS checkout. Comment added to be reverted. +// With the threadpool rewrite, a Worker will execute its callbacks with +// objects temporary unlock (to prevent deadlocks), and we'll wait until +// the callback is done to lock them back again (to make sure it's thread-safe). +// LFS checkout lost performance after this, and the proper way to fix it is +// to integrate nodegit-lfs into nodegit. Until this is implemented, a +// temporary workaround has been applied, which affects only Workers leveraging +// threaded libgit2 functions (at the moment only checkout) and does the +// following: +// - do not wait for the current callback to end, so that it can send the +// next callback to the main JS thread. +// - do not temporary unlock the objects, since they would be locked back +// again before the callback is executed. + namespace nodegit { class Context; class AsyncContextCleanupHandle; @@ -17,7 +31,9 @@ namespace nodegit { public: typedef std::function Callback; typedef std::function QueueCallbackFn; - typedef std::function OnPostCallbackFn; + // Temporary workaround for LFS checkout. Code modified to be reverted. + // typedef std::function OnPostCallbackFn; + typedef std::function OnPostCallbackFn; // Initializes thread pool and spins up the requested number of threads // The provided loop will be used for completion callbacks, whenever diff --git a/generate/templates/manual/src/async_baton.cc b/generate/templates/manual/src/async_baton.cc index 8078c0c45..56694f33f 100644 --- a/generate/templates/manual/src/async_baton.cc +++ b/generate/templates/manual/src/async_baton.cc @@ -46,7 +46,8 @@ namespace nodegit { ThreadPool::PostCallbackEvent( [jsCallback, cancelCallback]( ThreadPool::QueueCallbackFn queueCallback, - ThreadPool::Callback callbackCompleted + ThreadPool::Callback callbackCompleted, + bool isThreaded // Temporary workaround for LFS checkout. Code added to be reverted. ) -> ThreadPool::Callback { queueCallback(jsCallback, cancelCallback); callbackCompleted(); @@ -58,13 +59,22 @@ namespace nodegit { ThreadPool::PostCallbackEvent( [this, jsCallback, cancelCallback]( ThreadPool::QueueCallbackFn queueCallback, - ThreadPool::Callback callbackCompleted + ThreadPool::Callback callbackCompleted, + bool isThreaded // Temporary workaround for LFS checkout. Code added to be reverted. ) -> ThreadPool::Callback { - this->onCompletion = callbackCompleted; + // Temporary workaround for LFS checkout. Code modified to be reverted. + if (!isThreaded) { + this->onCompletion = callbackCompleted; - queueCallback(jsCallback, cancelCallback); + queueCallback(jsCallback, cancelCallback); - return std::bind(&AsyncBaton::SignalCompletion, this); + return std::bind(&AsyncBaton::SignalCompletion, this); + } + else { + this->onCompletion = std::bind(&AsyncBaton::SignalCompletion, this); + queueCallback(jsCallback, cancelCallback); + return []() {}; + } } ); diff --git a/generate/templates/manual/src/thread_pool.cc b/generate/templates/manual/src/thread_pool.cc index fa95ccb42..4cd5c095a 100644 --- a/generate/templates/manual/src/thread_pool.cc +++ b/generate/templates/manual/src/thread_pool.cc @@ -7,6 +7,7 @@ #include #include #include +#include // Temporary workaround for LFS checkout. Code added to be reverted. extern "C" { #include @@ -81,8 +82,11 @@ namespace nodegit { : Event(CALLBACK_TYPE), callback(initCallback) {} - ThreadPool::Callback operator()(ThreadPool::QueueCallbackFn queueCb, ThreadPool::Callback completedCb) { - return callback(queueCb, completedCb); + // Temporary workaround for LFS checkout. Code modified to be reverted. + // ThreadPool::Callback operator()(ThreadPool::QueueCallbackFn queueCb, ThreadPool::Callback completedCb) { + // return callback(queueCb, completedCb); + ThreadPool::Callback operator()(ThreadPool::QueueCallbackFn queueCb, ThreadPool::Callback completedCb, bool isThreaded) { + return callback(queueCb, completedCb, isThreaded); } private: @@ -102,6 +106,10 @@ namespace nodegit { // the Orchestrator's memory void WaitForThreadClose(); + // Temporary workaround for LFS checkout. Code added to be reverted. + // Returns true if the task running spawned threads within libgit2 + bool IsGitThreaded() { return currentGitThreads > kInitialGitThreads; } + static Nan::AsyncResource *GetCurrentAsyncResource(); static const nodegit::Context *GetCurrentContext(); @@ -139,6 +147,12 @@ namespace nodegit { PostCompletedEventToOrchestratorFn postCompletedEventToOrchestrator; TakeNextTaskFn takeNextTask; std::thread thread; + + // Temporary workaround for LFS checkout. Code added to be reverted. + static constexpr int kInitialGitThreads {0}; + // Number of threads spawned internally by libgit2 to deal with + // the task of this Executor instance. Defaults to kInitialGitThreads. + std::atomic currentGitThreads {kInitialGitThreads}; }; Executor::Executor( @@ -170,6 +184,9 @@ namespace nodegit { WorkTask *workTask = static_cast(task.get()); + // Temporary workaround for LFS checkout. Code added to be reverted. + currentGitThreads = kInitialGitThreads; + currentAsyncResource = workTask->asyncResource; currentCallbackErrorHandle = workTask->callbackErrorHandle; workTask->callback(); @@ -221,6 +238,8 @@ namespace nodegit { } void *Executor::RetrieveTLSForLibgit2ChildThread() { + // Temporary workaround for LFS checkout. Code added to be reverted. + ++Executor::executor->currentGitThreads; return Executor::executor; } @@ -230,6 +249,8 @@ namespace nodegit { void Executor::TeardownTLSOnLibgit2ChildThread() { if (!isExecutorThread) { + // Temporary workaround for LFS checkout. Code added to be reverted. + --Executor::executor->currentGitThreads; Executor::executor = nullptr; } } @@ -378,21 +399,46 @@ namespace nodegit { std::shared_ptr callbackCondition(new std::condition_variable); bool hasCompleted = false; - LockMaster::TemporaryUnlock temporaryUnlock; + // Temporary workaround for LFS checkout. Code removed to be reverted. + //LockMaster::TemporaryUnlock temporaryUnlock; + + // Temporary workaround for LFS checkout. Code added to be reverted. + bool isWorkerThreaded = executor.IsGitThreaded(); + ThreadPool::Callback callbackCompleted = []() {}; + if (!isWorkerThreaded) { + callbackCompleted = [callbackCondition, callbackMutex, &hasCompleted]() { + std::lock_guard lock(*callbackMutex); + hasCompleted = true; + callbackCondition->notify_one(); + }; + } + std::unique_ptr temporaryUnlock {nullptr}; + if (!isWorkerThreaded) { + temporaryUnlock = std::make_unique(); + } + auto onCompletedCallback = (*callbackEvent)( [this](ThreadPool::Callback callback, ThreadPool::Callback cancelCallback) { queueCallbackOnJSThread(callback, cancelCallback, false); }, + // Temporary workaround for LFS checkout. Code modified to be reverted. + /* [callbackCondition, callbackMutex, &hasCompleted]() { std::lock_guard lock(*callbackMutex); hasCompleted = true; callbackCondition->notify_one(); } + */ + callbackCompleted, + isWorkerThreaded ); - std::unique_lock lock(*callbackMutex); - while (!hasCompleted) callbackCondition->wait(lock); - onCompletedCallback(); + // Temporary workaround for LFS checkout. Code modified to be reverted. + if (!isWorkerThreaded) { + std::unique_lock lock(*callbackMutex); + while (!hasCompleted) callbackCondition->wait(lock); + onCompletedCallback(); + } } queueCallbackOnJSThread( @@ -479,10 +525,6 @@ namespace nodegit { void QueueCallbackOnJSThread(ThreadPool::Callback callback, ThreadPool::Callback cancelCallback, bool isWork); - static void RunJSThreadCallbacksFromOrchestrator(uv_async_t *handle); - - void RunJSThreadCallbacksFromOrchestrator(); - static void RunLoopCallbacks(uv_async_t *handle); void Shutdown(std::unique_ptr cleanupHandle); @@ -498,7 +540,6 @@ namespace nodegit { private: bool isMarkedForDeletion; - nodegit::Context *currentContext; struct JSThreadCallback { JSThreadCallback(ThreadPool::Callback callback, ThreadPool::Callback cancelCallback, bool isWork) @@ -539,9 +580,9 @@ namespace nodegit { std::vector orchestrators; }; + // context required to be passed to Orchestrators, but ThreadPoolImpl doesn't need to keep it ThreadPoolImpl::ThreadPoolImpl(int numberOfThreads, uv_loop_t *loop, nodegit::Context *context) : isMarkedForDeletion(false), - currentContext(context), orchestratorJobMutex(new std::mutex), jsThreadCallbackMutex(new std::mutex) { diff --git a/test/tests/filter.js b/test/tests/filter.js index 324375fce..3cb7467e8 100644 --- a/test/tests/filter.js +++ b/test/tests/filter.js @@ -333,6 +333,82 @@ describe("Filter", function() { }); }); + it("can run sync callback on checkout without deadlocking", function() { // jshint ignore:line + var test = this; + var syncCallbackResult = 1; + + return Registry.register(filterName, { + apply: function() { + syncCallbackResult = test.repository.isEmpty(); + }, + check: function() { + return NodeGit.Error.CODE.OK; + } + }, 0) + .then(function(result) { + assert.strictEqual(result, NodeGit.Error.CODE.OK); + return fse.writeFile( + packageJsonPath, + "Changing content to trigger checkout", + { encoding: "utf-8" } + ); + }) + .then(function() { + var opts = { + checkoutStrategy: Checkout.STRATEGY.FORCE, + paths: "package.json" + }; + return Checkout.head(test.repository, opts); + }) + .then(function() { + assert.strictEqual(syncCallbackResult, 0); + }); + }); + + // Temporary workaround for LFS checkout. Test skipped. + // To activate when reverting workaround. + // 'Checkout.head' and 'Submodule.lookup' do work with the repo locked. + // They should work together without deadlocking. + it.skip("can run async callback on checkout without deadlocking", function() { // jshint ignore:line + var test = this; + var submoduleNameIn = "vendor/libgit2"; + var asyncCallbackResult = ""; + + return Registry.register(filterName, { + apply: function() { + return NodeGit.Submodule.lookup(test.repository, submoduleNameIn) + .then(function(submodule) { + return submodule.name(); + }) + .then(function(name) { + asyncCallbackResult = name; + return NodeGit.Error.CODE.OK; + }); + }, + check: function() { + return NodeGit.Error.CODE.OK; + } + }, 0) + .then(function(result) { + assert.strictEqual(result, NodeGit.Error.CODE.OK); + return fse.writeFile( + packageJsonPath, + "Changing content to trigger checkout", + { encoding: "utf-8" } + ); + }) + .then(function() { + var opts = { + checkoutStrategy: Checkout.STRATEGY.FORCE, + paths: "package.json" + }; + return Checkout.head(test.repository, opts); + }) + .then(function() { + assert.equal(asyncCallbackResult, submoduleNameIn); + }); + }); + // this test is useless on 32 bit CI, because we cannot construct // a buffer big enough to test anything of significance :)... if (process.arch === "x64") { diff --git a/test/tests/submodule.js b/test/tests/submodule.js index f42758b12..2323fa56c 100644 --- a/test/tests/submodule.js +++ b/test/tests/submodule.js @@ -157,4 +157,38 @@ describe("Submodule", function() { assert.equal(entries[1].path, submodulePath); }); }); + + it("can run sync callback without deadlocking", function() { + var repo = this.workdirRepository; + var submodules = []; + var submoduleCallback = function(submodule, name, payload) { + var submoduleName = submodule.name(); + assert.equal(submoduleName, name); + submodules.push(name); + }; + + return Submodule.foreach(repo, submoduleCallback).then(function() { + assert.equal(submodules.length, 1); + }); + }); + + // 'Submodule.foreach' and 'Submodule.lookup' do work with the repo locked. + // They should work together without deadlocking. + it("can run async callback without deadlocking", function() { + var repo = this.workdirRepository; + var submodules = []; + var submoduleCallback = function(submodule, name, payload) { + var owner = submodule.owner(); + + return Submodule.lookup(owner, name) + .then(function(submodule) { + assert.equal(submodule.name(), name); + submodules.push(name); + }); + }; + + return Submodule.foreach(repo, submoduleCallback).then(function() { + assert.equal(submodules.length, 1); + }); + }); }); diff --git a/test/tests/worker.js b/test/tests/worker.js index b68b47e42..5fbfeacee 100644 --- a/test/tests/worker.js +++ b/test/tests/worker.js @@ -2,7 +2,9 @@ const path = require("path"); const assert = require("assert"); const fse = require("fs-extra"); const local = path.join.bind(path, __dirname); +const NodeGit = require("../../"); +let filterName = "psuedo_filter"; let Worker; try { @@ -24,6 +26,15 @@ if (Worker) { }); }); + afterEach(function() { + return NodeGit.FilterRegistry.unregister(filterName) + .catch(function(error) { + if (error === NodeGit.Error.CODE.ERROR) { + throw new Error("Cannot unregister filter"); + } + }); + }); + it("can perform basic functionality via worker thread", function(done) { const workerPath = local("../utils/worker.js"); const worker = new Worker(workerPath, { @@ -127,5 +138,78 @@ if (Worker) { } }); }); + + // This tests that while calling filter's apply callbacks and the worker + // is terminated, node exits gracefully. To make sure we terminate the + // worker during a checkout, continuous checkouts will be running in a loop. + it("can kill worker thread while doing a checkout and exit gracefully", function(done) { // jshint ignore:line + const workerPath = local("../utils/worker_checkout.js"); + const worker = new Worker(workerPath, { + workerData: { + clonePath, + url: "https://github.com/nodegit/test.git" + } + }); + worker.on("message", (message) => { + switch (message) { + case "init": + // give enough time for the worker to start applying the filter + // during continuous checkouts + setTimeout(() => { worker.terminate(); }, 10000); + break; + case "success": + assert.fail(); + break; + case "failure": + assert.fail(); + break; + } + }); + worker.on("error", () => assert.fail()); + worker.on("exit", (code) => { + if (code == 1) { + done(); + } else { + assert.fail(); + } + }); + }); + + // This tests that after calling filter's apply callbacks and the worker + // is terminated, there will be no memory leaks. + it("can track objects to free on context shutdown after multiple checkouts", function(done) { // jshint ignore:line + let testOk; + const workerPath = local("../utils/worker_context_aware_checkout.js"); + const worker = new Worker(workerPath, { + workerData: { + clonePath, + url: "https://github.com/nodegit/test.git" + } + }); + worker.on("message", (message) => { + switch (message) { + case "numbersMatch": + testOk = true; + worker.terminate(); + break; + case "numbersDoNotMatch": + testOk = false; + worker.terminate(); + break; + case "failure": + assert.fail(); + break; + } + }); + worker.on("error", () => assert.fail()); + worker.on("exit", (code) => { + if (code === 1 && testOk === true) { + done(); + } + else { + assert.fail(); + } + }); + }); }); } diff --git a/test/utils/loop_checkout.js b/test/utils/loop_checkout.js new file mode 100644 index 000000000..f6e6c0736 --- /dev/null +++ b/test/utils/loop_checkout.js @@ -0,0 +1,91 @@ +const fse = require("fs-extra"); +const path = require("path"); +const NodeGit = require("../../"); + +const getDirExtFiles = function(dir, ext, done) { + let results = []; + fse.readdir(dir, function(err, list) { + if (err) { + return done(err); + } + let i = 0; + (function next() { + let file = list[i++]; + if (!file) { + return done(null, results); + } + file = path.resolve(dir, file); + fse.stat(file, function(err, stat) { + if (stat && stat.isDirectory()) { + getDirExtFiles(file, ext, function(err, res) { + results = results.concat(res); + next(); + }); + } else { + if (path.extname(file) == ".".concat(ext)) { + results.push(file); + } + next(); + } + }); + })(); + }); +}; + +const getDirFilesToChange = function(dir, ext) { + return new Promise(function(resolve, reject) { + getDirExtFiles(dir, ext, function(err, results) { + if (err) { + reject(err); + } + resolve(results); + }); + }); +}; + +// Changes the content of files with extension 'ext' +// in directory 'dir' recursively. +// Returns relative file paths +const changeDirExtFiles = function (dir, ext, newText) { + let filesChanged = []; + return getDirFilesToChange(dir, ext) + .then(function(filesWithExt) { + filesWithExt.forEach(function(file) { + fse.writeFile( + file, + newText, + { encoding: "utf-8" } + ); + filesChanged.push(path.relative(dir, file)); + }); + return filesChanged; + }) + .catch(function(err) { + throw new Error("Error getting files with extension .".concat(ext)); + }); +}; + +// 'times' to limit the number of iterations in the loop. +// 0 means no limit. +const loopingCheckoutHead = async function(repoPath, repo, times) { + const text0 = "Text0: changing content to trigger checkout"; + const text1 = "Text1: changing content to trigger checkout"; + + let iteration = 0; + for (let i = 0; true; i = ++i%2) { + const newText = (i == 0) ? text0 : text1; + const jsRelativeFilePahts = await changeDirExtFiles(repoPath, "js", newText); // jshint ignore:line + let checkoutOpts = { + checkoutStrategy: NodeGit.Checkout.STRATEGY.FORCE, + paths: jsRelativeFilePahts + }; + await NodeGit.Checkout.head(repo, checkoutOpts); + + if (++iteration == times) { + break; + } + } + return; +}; + +module.exports = loopingCheckoutHead; \ No newline at end of file diff --git a/test/utils/worker_checkout.js b/test/utils/worker_checkout.js new file mode 100644 index 000000000..adfb7aa7c --- /dev/null +++ b/test/utils/worker_checkout.js @@ -0,0 +1,51 @@ +const { + isMainThread, + parentPort, + workerData +} = require("worker_threads"); +const assert = require("assert"); +const NodeGit = require("../../"); +const loopingCheckoutHead = require("./loop_checkout.js"); + +if (isMainThread) { + throw new Error("Must be run via worker thread"); +} + +parentPort.postMessage("init"); + +const { clonePath, url } = workerData; +const cloneOpts = { + fetchOpts: { + callbacks: { + certificateCheck: () => 0 + } + } +}; + +let repository; +let filterName = "psuedo_filter"; +let applyCallbackResult = 1; + +return NodeGit.Clone(url, clonePath, cloneOpts) +.then(function(_repository) { + repository = _repository; + assert.ok(repository instanceof NodeGit.Repository); + return NodeGit.FilterRegistry.register(filterName, { + apply: function() { + applyCallbackResult = 0; + }, + check: function() { + return NodeGit.Error.CODE.OK; + } + }, 0); +}) +.then(function(result) { + assert.strictEqual(result, NodeGit.Error.CODE.OK); + return loopingCheckoutHead(clonePath, repository, 0); +}).then(function() { + assert.strictEqual(applyCallbackResult, 0); + parentPort.postMessage("success"); +}) +.catch((err) => { + parentPort.postMessage("failure"); +}); \ No newline at end of file diff --git a/test/utils/worker_context_aware.js b/test/utils/worker_context_aware.js index a1a00fbc6..b7784fdae 100644 --- a/test/utils/worker_context_aware.js +++ b/test/utils/worker_context_aware.js @@ -41,7 +41,7 @@ return NodeGit.Clone(url, clonePath, opts) garbageCollect(); // Count total of objects left after being created/destroyed - const selfFreeingCount = + const freeingCount = NodeGit.Cert.getNonSelfFreeingConstructedCount() + NodeGit.Repository.getSelfFreeingInstanceCount() + NodeGit.Commit.getSelfFreeingInstanceCount() + @@ -50,7 +50,7 @@ return NodeGit.Clone(url, clonePath, opts) const numberOfTrackedObjects = NodeGit.getNumberOfTrackedObjects(); - if (selfFreeingCount === numberOfTrackedObjects) { + if (freeingCount === numberOfTrackedObjects) { parentPort.postMessage("numbersMatch"); } else { diff --git a/test/utils/worker_context_aware_checkout.js b/test/utils/worker_context_aware_checkout.js new file mode 100644 index 000000000..6562915a4 --- /dev/null +++ b/test/utils/worker_context_aware_checkout.js @@ -0,0 +1,66 @@ +const { + isMainThread, + parentPort, + workerData +} = require("worker_threads"); +const garbageCollect = require("./garbage_collect.js"); +const assert = require("assert"); +const NodeGit = require("../../"); +const loopingCheckoutHead = require("./loop_checkout.js"); +const { promisify } = require("util"); + +if (isMainThread) { + throw new Error("Must be run via worker thread"); +} + +const { clonePath, url } = workerData; +const cloneOpts = { + fetchOpts: { + callbacks: { + certificateCheck: () => 0 + } + } +}; + +let repository; +let filterName = "psuedo_filter"; +let applyCallbackResult = 1; + +return NodeGit.Clone(url, clonePath, cloneOpts) +.then(function(_repository) { + repository = _repository; + assert.ok(repository instanceof NodeGit.Repository); + return NodeGit.FilterRegistry.register(filterName, { + apply: function() { + applyCallbackResult = 0; + }, + check: function() { + return NodeGit.Error.CODE.OK; + } + }, 0); +}) +.then(function(result) { + assert.strictEqual(result, NodeGit.Error.CODE.OK); + return loopingCheckoutHead(clonePath, repository, 10); +}).then(function() { + assert.strictEqual(applyCallbackResult, 0); + // Tracked objects must work too when the Garbage Collector is triggered + garbageCollect(); + + // Count total of objects left after being created/destroyed + const freeingCount = + NodeGit.Cert.getNonSelfFreeingConstructedCount() + + NodeGit.FilterSource.getNonSelfFreeingConstructedCount() + + NodeGit.Buf.getNonSelfFreeingConstructedCount() + + NodeGit.Repository.getSelfFreeingInstanceCount(); + + const numberOfTrackedObjects = NodeGit.getNumberOfTrackedObjects(); + + if (freeingCount === numberOfTrackedObjects) { + parentPort.postMessage("numbersMatch"); + } + else { + parentPort.postMessage("numbersDoNotMatch"); + } + return promisify(setTimeout)(50000); +}).catch((err) => parentPort.postMessage("failure")); \ No newline at end of file