Merge pull request #6 from journeyapps/improvements

Improved error handling and performance

Author: rkistner

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 0.4.0
+
+- Ensure database connections are cleaned up on unhandled Isolate errors.
+- Minor performance improvements.
+
 ## 0.3.0
 
 - Better error messages for recursive transactions.

example/custom_functions_example.dart

@@ -1,9 +1,7 @@
-import 'dart:ffi';
 import 'dart:io';
 import 'dart:isolate';
 
 import 'package:sqlite_async/sqlite_async.dart';
-import 'package:sqlite3/open.dart' as sqlite_open;
 import 'package:sqlite3/sqlite3.dart' as sqlite;
 
 /// Since the functions need to be created on every SQLite connection,

example/json_example.dart

@@ -63,6 +63,7 @@ RETURNING id''', [users]);
   var ids = idRows.map((row) => row['id']).toList();
 
   // Alternatively, using json1 functions directly.
+  // ignore: unused_local_variable
   var idRows2 = await db.execute('''
 INSERT INTO users(name, email)
 SELECT e.value ->> 'name', e.value ->> 'email' FROM json_each(?) e

lib/src/connection_pool.dart

@@ -26,6 +26,7 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
 
   final Mutex mutex;
 
+  @override
   bool closed = false;
 
   /// Open a new connection pool.
@@ -58,6 +59,9 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       var completer = Completer<T>();
 
       var futures = _readConnections.sublist(0).map((connection) async {
+        if (connection.closed) {
+          _readConnections.remove(connection);
+        }
         try {
           return await connection.readLock((ctx) async {
             if (haveLock) {
@@ -107,6 +111,9 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
     if (closed) {
       throw AssertionError('Closed');
     }
+    if (_writeConnection?.closed == true) {
+      _writeConnection = null;
+    }
     _writeConnection ??= SqliteConnectionImpl(
         upstreamPort: _upstreamPort,
         primary: false,

lib/src/port_channel.dart

@@ -16,16 +16,20 @@ abstract class PortClient {
 }
 
 class ParentPortClient implements PortClient {
-  late Future<SendPort> sendPort;
+  late Future<SendPort> sendPortFuture;
+  SendPort? sendPort;
   ReceivePort receivePort = ReceivePort();
   bool closed = false;
   int _nextId = 1;
 
   Map<int, Completer<Object?>> handlers = HashMap();
 
   ParentPortClient() {
-    final initCompleter = Completer<SendPort>();
-    sendPort = initCompleter.future;
+    final initCompleter = Completer<SendPort>.sync();
+    sendPortFuture = initCompleter.future;
+    sendPortFuture.then((value) {
+      sendPort = value;
+    });
     receivePort.listen((message) {
       if (message is _InitMessage) {
         assert(!initCompleter.isCompleted);
@@ -56,7 +60,7 @@ class ParentPortClient implements PortClient {
   }
 
   Future<void> get ready async {
-    await sendPort;
+    await sendPortFuture;
   }
 
   void _cancelAll(Object error) {
@@ -72,10 +76,11 @@ class ParentPortClient implements PortClient {
     if (closed) {
       throw ClosedException();
     }
-    var completer = Completer<T>();
+    var completer = Completer<T>.sync();
     var id = _nextId++;
     handlers[id] = completer;
-    (await sendPort).send(_RequestMessage(id, message, receivePort.sendPort));
+    final port = sendPort ?? await sendPortFuture;
+    port.send(_RequestMessage(id, message, null));
     return await completer.future;
   }
 
@@ -84,7 +89,8 @@ class ParentPortClient implements PortClient {
     if (closed) {
       throw ClosedException();
     }
-    (await sendPort).send(_FireMessage(message));
+    final port = sendPort ?? await sendPortFuture;
+    port.send(_FireMessage(message));
   }
 
   RequestPortServer server() {
@@ -138,7 +144,7 @@ class ChildPortClient implements PortClient {
 
   @override
   Future<T> post<T>(Object message) async {
-    var completer = Completer<T>();
+    var completer = Completer<T>.sync();
     var id = _nextId++;
     handlers[id] = completer;
     sendPort.send(_RequestMessage(id, message, receivePort.sendPort));
@@ -177,12 +183,13 @@ class RequestPortServer {
 class PortServer {
   final ReceivePort _receivePort = ReceivePort();
   final Future<Object?> Function(Object? message) handle;
+  final SendPort? replyPort;
 
-  PortServer(this.handle) {
+  PortServer(this.handle) : replyPort = null {
     _init();
   }
 
-  PortServer.forSendPort(SendPort port, this.handle) {
+  PortServer.forSendPort(SendPort port, this.handle) : replyPort = port {
     port.send(_InitMessage(_receivePort.sendPort));
     _init();
   }
@@ -208,11 +215,12 @@ class PortServer {
           // Fire and forget
           handle(request.message);
         } else {
+          final replyPort = request.reply ?? this.replyPort;
           try {
             var result = await handle(request.message);
-            request.reply.send(_PortChannelResult.success(request.id, result));
+            replyPort!.send(_PortChannelResult.success(request.id, result));
           } catch (e, stacktrace) {
-            request.reply
+            replyPort!
                 .send(_PortChannelResult.error(request.id, e, stacktrace));
           }
         }
@@ -238,7 +246,7 @@ class _FireMessage {
 class _RequestMessage {
   final int id;
   final Object message;
-  final SendPort reply;
+  final SendPort? reply;
 
   _RequestMessage(this.id, this.message, this.reply);
 }

lib/src/sqlite_connection.dart

@@ -103,4 +103,7 @@ abstract class SqliteConnection extends SqliteWriteContext {
       {Duration? lockTimeout, String? debugContext});
 
   Future<void> close();
+
+  /// Returns true if the connection is closed
+  bool get closed;
 }

lib/src/sqlite_connection_impl.dart

@@ -44,6 +44,11 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
     await _isolateClient.ready;
   }
 
+  @override
+  bool get closed {
+    return _isolateClient.closed;
+  }
+
   Future<void> _open(SqliteOpenFactory openFactory,
       {required bool primary,
       required SerializedPortClient upstreamPort}) async {
@@ -68,7 +73,15 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
   @override
   Future<void> close() async {
     await _connectionMutex.lock(() async {
-      await _isolateClient.post(const _SqliteIsolateConnectionClose());
+      if (readOnly) {
+        await _isolateClient.post(const _SqliteIsolateConnectionClose());
+      } else {
+        // In some cases, disposing a write connection lock the database.
+        // We use the lock here to avoid "database is locked" errors.
+        await _writeMutex.lock(() async {
+          await _isolateClient.post(const _SqliteIsolateConnectionClose());
+        });
+      }
       _isolate.kill();
     });
   }
@@ -212,9 +225,20 @@ void _sqliteConnectionIsolate(_SqliteConnectionParams params) async {
     // running migrations, and other setup.
     await client.post(const InitDb());
   }
+
   final db = await params.openFactory.open(SqliteOpenOptions(
       primaryConnection: params.primary, readOnly: params.readOnly));
 
+  runZonedGuarded(() async {
+    await _sqliteConnectionIsolateInner(params, client, db);
+  }, (error, stack) {
+    db.dispose();
+    throw error;
+  });
+}
+
+Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
+    ChildPortClient client, sqlite.Database db) async {
   final server = params.portServer;
   final commandPort = ReceivePort();
 

lib/src/sqlite_database.dart

@@ -104,6 +104,11 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
     await _initialized;
   }
 
+  @override
+  bool get closed {
+    return _pool.closed;
+  }
+
   void _listenForEvents() {
     UpdateNotification? updates;
 

pubspec.yaml

@@ -1,6 +1,6 @@
 name: sqlite_async
 description: High-performance asynchronous interface for SQLite on Dart and Flutter.
-version: 0.3.0
+version: 0.4.0
 repository: https://github.com/journeyapps/sqlite_async.dart
 environment:
   sdk: '>=2.19.1 <3.0.0'

scripts/benchmark.dart

@@ -60,13 +60,13 @@ List<SqliteBenchmark> benchmarks = [
   }, maxBatchSize: 10000, enabled: true),
   SqliteBenchmark('Write lock',
       (SqliteDatabase db, List<List<String>> parameters) async {
-    for (var params in parameters) {
+    for (var _ in parameters) {
       await db.writeLock((tx) async {});
     }
   }, maxBatchSize: 5000, enabled: false),
   SqliteBenchmark('Read lock',
       (SqliteDatabase db, List<List<String>> parameters) async {
-    for (var params in parameters) {
+    for (var _ in parameters) {
       await db.readLock((tx) async {});
     }
   }, maxBatchSize: 5000, enabled: false),