diff --git a/cpp/DBHostObject.cpp b/cpp/DBHostObject.cpp index b3fbcc37..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 @@ -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, {}); }); @@ -32,7 +35,10 @@ 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; + } 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) { @@ -207,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); } @@ -242,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; } @@ -256,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( @@ -303,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 = {}; @@ -350,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 @@ -625,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); @@ -642,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")) { @@ -725,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 {}; })); @@ -768,12 +788,20 @@ 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 - // 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/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..79f228f2 100644 --- a/cpp/OPSqlite.cpp +++ b/cpp/OPSqlite.cpp @@ -12,6 +12,7 @@ #include "utils.hpp" #include #include +#include #include #include #include @@ -25,8 +26,13 @@ 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; // React native will try to clean the module on JS context invalidation // (CodePush/Hot Reload) The clearState function is called @@ -34,13 +40,28 @@ void invalidate() { // Global flag used by the threads to stop work invalidated = true; - for (const auto &db : dbs) { - db->invalidate(); + // 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); + } + + // 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, @@ -53,6 +74,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); @@ -92,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); }); @@ -146,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); }); @@ -208,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); }); diff --git a/cpp/OPThreadPool.cpp b/cpp/OPThreadPool.cpp index b06c00b3..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,30 +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); - // Notify one thread that there are requests to process - workQueueConditionVariable.notify_one(); + // 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 @@ -72,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_one(); + 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 9405681d..e7fa529c 100644 --- a/cpp/OPThreadPool.hpp +++ b/cpp/OPThreadPool.hpp @@ -1,5 +1,6 @@ #pragma once +#include #include #include #include @@ -14,31 +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 - bool done; + // tells the threads to stop looping and finish. + // 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/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; diff --git a/cpp/utils.cpp b/cpp/utils.cpp index e442625e..8537a273 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)); @@ -432,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); });