merge threadpool with connection state

Author: stevensJourney

.mtslconfig.json

@@ -1 +1 @@
-{ "ignore_dirs": [".git", "node_modules", "tests"] }
+{"ignore_dirs":[".git","node_modules","tests"]}

cpp/ConnectionPool.cpp

@@ -6,25 +6,24 @@
 
 ConnectionPool::ConnectionPool(std::string dbName, std::string docPath,
                                unsigned int numReadConnections)
-    : dbName(dbName), maxReads(numReadConnections) {
+    : dbName(dbName), maxReads(numReadConnections),
+      writeConnection(dbName, docPath,
+                      SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE |
+                          SQLITE_OPEN_FULLMUTEX) {
 
   onContextCallback = nullptr;
   isConcurrencyEnabled = maxReads > 0;
 
-  // Open the write connection
-  writeConnection = new ConnectionState(
-      dbName, docPath,
-      SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX);
-
+  readConnections = new ConnectionState *[maxReads];
   // Open the read connections
   for (int i = 0; i < maxReads; i++) {
-    readConnections.push_back(new ConnectionState(
-        dbName, docPath, SQLITE_OPEN_READONLY | SQLITE_OPEN_FULLMUTEX));
+    readConnections[i] = new ConnectionState(
+        dbName, docPath, SQLITE_OPEN_READONLY | SQLITE_OPEN_FULLMUTEX);
   }
 
   if (true == isConcurrencyEnabled) {
     // Write connection WAL setup
-    writeConnection->queueWork([](sqlite3 *db) {
+    writeConnection.queueWork([](sqlite3 *db) {
       sqliteExecuteLiteralWithDB(db, "PRAGMA journal_mode = WAL;");
       sqliteExecuteLiteralWithDB(
           db,
@@ -44,10 +43,10 @@ ConnectionPool::ConnectionPool(std::string dbName, std::string docPath,
 };
 
 ConnectionPool::~ConnectionPool() {
-  delete writeConnection;
-  for (auto con : readConnections) {
-    delete con;
+  for (int i = 0; i < maxReads; i++) {
+    delete readConnections[i];
   }
+  delete readConnections;
 }
 
 void ConnectionPool::readLock(ConnectionLockId contextId) {
@@ -65,7 +64,7 @@ void ConnectionPool::readLock(ConnectionLockId contextId) {
     for (int i = 0; i < maxReads; i++) {
       if (readConnections[i]->isEmptyLock()) {
         // There is an open slot
-        activateContext(readConnections[i], contextId);
+        activateContext(*readConnections[i], contextId);
         return;
       }
     }
@@ -77,7 +76,7 @@ void ConnectionPool::readLock(ConnectionLockId contextId) {
 
 void ConnectionPool::writeLock(ConnectionLockId contextId) {
   // Check if there are any available read connections
-  if (writeConnection->isEmptyLock()) {
+  if (writeConnection.isEmptyLock()) {
     activateContext(writeConnection, contextId);
     return;
   }
@@ -90,8 +89,8 @@ SQLiteOPResult
 ConnectionPool::queueInContext(ConnectionLockId contextId,
                                std::function<void(sqlite3 *)> task) {
   ConnectionState *state = nullptr;
-  if (writeConnection->matchesLock(contextId)) {
-    state = writeConnection;
+  if (writeConnection.matchesLock(contextId)) {
+    state = &writeConnection;
   } else {
     // Check if it's a read connection
     for (int i = 0; i < maxReads; i++) {
@@ -124,27 +123,27 @@ void ConnectionPool::setOnContextAvailable(void (*callback)(std::string,
 void ConnectionPool::setTableUpdateHandler(
     void (*callback)(void *, int, const char *, const char *, sqlite3_int64)) {
   // Only the write connection can make changes
-  sqlite3_update_hook(writeConnection->connection, callback,
+  sqlite3_update_hook(writeConnection.connection, callback,
                       (void *)(dbName.c_str()));
 }
 
 void ConnectionPool::closeContext(ConnectionLockId contextId) {
-  if (writeConnection->matchesLock(contextId)) {
+  if (writeConnection.matchesLock(contextId)) {
     if (writeQueue.size() > 0) {
       // There are items in the queue, activate the next one
       activateContext(writeConnection, writeQueue[0]);
       writeQueue.erase(writeQueue.begin());
     } else {
       // No items in the queue, clear the context
-      writeConnection->clearLock();
+      writeConnection.clearLock();
     }
   } else {
     // Check if it's a read connection
     for (int i = 0; i < maxReads; i++) {
       if (readConnections[i]->matchesLock(contextId)) {
         if (readQueue.size() > 0) {
           // There are items in the queue, activate the next one
-          activateContext(readConnections[i], readQueue[0]);
+          activateContext(*readConnections[i], readQueue[0]);
           readQueue.erase(readQueue.begin());
         } else {
           // No items in the queue, clear the context
@@ -157,7 +156,7 @@ void ConnectionPool::closeContext(ConnectionLockId contextId) {
 }
 
 void ConnectionPool::closeAll() {
-  writeConnection->close();
+  writeConnection.close();
   for (int i = 0; i < maxReads; i++) {
     readConnections[i]->close();
   }
@@ -177,7 +176,7 @@ SQLiteOPResult ConnectionPool::attachDatabase(std::string const dbFileName,
   auto dbConnections = getAllConnections();
 
   for (auto &connectionState : dbConnections) {
-    if (connectionState->isEmptyLock()) {
+    if (!connectionState->isEmptyLock()) {
       return SQLiteOPResult{
           .type = SQLiteError,
           .errorMessage = dbName + " was unable to attach another database: " +
@@ -214,7 +213,7 @@ SQLiteOPResult ConnectionPool::detachDatabase(std::string const alias) {
   auto dbConnections = getAllConnections();
 
   for (auto &connectionState : dbConnections) {
-    if (connectionState->isEmptyLock()) {
+    if (!connectionState->isEmptyLock()) {
       return SQLiteOPResult{
           .type = SQLiteError,
           .errorMessage = dbName + " was unable to detach another database: " +
@@ -243,16 +242,16 @@ SQLiteOPResult ConnectionPool::detachDatabase(std::string const alias) {
 
 std::vector<ConnectionState *> ConnectionPool::getAllConnections() {
   std::vector<ConnectionState *> result;
-  result.push_back(writeConnection);
+  result.push_back(&writeConnection);
   for (int i = 0; i < maxReads; i++) {
     result.push_back(readConnections[i]);
   }
   return result;
 }
 
-void ConnectionPool::activateContext(ConnectionState *state,
+void ConnectionPool::activateContext(ConnectionState &state,
                                      ConnectionLockId contextId) {
-  state->activateLock(contextId);
+  state.activateLock(contextId);
 
   if (onContextCallback != nullptr) {
     onContextCallback(dbName, contextId);

cpp/ConnectionPool.h

@@ -51,8 +51,8 @@ class ConnectionPool {
 private:
   int maxReads;
   std::string dbName;
-  std::vector<ConnectionState *> readConnections;
-  ConnectionState *writeConnection;
+  ConnectionState **readConnections;
+  ConnectionState writeConnection;
 
   std::vector<ConnectionLockId> readQueue;
   std::vector<ConnectionLockId> writeQueue;
@@ -118,7 +118,7 @@ class ConnectionPool {
 private:
   std::vector<ConnectionState *> getAllConnections();
 
-  void activateContext(ConnectionState *state, ConnectionLockId contextId);
+  void activateContext(ConnectionState &state, ConnectionLockId contextId);
 
   SQLiteOPResult genericSqliteOpenDb(string const dbName, string const docPath,
                                      sqlite3 **db, int sqlOpenFlags);

cpp/ConnectionState.cpp

@@ -29,9 +29,14 @@ ConnectionState::~ConnectionState() {
   delete thread;
 }
 
-void ConnectionState::clearLock() { _currentLockId = EMPTY_LOCK_ID; }
+void ConnectionState::clearLock() {
+  if (!workQueue.empty()) {
+    waitFinished();
+  }
+  _currentLockId = EMPTY_LOCK_ID;
+}
 
-SQLiteOPResult ConnectionState::activateLock(const ConnectionLockId &lockId) {
+void ConnectionState::activateLock(const ConnectionLockId &lockId) {
   _currentLockId = lockId;
 }
 
@@ -42,7 +47,11 @@ bool ConnectionState::matchesLock(const ConnectionLockId &lockId) {
 bool ConnectionState::isEmptyLock() { return _currentLockId == EMPTY_LOCK_ID; }
 
 void ConnectionState::close() {
-  waitFinished();
+  if (!workQueue.empty()) {
+    waitFinished();
+  }
+  // So that the thread can stop (if not already)
+  threadDone = true;
   sqlite3_close_v2(connection);
 }
 
@@ -84,6 +93,7 @@ void ConnectionState::doWork() {
     task(connection);
     --threadBusy;
   }
+  workQueueConditionVariable.notify_all();
 }
 
 void ConnectionState::waitFinished() {

cpp/ConnectionState.h

@@ -37,7 +37,7 @@ class ConnectionState {
   ~ConnectionState();
 
   void clearLock();
-  SQLiteOPResult activateLock(const ConnectionLockId &lockId);
+  void activateLock(const ConnectionLockId &lockId);
   bool matchesLock(const ConnectionLockId &lockId);
   bool isEmptyLock();
 

cpp/bindings.cpp

@@ -471,7 +471,7 @@ void osp::install(jsi::Runtime &rt,
   module.setProperty(rt, "open", move(open));
   module.setProperty(rt, "requestLock", move(requestLock));
   module.setProperty(rt, "releaseLock", move(releaseLock));
-  module.setProperty(rt, "executeInContextAsync", move(executeInContext));
+  module.setProperty(rt, "executeInContext", move(executeInContext));
   module.setProperty(rt, "close", move(close));
 
   module.setProperty(rt, "attach", move(attach));

cpp/sqliteBridge.cpp

@@ -65,8 +65,8 @@ SQLiteOPResult sqliteCloseDb(string const dbName) {
   ConnectionPool *connection = dbMap[dbName];
 
   connection->closeAll();
-  delete connection;
   dbMap.erase(dbName);
+  delete connection;
 
   return SQLiteOPResult{
       .type = SQLiteOk,