feat: pool read connections in MDSQLiteAdapter

PowerSync/PowerSync.Common/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## 0.0.11-alpha.1
 
+- Pool read connections in `MDSQLiteAdapter`, improving performance in any case where multiple queries run simultaneously (eg. via `Watch`). The number of connections can be set via `MDSQLiteOptions.ReadPoolSize` and defaults to 5.
 - Removed the `RunListener` and `RunListenerAsync` APIs from `IEventStream`. Users are encouraged to use the `Listen` or `ListenAsync` APIs instead (`RunListener` itself was implemented using the `Listen` API).
 - Changed the `PowerSyncDatabase.Watch` syntax to return an IAsyncEnumerable instead of accepting a callback. This allows users to handle database changes when they are ready instead of us eagerly running the callback as soon as a change is detected.
 

PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs

@@ -496,7 +496,7 @@ public ISyncStream SyncStream(string name, Dictionary<string, object>? parameter
         ConnectionManager.Close();
         BucketStorageAdapter?.Close();
 
-        Database.Close();
+        await Database.Close();
         Closed = true;
         Emit(new PowerSyncDBEvent { Closed = true });
     }

PowerSync/PowerSync.Common/DB/IDBAdapter.cs

@@ -118,7 +118,7 @@ public interface IDBAdapter : IEventStream<DBAdapterEvent>, ILockContext
     /// <summary>
     /// Closes the adapter. 
     /// </summary>
-    new void Close();
+    new Task Close();
 
     /// <summary>
     /// The name of the adapter.

PowerSync/PowerSync.Common/MDSQLite/MDSQLiteAdapter.cs

@@ -1,6 +1,8 @@
 namespace PowerSync.Common.MDSQLite;
 
 using System;
+using System.Collections.Generic;
+using System.Threading.Channels;
 using System.Threading.Tasks;
 
 using Microsoft.Data.Sqlite;
@@ -15,35 +17,35 @@ public class MDSQLiteAdapterOptions()
     public string Name { get; set; } = null!;
 
     public MDSQLiteOptions? SqliteOptions;
-
 }
 
 public class MDSQLiteAdapter : EventStream<DBAdapterEvent>, IDBAdapter
 {
     public string Name => options.Name;
 
-    public MDSQLiteConnection? writeConnection;
-    public MDSQLiteConnection? readConnection;
+    // One writer
+    private MDSQLiteConnection writeConnection = null!;
+    private readonly AsyncLock writeMutex = new();
+
+    // Many readers
+    private Channel<MDSQLiteConnection> readPool = null!;
 
     private readonly Task initialized;
 
     protected MDSQLiteAdapterOptions options;
 
-    protected RequiredMDSQLiteOptions resolvedMDSQLiteOptions;
+    protected RequiredMDSQLiteOptions resolvedOptions;
     private CancellationTokenSource? tablesUpdatedCts;
     private Task? tablesUpdatedTask;
 
-    private readonly AsyncLock writeMutex = new();
-    private readonly AsyncLock readMutex = new();
-
     public MDSQLiteAdapter(MDSQLiteAdapterOptions options)
     {
         this.options = options;
-        resolvedMDSQLiteOptions = resolveMDSQLiteOptions(options.SqliteOptions);
+        resolvedOptions = ResolveMDSQLiteOptions(options.SqliteOptions);
         initialized = Init();
     }
 
-    private RequiredMDSQLiteOptions resolveMDSQLiteOptions(MDSQLiteOptions? options)
+    private RequiredMDSQLiteOptions ResolveMDSQLiteOptions(MDSQLiteOptions? options)
     {
         var defaults = RequiredMDSQLiteOptions.DEFAULT_SQLITE_OPTIONS;
         return new RequiredMDSQLiteOptions
@@ -55,28 +57,28 @@ private RequiredMDSQLiteOptions resolveMDSQLiteOptions(MDSQLiteOptions? options)
             TemporaryStorage = options?.TemporaryStorage ?? defaults.TemporaryStorage,
             LockTimeoutMs = options?.LockTimeoutMs ?? defaults.LockTimeoutMs,
             EncryptionKey = options?.EncryptionKey ?? defaults.EncryptionKey,
-            Extensions = options?.Extensions ?? defaults.Extensions
+            Extensions = options?.Extensions ?? defaults.Extensions,
+            ReadPoolSize = options?.ReadPoolSize ?? defaults.ReadPoolSize,
         };
     }
 
     private async Task Init()
     {
         writeConnection = await OpenConnection(options.Name);
-        readConnection = await OpenConnection(options.Name);
 
         string[] baseStatements =
         [
-            $"PRAGMA busy_timeout = {resolvedMDSQLiteOptions.LockTimeoutMs}",
-            $"PRAGMA cache_size = -{resolvedMDSQLiteOptions.CacheSizeKb}",
-            $"PRAGMA temp_store = {resolvedMDSQLiteOptions.TemporaryStorage}"
+            $"PRAGMA busy_timeout = {resolvedOptions.LockTimeoutMs}",
+            $"PRAGMA cache_size = -{resolvedOptions.CacheSizeKb}",
+            $"PRAGMA temp_store = {resolvedOptions.TemporaryStorage}"
         ];
 
         string[] writeConnectionStatements =
         [
             .. baseStatements,
-            $"PRAGMA journal_mode = {resolvedMDSQLiteOptions.JournalMode}",
-            $"PRAGMA journal_size_limit = {resolvedMDSQLiteOptions.JournalSizeLimit}",
-            $"PRAGMA synchronous = {resolvedMDSQLiteOptions.Synchronous}",
+            $"PRAGMA journal_mode = {resolvedOptions.JournalMode}",
+            $"PRAGMA journal_size_limit = {resolvedOptions.JournalSizeLimit}",
+            $"PRAGMA synchronous = {resolvedOptions.Synchronous}",
         ];
 
         string[] readConnectionStatements =
@@ -90,15 +92,21 @@ private async Task Init()
             await writeConnection!.Execute(statement);
         }
 
-        foreach (var statement in readConnectionStatements)
+        readPool = Channel.CreateBounded<MDSQLiteConnection>(resolvedOptions.ReadPoolSize);
+        for (int i = 0; i < resolvedOptions.ReadPoolSize; i++)
         {
-            await readConnection!.Execute(statement);
+            var readConnection = await OpenConnection(options.Name);
+            foreach (var statement in readConnectionStatements)
+            {
+                await readConnection.Execute(statement);
+            }
+            await readPool.Writer.WriteAsync(readConnection);
         }
 
         tablesUpdatedCts = new CancellationTokenSource();
         tablesUpdatedTask = Task.Run(async () =>
         {
-            await foreach (var notification in writeConnection!.ListenAsync(tablesUpdatedCts.Token))
+            await foreach (var notification in writeConnection.ListenAsync(tablesUpdatedCts.Token))
             {
                 if (notification.TablesUpdated != null)
                 {
@@ -121,7 +129,8 @@ protected async Task<MDSQLiteConnection> OpenConnection(string dbFilename)
 
     private static SqliteConnection OpenDatabase(string dbFilename)
     {
-        var connection = new SqliteConnection($"Data Source={dbFilename}");
+        string connectionString = $"Data Source={dbFilename};Pooling=False;";
+        var connection = new SqliteConnection(connectionString);
         connection.Open();
         return connection;
     }
@@ -133,13 +142,17 @@ protected virtual void LoadExtension(SqliteConnection db)
         db.LoadExtension(extensionPath, "sqlite3_powersync_init");
     }
 
-    public new void Close()
+    public new async Task Close()
     {
         tablesUpdatedCts?.Cancel();
-        try { tablesUpdatedTask?.Wait(TimeSpan.FromSeconds(2)); } catch { }
+        try { tablesUpdatedTask?.Wait(2000); } catch { /* expected */ }
         base.Close();
         writeConnection?.Close();
-        readConnection?.Close();
+        await WithAllReaders((connections) =>
+        {
+            foreach (var conn in connections) conn.Close();
+        });
+        readPool.Writer.Complete();
     }
 
     public async Task<NonQueryResult> Execute(string query, object?[]? parameters = null)
@@ -184,20 +197,22 @@ public async Task<dynamic> Get(string sql, object?[]? parameters = null)
 
     public async Task<T> ReadTransaction<T>(Func<ITransaction, Task<T>> fn, DBLockOptions? options = null)
     {
-        return await ReadLock((ctx) => InternalTransaction(new MDSQLiteTransaction(readConnection!)!, fn));
+        return await ReadLock((ctx) => InternalTransaction(new MDSQLiteTransaction((MDSQLiteConnection)ctx), fn));
     }
 
     public async Task<T> ReadLock<T>(Func<ILockContext, Task<T>> fn, DBLockOptions? options = null)
     {
         await initialized;
 
-        T result;
-        using (await readMutex.LockAsync())
+        var conn = await readPool.Reader.ReadAsync();
+        try
         {
-            result = await fn(readConnection!);
+            return await fn(conn);
+        }
+        finally
+        {
+            readPool.Writer.TryWrite(conn);
         }
-
-        return result;
     }
 
     public async Task WriteLock(Func<ILockContext, Task> fn, DBLockOptions? options = null)
@@ -206,10 +221,10 @@ public async Task WriteLock(Func<ILockContext, Task> fn, DBLockOptions? options
 
         using (await writeMutex.LockAsync())
         {
-            await fn(writeConnection!);
+            await fn(writeConnection);
         }
 
-        writeConnection!.FlushUpdates();
+        writeConnection.FlushUpdates();
 
     }
 
@@ -220,22 +235,22 @@ public async Task<T> WriteLock<T>(Func<ILockContext, Task<T>> fn, DBLockOptions?
         T result;
         using (await writeMutex.LockAsync())
         {
-            result = await fn(writeConnection!);
+            result = await fn(writeConnection);
         }
 
-        writeConnection!.FlushUpdates();
+        writeConnection.FlushUpdates();
 
         return result;
     }
 
     public async Task WriteTransaction(Func<ITransaction, Task> fn, DBLockOptions? options = null)
     {
-        await WriteLock(ctx => InternalTransaction(new MDSQLiteTransaction(writeConnection!)!, fn));
+        await WriteLock(ctx => InternalTransaction(new MDSQLiteTransaction(writeConnection), fn));
     }
 
     public async Task<T> WriteTransaction<T>(Func<ITransaction, Task<T>> fn, DBLockOptions? options = null)
     {
-        return await WriteLock((ctx) => InternalTransaction(new MDSQLiteTransaction(writeConnection!)!, fn));
+        return await WriteLock((ctx) => InternalTransaction(new MDSQLiteTransaction(writeConnection), fn));
     }
 
     protected static async Task InternalTransaction(
@@ -283,8 +298,53 @@ private static async Task RunTransaction(
     public async Task RefreshSchema()
     {
         await initialized;
-        await writeConnection!.RefreshSchema();
-        await readConnection!.RefreshSchema();
+        await writeConnection.RefreshSchema();
+        await WithAllReaders(async (connections) =>
+        {
+            foreach (var conn in connections) await conn.RefreshSchema();
+        });
+    }
+
+    private async Task WithAllReaders(Func<List<MDSQLiteConnection>, Task> callback)
+    {
+        var connections = new List<MDSQLiteConnection>(resolvedOptions.ReadPoolSize);
+        for (int i = 0; i < resolvedOptions.ReadPoolSize; i++)
+        {
+            connections.Add(await readPool.Reader.ReadAsync());
+        }
+
+        try
+        {
+            await callback(connections);
+        }
+        finally
+        {
+            foreach (var conn in connections)
+            {
+                readPool.Writer.TryWrite(conn);
+            }
+        }
+    }
+
+    private async Task WithAllReaders(Action<List<MDSQLiteConnection>> callback)
+    {
+        var connections = new List<MDSQLiteConnection>(resolvedOptions.ReadPoolSize);
+        for (int i = 0; i < resolvedOptions.ReadPoolSize; i++)
+        {
+            connections.Add(await readPool.Reader.ReadAsync());
+        }
+
+        try
+        {
+            callback(connections);
+        }
+        finally
+        {
+            foreach (var conn in connections)
+            {
+                readPool.Writer.TryWrite(conn);
+            }
+        }
     }
 }
 

PowerSync/PowerSync.Common/MDSQLite/MDSQLiteOptions.cs

@@ -102,6 +102,11 @@ public class MDSQLiteOptions
     /// Load extensions using the path and entryPoint.
     /// </summary>
     public SqliteExtension[]? Extensions { get; set; }
+
+    /// <summary>
+    /// The number of MDSQLiteConnection objects to create for the read pool.
+    /// </summary>
+    public int? ReadPoolSize { get; set; }
 }
 
 public class RequiredMDSQLiteOptions : MDSQLiteOptions
@@ -115,7 +120,8 @@ public class RequiredMDSQLiteOptions : MDSQLiteOptions
         TemporaryStorage = TemporaryStorageOption.MEMORY,
         LockTimeoutMs = 30000,
         EncryptionKey = null,
-        Extensions = []
+        Extensions = [],
+        ReadPoolSize = 5,
     };
 
     public new SqliteJournalMode JournalMode { get; set; } = null!;
@@ -131,5 +137,8 @@ public class RequiredMDSQLiteOptions : MDSQLiteOptions
     public new TemporaryStorageOption TemporaryStorage { get; set; } = null!;
 
     public new int CacheSizeKb { get; set; }
+
     public new SqliteExtension[] Extensions { get; set; } = null!;
+
+    public new int ReadPoolSize { get; set; }
 }