From aa4133bc16350f32bfc529dbf88ca86643b5a305 Mon Sep 17 00:00:00 2001 From: Babs Craig Date: Thu, 13 Aug 2026 00:36:38 +0100 Subject: [PATCH 1/8] Bind hooks and reactive flush to their runtime generation During a bridgeless reload two JS runtime generations overlap: the outgoing one is invalidated while the incoming one is already installing. The process globals invoker and invalidated cannot describe that, so a database created by the outgoing generation would post its hook and reactive-query callbacks into the runtime that replaced it, and then call asFunction() on a jsi::Value owned by the dead one. Each generation now gets its own liveness flag, and DBHostObject binds both the invoker and that flag at construction, on the JS thread. The members shadow the globals inside member functions, so every invokeAsync site is fixed without touching each call site. --- cpp/DBHostObject.cpp | 19 +++++++++++++++++++ cpp/DBHostObject.hpp | 12 ++++++++++++ cpp/OPSqlite.cpp | 9 +++++++++ cpp/types.hpp | 13 +++++++++++++ 4 files changed, 53 insertions(+) diff --git a/cpp/DBHostObject.cpp b/cpp/DBHostObject.cpp index b3fbcc37..285ea9d1 100644 --- a/cpp/DBHostObject.cpp +++ b/cpp/DBHostObject.cpp @@ -20,6 +20,9 @@ namespace react = facebook::react; #ifdef OP_SQLITE_USE_LIBSQL void DBHostObject::flush_pending_reactive_queries( const std::shared_ptr &resolve) { + if (alive != nullptr && !alive->load()) { + return; + } invoker->invokeAsync([resolve](jsi::Runtime &rt) { resolve->asObject(rt).asFunction(rt).call(rt, {}); }); @@ -33,6 +36,9 @@ std::string turso_remote_db_name(const std::string &url) { void DBHostObject::flush_pending_reactive_queries( const std::shared_ptr &resolve) { + if (alive != nullptr && !alive->load()) { + return; + } invoker->invokeAsync([resolve](jsi::Runtime &rt) { resolve->asObject(rt).asFunction(rt).call(rt, {}); }); @@ -40,6 +46,9 @@ void DBHostObject::flush_pending_reactive_queries( #else void DBHostObject::flush_pending_reactive_queries( const std::shared_ptr &resolve) { + if (alive != nullptr && !alive->load()) { + return; + } for (const auto &query_ptr : pending_reactive_queries) { auto query = query_ptr.get(); @@ -67,12 +76,18 @@ void DBHostObject::flush_pending_reactive_queries( } void DBHostObject::on_commit() { + if (alive != nullptr && !alive->load()) { + return; + } invoker->invokeAsync([this](jsi::Runtime &rt) { commit_hook_callback->asObject(rt).asFunction(rt).call(rt); }); } void DBHostObject::on_rollback() { + if (alive != nullptr && !alive->load()) { + return; + } invoker->invokeAsync([this](jsi::Runtime &rt) { rollback_hook_callback->asObject(rt).asFunction(rt).call(rt); }); @@ -80,6 +95,10 @@ void DBHostObject::on_rollback() { void DBHostObject::on_update(const std::string &table, const std::string &operation, long long row_id) { + if (alive != nullptr && !alive->load()) { + return; + } + if (update_hook_callback != nullptr) { invoker->invokeAsync([callback = update_hook_callback, table, operation, row_id](jsi::Runtime &rt) { diff --git a/cpp/DBHostObject.hpp b/cpp/DBHostObject.hpp index 19b1180d..81ba8281 100644 --- a/cpp/DBHostObject.hpp +++ b/cpp/DBHostObject.hpp @@ -87,6 +87,18 @@ class JSI_EXPORT DBHostObject : public jsi::HostObject { std::unordered_map function_map; std::string base_path; + // Bound at construction, on the JS thread, to the generation that created + // this database. + // + // NOTE: these deliberately shadow the process-global opsqlite::invoker and + // opsqlite::generation_alive inside every member function, which is what + // fixes the update/commit/rollback hooks and flush_pending_reactive_queries + // without touching each call site. Reading the globals at callback time + // instead lets a database belonging to a torn-down runtime post work into the + // runtime that replaced it, and then call asFunction() on a jsi::Value owned + // by the dead one. + std::shared_ptr invoker = opsqlite::invoker; + std::shared_ptr> alive = opsqlite::generation_alive; std::shared_ptr thread_pool; std::string db_name; std::string delete_db_name; diff --git a/cpp/OPSqlite.cpp b/cpp/OPSqlite.cpp index 31401f2a..2e2a52e8 100644 --- a/cpp/OPSqlite.cpp +++ b/cpp/OPSqlite.cpp @@ -27,6 +27,7 @@ std::string _sqlite_vec_path; std::vector> dbs; bool invalidated = false; std::shared_ptr invoker; +std::shared_ptr> generation_alive; // React native will try to clean the module on JS context invalidation // (CodePush/Hot Reload) The clearState function is called @@ -34,6 +35,13 @@ void invalidate() { // Global flag used by the threads to stop work invalidated = true; + // Mark THIS generation dead. Work queued by it holds a copy of the flag, so + // it drops its completions instead of resolving into a runtime that is being + // torn down. + if (generation_alive != nullptr) { + generation_alive->store(false); + } + for (const auto &db : dbs) { db->invalidate(); } @@ -53,6 +61,7 @@ void install(jsi::Runtime &rt, _sqlite_vec_path = std::string(sqlite_vec_path); opsqlite::invoker = _invoker; opsqlite::invalidated = false; + opsqlite::generation_alive = std::make_shared>(true); auto open = HFN0 { jsi::Object options = args[0].asObject(rt); diff --git a/cpp/types.hpp b/cpp/types.hpp index 6c2ff2df..5844e615 100644 --- a/cpp/types.hpp +++ b/cpp/types.hpp @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include #include @@ -12,6 +13,18 @@ namespace opsqlite { extern std::shared_ptr invoker; extern bool invalidated; +// Liveness of the current JS runtime generation. Replaced by install() and +// cleared by invalidate(), so each generation gets its own flag rather than +// sharing the process-global `invalidated` bool. +// +// Whoever queues work copies the shared_ptr when the work is created, so it +// always observes ITS OWN generation's liveness. Checking a process-global +// instead is wrong in both directions during a bridgeless reload, where two +// generations overlap: an outgoing generation clearing it would suppress the +// incoming generation's callbacks, and an incoming generation setting it would +// re-enable the outgoing generation's. +extern std::shared_ptr> generation_alive; + struct ArrayBuffer { std::shared_ptr data; size_t size; From 4b097c1f14c45c3eb9f49267e77fef4b73b9c3c0 Mon Sep 17 00:00:00 2001 From: Babs Craig Date: Thu, 13 Aug 2026 00:37:15 +0100 Subject: [PATCH 2/8] Bind promisify tasks to their runtime generation Same problem as the hooks, on the thread pool. A task read opsqlite::invoker when it completed rather than when it was queued, so a query started by the outgoing runtime could resolve into the incoming one and call asFunction() on a jsi::Value that belongs to the dead runtime. The invoker and the liveness flag are now captured on the JS thread while the promise is constructed, and the early return checks this generation rather than the process-global invalidated, which is set by whichever generation tore down last. Note the early return still drops the last references on the pool thread, so ~jsi::Value runs there. That is pre-existing, and the sqlite3_interrupt change that follows makes it very hard to reach, but it is not airtight without either leaking the values or handing them to the JS thread. --- cpp/utils.cpp | 38 +++++++++++++++++++++++++++++--------- 1 file changed, 29 insertions(+), 9 deletions(-) diff --git a/cpp/utils.cpp b/cpp/utils.cpp index e442625e..0bf6d2d9 100644 --- a/cpp/utils.cpp +++ b/cpp/utils.cpp @@ -383,18 +383,34 @@ promisify(jsi::Runtime &rt, std::shared_ptr thread_pool, auto resolve = std::make_shared(rt, args[0]); auto reject = std::make_shared(rt, args[1]); + // Bind this generation's invoker and liveness flag here, on the JS thread, + // while the promise is being constructed. Reading the process globals from + // the worker instead lets a task queued by a torn-down runtime post into the + // runtime that replaced it, and then call asFunction() on a jsi::Value that + // belongs to the dead one. + auto invoker = opsqlite::invoker; + auto alive = opsqlite::generation_alive; + auto task = [lambda = lambda, resolve_callback = resolve_callback, - resolve = std::move(resolve), reject = std::move(reject)]() { + resolve = std::move(resolve), reject = std::move(reject), + invoker, alive]() { + if (invoker == nullptr) { + return; + } + try { std::any result = lambda(); - if (opsqlite::invalidated) { + // This generation is gone. Posting now would schedule onto a runtime + // that is being torn down, where asFunction() sees an already + // invalidated PointerValue. + if (alive != nullptr && !alive->load()) { return; } // reject is also captured in the invokeAsync lambda // so it can be safely disposed on the JS thread - opsqlite::invoker->invokeAsync( + invoker->invokeAsync( [result = std::move(result), resolve = resolve, reject = reject, resolve_callback = resolve_callback](jsi::Runtime &rt) { auto jsi_result = resolve_callback(rt, result); @@ -409,9 +425,11 @@ promisify(jsi::Runtime &rt, std::shared_ptr thread_pool, // resolve is also captured in the invokeAsync lambda // so it can be safely disposed on the JS thread auto what = e.what(); - opsqlite::invoker->invokeAsync([what = std::string(what), - resolve = resolve, - reject = reject](jsi::Runtime &rt) { + if (alive != nullptr && !alive->load()) { + return; + } + invoker->invokeAsync([what = std::string(what), resolve = resolve, + reject = reject](jsi::Runtime &rt) { auto errorCtr = rt.global().getPropertyAsFunction(rt, "Error"); auto error = errorCtr.callAsConstructor( rt, jsi::String::createFromAscii(rt, what)); @@ -419,11 +437,13 @@ promisify(jsi::Runtime &rt, std::shared_ptr thread_pool, }); } catch (std::exception &exc) { auto what = exc.what(); + if (alive != nullptr && !alive->load()) { + return; + } // resolve is also captured in the invokeAsync lambda // so it can be safely disposed on the JS thread - opsqlite::invoker->invokeAsync([what = std::string(what), - resolve = resolve, - reject = reject](jsi::Runtime &rt) { + invoker->invokeAsync([what = std::string(what), resolve = resolve, + reject = reject](jsi::Runtime &rt) { auto errorCtr = rt.global().getPropertyAsFunction(rt, "Error"); auto error = errorCtr.callAsConstructor( rt, jsi::String::createFromAscii(rt, what)); From 0ec65b218208737e825fb5d4d782c063cd5cb589 Mon Sep 17 00:00:00 2001 From: Babs Craig Date: Thu, 13 Aug 2026 00:37:28 +0100 Subject: [PATCH 3/8] Interrupt in-flight queries before draining on invalidate invalidate() waits on the thread pool but, unlike close() and delete(), never asks SQLite to stop. A query inside sqlite3_step therefore runs to completion while the JS runtime is being torn down. React Native gives module invalidation ten seconds before it destroys the runtime regardless, so an uninterrupted drain can lose that race and leave a worker touching state the runtime owned. This is the same three lines close() already runs, moved to the one teardown path that lacked them. --- cpp/DBHostObject.cpp | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/cpp/DBHostObject.cpp b/cpp/DBHostObject.cpp index 285ea9d1..c2b2f631 100644 --- a/cpp/DBHostObject.cpp +++ b/cpp/DBHostObject.cpp @@ -787,6 +787,18 @@ void DBHostObject::invalidate() { } invalidated = true; + + // Abort whatever is currently inside sqlite3_step so the drain below can + // actually finish. Parity with the close and delete host functions, which + // already do this. Without it a long running query holds the pool past React + // Native's module invalidation budget, after which the runtime is destroyed + // anyway and the drain has bought nothing. +#if !defined(OP_SQLITE_USE_LIBSQL) && !defined(OP_SQLITE_USE_TURSO) + if (db != nullptr) { + sqlite3_interrupt(db); + } +#endif + // Drain in-flight thread pool work before closing the db handle. // restartPool() joins threads (waiting for the current task) but then // needlessly re-creates the pool. waitFinished() is sufficient: it From bf01e6c38bbe2a8885cdfb412e8ddbf317f15e4e Mon Sep 17 00:00:00 2001 From: Babs Craig Date: Thu, 13 Aug 2026 00:38:00 +0100 Subject: [PATCH 4/8] Guard the database registry with a mutex open() runs on the JS thread and invalidate() runs on the TurboModule queue. During a bridgeless reload those belong to different generations and overlap, because the new instance is constructed before the old one finishes invalidating, so both can reach the dbs vector at once. Iterating it while another thread emplaces can walk a reallocated buffer. invalidate() now swaps the vector out under the lock and works on its own copy, which also keeps the lock off the invalidate() calls themselves. --- cpp/OPSqlite.cpp | 38 ++++++++++++++++++++++++++++++-------- 1 file changed, 30 insertions(+), 8 deletions(-) diff --git a/cpp/OPSqlite.cpp b/cpp/OPSqlite.cpp index 2e2a52e8..79f228f2 100644 --- a/cpp/OPSqlite.cpp +++ b/cpp/OPSqlite.cpp @@ -12,6 +12,7 @@ #include "utils.hpp" #include #include +#include #include #include #include @@ -25,6 +26,10 @@ std::string _base_path; std::string _crsqlite_path; std::string _sqlite_vec_path; std::vector> dbs; +// Guards `dbs`. Two JS runtime generations overlap during a bridgeless reload, +// so open() and invalidate() can touch this vector from different threads at +// the same time. +std::mutex dbs_mutex; bool invalidated = false; std::shared_ptr invoker; std::shared_ptr> generation_alive; @@ -42,13 +47,21 @@ void invalidate() { generation_alive->store(false); } - for (const auto &db : dbs) { - db->invalidate(); + // Take ownership of the registry under the lock before touching it. This runs + // on the outgoing generation's TurboModule queue, while the incoming + // generation's open() may already be emplacing into `dbs` on its own JS + // thread: RCTHost constructs the new RCTInstance without waiting for the old + // one to finish invalidating. Iterating the vector directly can therefore run + // off a reallocated buffer. + std::vector> closing; + { + std::lock_guard g(dbs_mutex); + closing.swap(dbs); } - // Clear our existing vector of shared pointers so they can be garbage - // collected - dbs.clear(); + for (const auto &db : closing) { + db->invalidate(); + } } void install(jsi::Runtime &rt, @@ -101,7 +114,10 @@ void install(jsi::Runtime &rt, std::shared_ptr db = std::make_shared( rt, path, name, path, readOnly, failOnCreate, encryption_key); - dbs.emplace_back(db); + { + std::lock_guard g(dbs_mutex); + dbs.emplace_back(db); + } return jsi::Object::createFromHostObject(rt, db); }); @@ -155,7 +171,10 @@ void install(jsi::Runtime &rt, std::make_shared(rt, url, auth_token, path); #endif - dbs.emplace_back(db); + { + std::lock_guard g(dbs_mutex); + dbs.emplace_back(db); + } return jsi::Object::createFromHostObject(rt, db); }); @@ -217,7 +236,10 @@ void install(jsi::Runtime &rt, rt, name, path, url, auth_token, remote_encryption_key); #endif - dbs.emplace_back(db); + { + std::lock_guard g(dbs_mutex); + dbs.emplace_back(db); + } return jsi::Object::createFromHostObject(rt, db); }); From db5343d2d61247de297cde04493c0bf8f4a9df5b Mon Sep 17 00:00:00 2001 From: Babs Craig Date: Thu, 13 Aug 2026 00:38:20 +0100 Subject: [PATCH 5/8] Wake all thread pool waiters and make done atomic waitFinished() and doWork() wait on the same condition variable, so notify_one() can wake a worker when the waiter needed the wakeup, or the other way round, and leave the other one asleep until the next event. With a single pool thread that is easy to hit at teardown, where waitFinished() is the only waiter that matters. done is read in the doWork() loop condition outside the mutex while the destructor writes it, which is a data race; make it atomic. --- cpp/OPThreadPool.cpp | 8 +++++--- cpp/OPThreadPool.hpp | 6 ++++-- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/cpp/OPThreadPool.cpp b/cpp/OPThreadPool.cpp index b06c00b3..3e56c64d 100644 --- a/cpp/OPThreadPool.cpp +++ b/cpp/OPThreadPool.cpp @@ -48,8 +48,10 @@ void ThreadPool::queueWork(const std::function &task) { // Push the request to the queue workQueue.push(task); - // Notify one thread that there are requests to process - workQueueConditionVariable.notify_one(); + // Wake every waiter. waitFinished() and doWork() share this condition + // variable, so notify_one() can hand the wakeup to the wrong one and leave + // the other asleep. + workQueueConditionVariable.notify_all(); } // Function used by the threads to grab work from the queue @@ -85,7 +87,7 @@ void ThreadPool::doWork() { std::lock_guard g(workQueueMutex); --busy; } - workQueueConditionVariable.notify_one(); + workQueueConditionVariable.notify_all(); } } diff --git a/cpp/OPThreadPool.hpp b/cpp/OPThreadPool.hpp index 9405681d..6cadfea3 100644 --- a/cpp/OPThreadPool.hpp +++ b/cpp/OPThreadPool.hpp @@ -1,5 +1,6 @@ #pragma once +#include #include #include #include @@ -34,8 +35,9 @@ class ThreadPool { std::queue> workQueue; // This will be set to true when the thread pool is shutting down. This - // tells the threads to stop looping and finish - bool done; + // tells the threads to stop looping and finish. + // Atomic because doWork() reads it in `while (!done)` outside the mutex. + std::atomic done; // Function used by the threads to grab work from the queue void doWork(); From d9225b5098cbb77ce5dd3cefe2714ccff52e7ba6 Mon Sep 17 00:00:00 2001 From: Babs Craig Date: Thu, 13 Aug 2026 00:38:56 +0100 Subject: [PATCH 6/8] Never let a pool thread join itself If the last shared_ptr to a ThreadPool is released on one of its own workers, the destructor tries to join the running thread and std::thread::join throws "Resource deadlock avoided", which aborts. Dropping the pool capture from the promisify task removes the known way to get there, so this is a backstop for any future owner that ends up released on a worker. --- cpp/OPThreadPool.cpp | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/cpp/OPThreadPool.cpp b/cpp/OPThreadPool.cpp index 3e56c64d..d44960c1 100644 --- a/cpp/OPThreadPool.cpp +++ b/cpp/OPThreadPool.cpp @@ -31,6 +31,16 @@ ThreadPool::~ThreadPool() { workQueueConditionVariable.notify_all(); for (auto &thread : threads) { + // Never join ourselves. If the pool's last owner is released on one of its + // own workers, join() throws std::system_error ("thread::join failed: + // Resource deadlock avoided") and takes the process with it. Not capturing + // the pool in the promisify task should make this unreachable; this is a + // backstop, not the fix. + if (thread.get_id() == std::this_thread::get_id()) { + thread.detach(); + continue; + } + if (thread.joinable()) { thread.join(); } From b59cd8f2cd7f9fb7082cb5d9d5b68ee8c7092002 Mon Sep 17 00:00:00 2001 From: Oscar Franco Date: Thu, 13 Aug 2026 07:48:54 -0400 Subject: [PATCH 7/8] Undo no longer needed thread join check --- cpp/OPThreadPool.cpp | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/cpp/OPThreadPool.cpp b/cpp/OPThreadPool.cpp index d44960c1..3e56c64d 100644 --- a/cpp/OPThreadPool.cpp +++ b/cpp/OPThreadPool.cpp @@ -31,16 +31,6 @@ ThreadPool::~ThreadPool() { workQueueConditionVariable.notify_all(); for (auto &thread : threads) { - // Never join ourselves. If the pool's last owner is released on one of its - // own workers, join() throws std::system_error ("thread::join failed: - // Resource deadlock avoided") and takes the process with it. Not capturing - // the pool in the promisify task should make this unreachable; this is a - // backstop, not the fix. - if (thread.get_id() == std::this_thread::get_id()) { - thread.detach(); - continue; - } - if (thread.joinable()) { thread.join(); } From 0905b885dd2e613aa2f06e57928ac7d7066d1a46 Mon Sep 17 00:00:00 2001 From: Oscar Franco Date: Thu, 13 Aug 2026 08:21:17 -0400 Subject: [PATCH 8/8] Clean up ThreadPool --- cpp/DBHostObject.cpp | 39 ++++++++---------- cpp/OPThreadPool.cpp | 98 +++++++++++++++++--------------------------- cpp/OPThreadPool.hpp | 17 ++++---- cpp/utils.cpp | 2 +- 4 files changed, 65 insertions(+), 91 deletions(-) diff --git a/cpp/DBHostObject.cpp b/cpp/DBHostObject.cpp index c2b2f631..9cbc460f 100644 --- a/cpp/DBHostObject.cpp +++ b/cpp/DBHostObject.cpp @@ -6,9 +6,9 @@ #include "bridge.hpp" #endif #include "logs.h" -#include #include "macros.hpp" #include "utils.hpp" +#include #include #include @@ -35,7 +35,7 @@ std::string turso_remote_db_name(const std::string &url) { } void DBHostObject::flush_pending_reactive_queries( - const std::shared_ptr &resolve) { + const std::shared_ptr &resolve) { if (alive != nullptr && !alive->load()) { return; } @@ -226,8 +226,8 @@ DBHostObject::DBHostObject(jsi::Runtime &rt, std::string &db_name, thread_pool = std::make_shared(); - db = opsqlite_open_sync(db_name, path, url, auth_token, - remote_encryption_key); + db = + opsqlite_open_sync(db_name, path, url, auth_token, remote_encryption_key); create_jsi_functions(rt); } @@ -261,12 +261,13 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) { auto obj_params = args[0].asObject(rt); std::string secondary_db_name = - obj_params.getProperty(rt, "secondaryDbFileName").asString(rt).utf8(rt); - std::string alias = obj_params.getProperty(rt, "alias").asString(rt).utf8(rt); + obj_params.getProperty(rt, "secondaryDbFileName").asString(rt).utf8(rt); + std::string alias = + obj_params.getProperty(rt, "alias").asString(rt).utf8(rt); if (obj_params.hasProperty(rt, "location")) { std::string location = - obj_params.getProperty(rt, "location").asString(rt).utf8(rt); + obj_params.getProperty(rt, "location").asString(rt).utf8(rt); secondary_db_path = secondary_db_path + location; } @@ -275,8 +276,8 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) { // SQLite and Turso bind with explicit lengths. Failing loudly across // all backends keeps behaviour consistent. if (secondary_db_name.find('\0') != std::string::npos) { - throw std::runtime_error( - "[op-sqlite] attach secondaryDbFileName must not contain a zero byte"); + throw std::runtime_error("[op-sqlite] attach secondaryDbFileName must " + "not contain a zero byte"); } if (alias.find('\0') != std::string::npos) { throw std::runtime_error( @@ -322,7 +323,7 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) { // Drain any in-flight async queries before closing the db handle. // Without this, a queued/running execute() on the thread pool may // dereference the freed sqlite3* pointer → heap corruption / SIGABRT. - thread_pool->waitFinished(); + thread_pool->wait_finished(); #ifdef OP_SQLITE_USE_LIBSQL opsqlite_libsql_close(db); db = {}; @@ -369,11 +370,11 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) { #endif // Drain any in-flight async queries before closing/removing the db handle. // Without this, queued/running work may dereference a freed sqlite handle. - thread_pool->waitFinished(); + thread_pool->wait_finished(); if (delete_db_name.empty()) { - throw std::runtime_error( - "[op-sqlite][delete] delete() is not supported for remote-only databases"); + throw std::runtime_error("[op-sqlite][delete] delete() is not supported " + "for remote-only databases"); } #ifdef OP_SQLITE_USE_LIBSQL @@ -644,7 +645,7 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) { auto query = args[0].asObject(rt); const std::string query_str = - query.getProperty(rt, "query").asString(rt).utf8(rt); + query.getProperty(rt, "query").asString(rt).utf8(rt); auto js_args = query.getProperty(rt, "arguments"); auto js_discriminators = query.getProperty(rt, "fireOn").asObject(rt).asArray(rt); @@ -661,7 +662,7 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) { for (size_t i = 0; i < js_discriminators.length(rt); i++) { auto js_discriminator = js_discriminators.getValueAtIndex(rt, i).asObject(rt); - std::string table = + std::string table = js_discriminator.getProperty(rt, "table").asString(rt).utf8(rt); std::vector ids; if (js_discriminator.hasProperty(rt, "ids")) { @@ -744,7 +745,7 @@ void DBHostObject::create_jsi_functions(jsi::Runtime &rt) { flush_pending_reactive_queries(resolve); }; - thread_pool->queueWork(task); + thread_pool->queue_work(task); return {}; })); @@ -800,11 +801,7 @@ void DBHostObject::invalidate() { #endif // Drain in-flight thread pool work before closing the db handle. - // restartPool() joins threads (waiting for the current task) but then - // needlessly re-creates the pool. waitFinished() is sufficient: it - // blocks until the queue is empty and no worker is busy, then the - // ThreadPool destructor (via shared_ptr release) joins the threads. - thread_pool->waitFinished(); + thread_pool->wait_finished(); #ifdef OP_SQLITE_USE_LIBSQL opsqlite_libsql_close(db); diff --git a/cpp/OPThreadPool.cpp b/cpp/OPThreadPool.cpp index 3e56c64d..6b7fd24a 100644 --- a/cpp/OPThreadPool.cpp +++ b/cpp/OPThreadPool.cpp @@ -6,29 +6,36 @@ ThreadPool::ThreadPool() : done(false) { // This returns the number of threads supported by the system. If the // function can't figure out this information, it returns 0. 0 is not good, // so we create at least 1 - // auto numberOfThreads = std::thread::hardware_concurrency(); - // if (numberOfThreads == 0) { - // numberOfThreads = 1; + // auto number_of_threads = std::thread::hardware_concurrency(); + // if (number_of_threads == 0) { + // number_of_threads = 1; // } - auto numberOfThreads = 1; - for (unsigned i = 0; i < numberOfThreads; ++i) { - // The threads will execute the private member `doWork`. Note that we + // To keep sqlite threading model working, we cannot access via multiple + // threads in the same process therefore we cap the number of off-threads + // to one per connection + // + // You can achieve more performance when using more threads but you run + // into race conditions. This request was brought forth by PowerSync. + auto number_of_threads = 1; + for (unsigned i = 0; i < number_of_threads; ++i) { + // The threads will execute the private member `do_work`. Note that we // need to pass a reference to the function (namespaced with the class // name) as the first argument, and the current object as second // argument - threads.emplace_back(&ThreadPool::doWork, this); + threads.emplace_back(&ThreadPool::do_work, this); } } -// The destructor joins all the threads so the program can exit gracefully. -// This will be executed if there is any exception (e.g. creating the threads) +// The destructor joins all the threads so the process (or in our RN case, this +// main-thread/runtime generation) can exit gracefully. This will be executed if +// there is any exception (e.g. creating the threads) ThreadPool::~ThreadPool() { // So threads know it's time to shut down done = true; // Wake up all the threads, so they can finish and be joined - workQueueConditionVariable.notify_all(); + work_pending.notify_all(); for (auto &thread : threads) { if (thread.joinable()) { @@ -41,32 +48,31 @@ ThreadPool::~ThreadPool() { // This function will be called by the server every time there is a request // that needs to be processed by the thread pool -void ThreadPool::queueWork(const std::function &task) { +void ThreadPool::queue_work(const std::function &task) { // Grab the mutex - std::lock_guard g(workQueueMutex); + std::lock_guard g(work_queue_mutex); // Push the request to the queue - workQueue.push(task); + work_queue.push(task); - // Wake every waiter. waitFinished() and doWork() share this condition - // variable, so notify_one() can hand the wakeup to the wrong one and leave - // the other asleep. - workQueueConditionVariable.notify_all(); + // Wake every waiter. wait_finished() and do_work() share this condition + // so we force all conditions to check + work_pending.notify_all(); } // Function used by the threads to grab work from the queue -void ThreadPool::doWork() { +void ThreadPool::do_work() { // Loop while the queue is not destructing while (!done) { std::function task; // Create a scope, so we don't lock the queue for longer than necessary { - std::unique_lock g(workQueueMutex); - workQueueConditionVariable.wait(g, [&] { + std::unique_lock g(work_queue_mutex); + work_pending.wait(g, [&] { // Only wake up if there are elements in the queue or the // program is shutting down - return !workQueue.empty() || done; + return done || !work_queue.empty(); }); // If we are shutting down exit without trying to process more work @@ -74,57 +80,29 @@ void ThreadPool::doWork() { break; } - task = workQueue.front(); - workQueue.pop(); + task = work_queue.front(); + work_queue.pop(); ++busy; } + task(); + // Release the task (and everything it captured, e.g. JSI values) before - // signalling idle, so waitFinished()/close() can't observe busy == 0 + // signalling idle, so wait_finished()/close() can't observe busy == 0 // while task-owned resources are still pending destruction. task = nullptr; + { - std::lock_guard g(workQueueMutex); + std::lock_guard g(work_queue_mutex); --busy; } - workQueueConditionVariable.notify_all(); + work_pending.notify_all(); } } -void ThreadPool::waitFinished() { - std::unique_lock g(workQueueMutex); - workQueueConditionVariable.wait( - g, [&] { return workQueue.empty() && (busy == 0); }); +void ThreadPool::wait_finished() { + std::unique_lock g(work_queue_mutex); + work_pending.wait(g, [&] { return work_queue.empty() && (busy == 0); }); } -void ThreadPool::restartPool() { - // So threads know it's time to shut down - done = true; - - // Wake up all the threads, so they can finish and be joined - workQueueConditionVariable.notify_all(); - - for (auto &thread : threads) { - if (thread.joinable()) { - thread.join(); - } - } - - threads.clear(); - - auto numberOfThreads = std::thread::hardware_concurrency(); - if (numberOfThreads == 0) { - numberOfThreads = 1; - } - - for (unsigned i = 0; i < numberOfThreads; ++i) { - // The threads will execute the private member `doWork`. Note that we - // need to pass a reference to the function (namespaced with the class - // name) as the first argument, and the current object as second - // argument - threads.emplace_back(&ThreadPool::doWork, this); - } - - done = false; -} } // namespace opsqlite diff --git a/cpp/OPThreadPool.hpp b/cpp/OPThreadPool.hpp index 6cadfea3..e7fa529c 100644 --- a/cpp/OPThreadPool.hpp +++ b/cpp/OPThreadPool.hpp @@ -15,32 +15,31 @@ class ThreadPool { public: ThreadPool(); ~ThreadPool(); - void queueWork(const std::function &task); - void waitFinished(); - void restartPool(); + void queue_work(const std::function &task); + void wait_finished(); private: unsigned int busy{}; // This condition variable is used for the threads to wait until there is // work to do - std::condition_variable_any workQueueConditionVariable; + std::condition_variable_any work_pending; // We store the threads in a vector, so we can later stop them gracefully std::vector threads; - // Mutex to protect workQueue - std::mutex workQueueMutex; + // Mutex to protect work_queue + std::mutex work_queue_mutex; // Queue of requests waiting to be processed - std::queue> workQueue; + std::queue> work_queue; // This will be set to true when the thread pool is shutting down. This // tells the threads to stop looping and finish. - // Atomic because doWork() reads it in `while (!done)` outside the mutex. + // Atomic because do_work() reads it in `while (!done)` outside the mutex. std::atomic done; // Function used by the threads to grab work from the queue - void doWork(); + void do_work(); }; } // namespace opsqlite \ No newline at end of file diff --git a/cpp/utils.cpp b/cpp/utils.cpp index 0bf6d2d9..8537a273 100644 --- a/cpp/utils.cpp +++ b/cpp/utils.cpp @@ -452,7 +452,7 @@ promisify(jsi::Runtime &rt, std::shared_ptr thread_pool, } }; - thread_pool->queueWork(task); + thread_pool->queue_work(task); return jsi::Value(nullptr); });