wip

Author: davidmartos96

packages/sqlite_async/lib/src/native/database/connection_pool.dart

@@ -241,6 +241,55 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
     connections.addAll(_allReadConnections);
     return connections;
   }
+
+  Future<T> withAllConnections<T>(
+      Future<T> Function(
+              SqliteWriteContext writer, List<SqliteReadContext> readers)
+          block) async {
+    final blockCompleter = Completer<T>();
+    final (write, reads) = await _lockAllConns<T>(blockCompleter);
+
+    try {
+      final res = await block(write, reads);
+      blockCompleter.complete(res);
+      return res;
+    } catch (e, st) {
+      blockCompleter.completeError(e, st);
+      rethrow;
+    }
+  }
+
+  /// Locks all connections, returning the acquired contexts.
+  /// We pass a completer that would be called after the locks are taken.
+  Future<(SqliteWriteContext, List<SqliteReadContext>)> _lockAllConns<T>(
+      Completer<T> lockCompleter) async {
+    final List<Completer<SqliteReadContext>> readLockedCompleters = [];
+    final Completer<SqliteWriteContext> writeLockedCompleter = Completer();
+
+    // Take the write lock
+    writeLock((ctx) {
+      writeLockedCompleter.complete(ctx);
+      return lockCompleter.future;
+    });
+
+    // Take all the read locks
+    for (final readConn in _allReadConnections) {
+      final completer = Completer<SqliteReadContext>();
+      readLockedCompleters.add(completer);
+
+      readConn.readLock((ctx) {
+        completer.complete(ctx);
+        return lockCompleter.future;
+      });
+    }
+
+    // Wait after all locks are taken
+    final contexts = await Future.wait([
+      writeLockedCompleter.future,
+      ...readLockedCompleters.map((e) => e.future)
+    ]);
+    return (contexts.first as SqliteWriteContext, contexts.sublist(1));
+  }
 }
 
 typedef ReadCallback<T> = Future<T> Function(SqliteReadContext tx);

packages/sqlite_async/test/native/basic_test.dart

@@ -69,7 +69,35 @@ void main() {
       }
     });
 
-    test('Concurrency 2', () async {
+    test('with all connections', () async {
+      final db = SqliteDatabase.withFactory(
+          await testUtils.testFactory(path: path),
+          maxReaders: 3);
+      await db.initialize();
+      await createTables(db);
+
+      print("${DateTime.now()} start");
+
+      final withAllConnsFut = () async {
+        await Future.delayed(const Duration(milliseconds: 20));
+        await db.withAllConnections((writer, readers) async {
+          print("${DateTime.now()} in withAllConnections");
+          await Future.delayed(const Duration(seconds: 5));
+        });
+      }();
+
+      var readFutures = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11].map((i) => db.get(
+          'SELECT ? as i, test_sleep(?) as sleep, test_connection_name() as connection',
+          [i, 5 + Random().nextInt(10)]));
+
+      final futures = [...readFutures, withAllConnsFut];
+
+      await for (var result in Stream.fromFutures(futures)) {
+        print("${DateTime.now()} $result");
+      }
+    });
+
+    test('Concurren 2', () async {
       final db1 = await testUtils.setupDatabase(path: path, maxReaders: 3);
       final db2 = await testUtils.setupDatabase(path: path, maxReaders: 3);