Add IsolateConnectionFactory.

Author: rkistner

lib/src/connection_pool.dart

@@ -26,6 +26,8 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
 
   final Mutex mutex;
 
+  bool closed = false;
+
   /// Open a new connection pool.
   ///
   /// The provided factory is used to open connections on demand. Connections
@@ -100,6 +102,9 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
   @override
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
       {Duration? lockTimeout}) {
+    if (closed) {
+      throw AssertionError('Closed');
+    }
     _writeConnection ??= SqliteConnectionImpl(
         upstreamPort: _upstreamPort,
         primary: false,
@@ -112,7 +117,7 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
   }
 
   Future<void> _expandPool() async {
-    if (_readConnections.length >= maxReaders) {
+    if (closed || _readConnections.length >= maxReaders) {
       return;
     }
     bool hasCapacity = _readConnections.any((connection) => !connection.locked);
@@ -138,4 +143,13 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       await connection.ready;
     }
   }
+
+  @override
+  Future<void> close() async {
+    closed = true;
+    await _writeConnection?.close();
+    for (var connection in _readConnections) {
+      await connection.close();
+    }
+  }
 }

lib/src/isolate_connection_factory.dart

@@ -0,0 +1,26 @@
+import 'sqlite_connection_impl.dart';
+import 'sqlite_connection.dart';
+import 'mutex.dart';
+import 'port_channel.dart';
+import 'sqlite_open_factory.dart';
+
+class IsolateConnectionFactory {
+  SqliteOpenFactory openFactory;
+  Mutex mutex;
+  SerializedPortClient upstreamPort;
+
+  IsolateConnectionFactory(
+      {required this.openFactory,
+      required this.mutex,
+      required this.upstreamPort});
+
+  SqliteConnection open({String? debugName, bool readOnly = false}) {
+    return SqliteConnectionImpl(
+        openFactory: openFactory,
+        mutex: mutex,
+        upstreamPort: upstreamPort,
+        readOnly: readOnly,
+        debugName: debugName,
+        primary: false);
+  }
+}

lib/src/mutex.dart

@@ -39,6 +39,8 @@ class SimpleMutex implements Mutex {
 
   bool get locked => last != null;
 
+  SharedMutex? _shared;
+
   @override
   Future<T> lock<T>(Future<T> Function() callback, {Duration? timeout}) async {
     if (Zone.current[this] != null) {
@@ -98,17 +100,28 @@ class SimpleMutex implements Mutex {
 
   @override
   Future<void> close() async {
+    _shared?.close();
     await lock(() async {});
   }
+
+  SharedMutex get shared {
+    _shared ??= SharedMutex._withMutex(this);
+    return _shared!;
+  }
 }
 
 /// Like Mutex, but can be coped across Isolates.
 class SharedMutex implements Mutex {
   late final SendPort _lockPort;
 
-  SharedMutex._() {
-    final ReceivePort receivePort = ReceivePort();
+  factory SharedMutex._() {
     final Mutex mutex = Mutex();
+    return SharedMutex._withMutex(mutex);
+  }
+
+  SharedMutex._withMutex(Mutex mutex) {
+    final ReceivePort receivePort = ReceivePort();
+
     receivePort.listen((dynamic arg) {
       if (arg is _AcquireMessage) {
         IsolateResult unlock = IsolateResult();
@@ -134,7 +147,7 @@ class SharedMutex implements Mutex {
 
   @override
   Future<void> close() async {
-    final r = IsolateResult<bool>();
+    final r = IsolateResult<void>();
     _lockPort.send(_CloseMessage(r.completer));
     await r.future;
   }

lib/src/port_channel.dart

@@ -59,7 +59,7 @@ class ParentPortClient implements PortClient {
     await sendPort;
   }
 
-  _cancelAll(Object error) {
+  void _cancelAll(Object error) {
     var handlers = this.handlers;
     this.handlers = {};
     for (var message in handlers.values) {
@@ -91,7 +91,7 @@ class ParentPortClient implements PortClient {
     return RequestPortServer(receivePort.sendPort);
   }
 
-  close() async {
+  void close() async {
     if (!closed) {
       closed = true;
 
@@ -182,11 +182,11 @@ class PortServer {
     return SerializedPortClient(sendPort);
   }
 
-  close() {
+  void close() {
     _receivePort.close();
   }
 
-  _init() {
+  void _init() {
     _receivePort.listen((request) async {
       if (request is _FireMessage) {
         handle(request.message);

lib/src/sqlite_connection.dart

@@ -101,4 +101,6 @@ abstract class SqliteConnection extends SqliteWriteContext {
   /// In most cases, [writeTransaction] should be used instead.
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
       {Duration? lockTimeout});
+
+  Future<void> close();
 }

lib/src/sqlite_connection_impl.dart

@@ -22,7 +22,8 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
 
   @override
   final Stream<UpdateNotification>? updates;
-  final ParentPortClient _dbIsolate = ParentPortClient();
+  final ParentPortClient _isolateClient = ParentPortClient();
+  late final Isolate _isolate;
   final String? debugName;
   final bool readOnly;
 
@@ -39,27 +40,35 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
   }
 
   Future<void> get ready async {
-    await _dbIsolate.ready;
+    await _isolateClient.ready;
   }
 
   Future<void> _open(SqliteOpenFactory openFactory,
       {required bool primary,
       required SerializedPortClient upstreamPort}) async {
     await _connectionMutex.lock(() async {
-      var isolate = await Isolate.spawn(
+      _isolate = await Isolate.spawn(
           _sqliteConnectionIsolate,
           _SqliteConnectionParams(
               openFactory: openFactory,
               port: upstreamPort,
               primary: primary,
-              portServer: _dbIsolate.server(),
+              portServer: _isolateClient.server(),
               readOnly: readOnly),
           debugName: debugName,
           paused: true);
-      _dbIsolate.tieToIsolate(isolate);
-      isolate.resume(isolate.pauseCapability!);
+      _isolateClient.tieToIsolate(_isolate);
+      _isolate.resume(_isolate.pauseCapability!);
 
-      await _dbIsolate.ready;
+      await _isolateClient.ready;
+    });
+  }
+
+  @override
+  Future<void> close() async {
+    await _connectionMutex.lock(() async {
+      await _isolateClient.post(_SqliteIsolateConnectionClose());
+      _isolate.kill();
     });
   }
 
@@ -73,7 +82,7 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
     // Private lock to synchronize this with other statements on the same connection,
     // to ensure that transactions aren't interleaved.
     return _connectionMutex.lock(() async {
-      final ctx = _TransactionContext(_dbIsolate);
+      final ctx = _TransactionContext(_isolateClient);
       try {
         return await callback(ctx);
       } finally {
@@ -96,7 +105,7 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
       }
       // DB lock so that only one write happens at a time
       return await _writeMutex.lock(() async {
-        final ctx = _TransactionContext(_dbIsolate);
+        final ctx = _TransactionContext(_isolateClient);
         try {
           return await callback(ctx);
         } finally {
@@ -269,6 +278,8 @@ void _sqliteConnectionIsolate(_SqliteConnectionParams params) async {
           updatedTables = {};
         }
       }
+    } else if (data is _SqliteIsolateConnectionClose) {
+      db.dispose();
     }
   });
 
@@ -312,3 +323,7 @@ class _SqliteIsolateClose {
 
   const _SqliteIsolateClose(this.ctxId);
 }
+
+class _SqliteIsolateConnectionClose {
+  const _SqliteIsolateConnectionClose();
+}

lib/src/sqlite_database.dart

@@ -2,6 +2,7 @@ import 'dart:async';
 
 import 'connection_pool.dart';
 import 'database_utils.dart';
+import 'isolate_connection_factory.dart';
 import 'mutex.dart';
 import 'port_channel.dart';
 import 'sqlite_connection.dart';
@@ -22,7 +23,7 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
   final int maxReaders;
 
   /// Global lock to serialize write transactions.
-  final Mutex mutex = Mutex();
+  final SimpleMutex mutex = SimpleMutex();
 
   /// Factory that opens a raw database connection in each isolate.
   ///
@@ -122,6 +123,13 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
     });
   }
 
+  IsolateConnectionFactory isolateConnectionFactory() {
+    return IsolateConnectionFactory(
+        openFactory: openFactory,
+        mutex: mutex.shared,
+        upstreamPort: _eventsPort.client());
+  }
+
   SqliteConnectionImpl _openPrimaryConnection({String? debugName}) {
     return SqliteConnectionImpl(
         upstreamPort: _eventsPort.client(),
@@ -133,6 +141,14 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
         openFactory: openFactory);
   }
 
+  @override
+  Future<void> close() async {
+    await _pool.close();
+    _updatesController.close();
+    _eventsPort.close();
+    await mutex.close();
+  }
+
   /// Open a read-only transaction.
   ///
   /// Up to [maxReaders] read transactions can run concurrently.

scripts/benchmark.dart

@@ -1,3 +1,5 @@
+import 'dart:isolate';
+
 import 'package:benchmarking/benchmarking.dart';
 import 'package:collection/collection.dart';
 
@@ -47,6 +49,20 @@ List<SqliteBenchmark> benchmarks = [
       }
     });
   }, maxBatchSize: 1000),
+  SqliteBenchmark('Insert: writeTransaction in isolate',
+      (SqliteDatabase db, List<List<String>> parameters) async {
+    var factory = db.isolateConnectionFactory();
+    await Isolate.run(() async {
+      final db = factory.open();
+      await db.writeTransaction((tx) async {
+        for (var params in parameters) {
+          await tx.execute(
+              'INSERT INTO customers(name, email) VALUES(?, ?)', params);
+        }
+      });
+      await db.close();
+    });
+  }, maxBatchSize: 1000),
   SqliteBenchmark('Insert: writeTransaction no await',
       (SqliteDatabase db, List<List<String>> parameters) async {
     await db.writeTransaction((tx) async {
@@ -148,4 +164,6 @@ void main() async {
   for (var entry in benchmarks) {
     await benchmark(entry);
   }
+
+  await db.close();
 }

test/isolate_test.dart

@@ -0,0 +1,37 @@
+import 'dart:isolate';
+import 'dart:math';
+
+import 'package:sqlite_async/sqlite_async.dart';
+import 'package:test/test.dart';
+import 'package:sqlite3/sqlite3.dart' as sqlite;
+import 'util.dart';
+
+void main() {
+  setupLogger();
+
+  group('Isolate Tests', () {
+    late String path;
+
+    setUp(() async {
+      path = dbPath();
+      await cleanDb(path: path);
+    });
+
+    tearDown(() async {
+      await cleanDb(path: path);
+    });
+
+    test('Basic Isolate usage', () async {
+      final db = await setupDatabase(path: path);
+      final factory = db.isolateConnectionFactory();
+
+      final result = await Isolate.run(() async {
+        final db = factory.open();
+        await db
+            .execute('CREATE TABLE test_in_isolate(id INTEGER PRIMARY KEY)');
+        return await db.get('SELECT count() as count FROM test_in_isolate');
+      });
+      expect(result, equals({'count': 0}));
+    });
+  });
+}