[WIP] readwrite pool.

Author: rkistner

packages/driver/src/util/MultiConnectionPool.ts

@@ -20,6 +20,7 @@ export class MultiConnectionPool implements SqliteDriverConnectionPool {
   private _availableReadConnections: SqliteDriverConnection[] = [];
   private _queue: QueuedPoolItem[] = [];
   private _maxConnections: number;
+  private _nextConnectionNumber = 1;
 
   private options: ConnectionPoolOptions;
 
@@ -42,7 +43,8 @@ export class MultiConnectionPool implements SqliteDriverConnectionPool {
     const promise = new Promise<ReservedConnection>((resolve, reject) => {
       this._queue.push({
         resolve,
-        reject
+        reject,
+        options: options ?? {}
       });
     });
 
@@ -57,7 +59,7 @@ export class MultiConnectionPool implements SqliteDriverConnectionPool {
     const connection = await this.factory.openConnection({
       ...this.options,
       ...options,
-      connectionName: `connection-${this._allConnections.size + 1}`
+      connectionName: `connection-${this._nextConnectionNumber++}`
     });
     this._allConnections.add(connection);
     return connection;
@@ -82,7 +84,7 @@ export class MultiConnectionPool implements SqliteDriverConnectionPool {
     let connection: SqliteDriverConnection;
     if (this._availableReadConnections.length == 0) {
       // FIXME: prevent opening more than the max
-      connection = await this.expandPool();
+      connection = await this.expandPool(item.options);
     } else {
       connection = this._availableReadConnections.shift()!;
     }

packages/driver/src/util/connection-pools.ts

@@ -22,6 +22,7 @@ export interface DriverFactory {
 export interface QueuedPoolItem {
   resolve: (reserved: ReservedConnection) => void;
   reject: (err: any) => void;
+  options: ReserveConnectionOptions;
 }
 
 export class ReservedConnectionImpl implements ReservedConnection {

packages/wa-sqlite-driver/src/OPFSCoopSyncVFS2.ts

@@ -72,18 +72,20 @@ export class OPFSCoopSyncVFS2 extends FacadeVFS {
   unboundAccessHandles = new Set<FileSystemSyncAccessHandle>();
   accessiblePaths = new Set<string>();
   releaser: null | (() => void) = null;
+  private readonly: boolean;
 
-  static async create(name, module) {
-    const vfs = new OPFSCoopSyncVFS2(name, module);
+  static async create(name: string, module, readonly: boolean) {
+    const vfs = new OPFSCoopSyncVFS2(name, module, readonly);
     await Promise.all([
       (vfs as any).isReady(),
       vfs.#initialize(DEFAULT_TEMPORARY_FILES)
     ]);
     return vfs;
   }
 
-  constructor(name, module) {
+  constructor(name: string, module, readonly: boolean) {
     super(name, module);
+    this.readonly = readonly;
   }
 
   get #module() {
@@ -143,7 +145,7 @@ export class OPFSCoopSyncVFS2 extends FacadeVFS {
     for (let i = 0; i < nTemporaryFiles; i++) {
       const tmpFile = await tmpDir.getFileHandle(`${i}.tmp`, { create: true });
       const tmpAccessHandle = await (tmpFile as any).createSyncAccessHandle({
-        mode: 'readwrite-unsafe'
+        mode: this.readonly ? 'read-only' : 'readwrite'
       });
       this.unboundAccessHandles.add(tmpAccessHandle);
     }
@@ -543,7 +545,9 @@ export class OPFSCoopSyncVFS2 extends FacadeVFS {
               if (subPersistentFile) {
                 subPersistentFile.accessHandle = await (
                   subPersistentFile.fileHandle as any
-                ).createSyncAccessHandle({ mode: 'readwrite-unsafe' });
+                ).createSyncAccessHandle({
+                  mode: this.readonly ? 'read-only' : 'readwrite'
+                });
               }
             })
           );
@@ -584,14 +588,18 @@ export class OPFSCoopSyncVFS2 extends FacadeVFS {
       setTimeout(notify);
 
       this.log?.(`lock requested: ${lockName}`);
-      navigator.locks.request(lockName, { mode: 'shared' }, (lock) => {
-        // We have the lock. Stop asking other connections for it.
-        this.log?.(`lock acquired: ${lockName}`, lock);
-        clearInterval(notifyId);
-        return new Promise<() => void>((res) => {
-          resolve(res as () => void);
-        });
-      });
+      navigator.locks.request(
+        lockName,
+        { mode: this.readonly ? 'shared' : 'exclusive' },
+        (lock) => {
+          // We have the lock. Stop asking other connections for it.
+          this.log?.(`lock acquired: ${lockName}`, lock);
+          clearInterval(notifyId);
+          return new Promise<() => void>((res) => {
+            resolve(res as () => void);
+          });
+        }
+      );
     });
   }
 }

packages/wa-sqlite-driver/src/pool.ts

@@ -1,7 +1,8 @@
 import { SqliteDriverConnectionPool } from '@sqlite-js/driver';
 import {
   LazyConnectionPool,
-  MultiConnectionPool
+  MultiConnectionPool,
+  ReadWriteConnectionPool
 } from '@sqlite-js/driver/util';
 import {
   ReserveConnectionOptions,
@@ -32,21 +33,26 @@ export function waSqliteSingleWorker(path: string): SqliteDriverConnectionPool {
 }
 
 export function waSqliteWorkerPool(path: string): SqliteDriverConnectionPool {
-  return new MultiConnectionPool(
+  return new ReadWriteConnectionPool(
     {
       async openConnection(
         options?: ReserveConnectionOptions & { connectionName?: string }
       ): Promise<SqliteDriverConnection> {
+        console.log('openConnection', options);
         const connection = new WorkerDriverConnection(
           new Worker(new URL('./wa-sqlite-worker.js', import.meta.url), {
             type: 'module'
           }),
-          { path }
+          {
+            path,
+            readonly: options?.readonly ?? false,
+            connectionName: options?.connectionName
+          }
         );
         await connection.open();
         return connection;
       }
     },
-    {}
+    { maxConnections: 5 }
   );
 }

packages/wa-sqlite-driver/src/wa-sqlite-driver.ts

@@ -190,6 +190,7 @@ class StatementImpl implements SqliteDriverStatement {
   }
 
   async _finalize() {
+    console.log('finalizing...', this.con.path);
     // Wait for these to complete, but ignore any errors.
     // TODO: also wait for run/step to complete
     await this.preparePromise;
@@ -202,7 +203,7 @@ class StatementImpl implements SqliteDriverStatement {
   }
 
   finalize(): void {
-    this._finalize();
+    m.runExclusive(() => this._finalize());
   }
 
   reset(options?: ResetOptions): void {
@@ -287,6 +288,7 @@ export class WaSqliteConnection implements SqliteDriverConnection {
 
   async close() {
     await m.runExclusive(async () => {
+      console.log('closing...', this.path);
       for (let statement of this.statements) {
         if (statement.options.persist) {
           statement.finalize();

packages/wa-sqlite-driver/src/wa-sqlite-worker.ts

@@ -2,13 +2,22 @@ import { OPFSCoopSyncVFS2 } from './OPFSCoopSyncVFS2';
 import { sqlite3, module, WaSqliteConnection } from './wa-sqlite-driver';
 import { setupDriverWorker } from './worker_threads';
 
-// Register a custom file system.
-const vfs = await OPFSCoopSyncVFS2.create('test.db', module);
-// @ts-ignore
-sqlite3.vfs_register(vfs as any, true);
-
+let vfs: OPFSCoopSyncVFS2 | null = null;
 setupDriverWorker({
   async openConnection(options) {
+    // Register a custom file system.
+    if (vfs != null) {
+      throw new Error('Can only open one connection');
+    }
+    console.log('open', options);
+    vfs = await OPFSCoopSyncVFS2.create(
+      'test.db',
+      module,
+      options.readonly ?? false
+    );
+    // @ts-ignore
+    sqlite3.vfs_register(vfs as any, true);
+
     return await WaSqliteConnection.open(options.path, vfs);
   }
 });

packages/wa-sqlite-driver/src/worker_threads/WorkerDriverAdapter.ts

@@ -26,9 +26,12 @@ export class WorkerConnectionAdapter implements WorkerDriver {
   constructor(public connnection: SqliteDriverConnection) {}
 
   statements = new Map<number, SqliteDriverStatement>();
+  private promise: Promise<void> = Promise.resolve();
 
   async close() {
-    await this.connnection.close();
+    const p = this.promise.then(() => this.connnection.close());
+    this.promise = p;
+    return p;
   }
 
   private requireStatement(id: number) {
@@ -123,6 +126,14 @@ export class WorkerConnectionAdapter implements WorkerDriver {
 
   async execute<const T extends SqliteCommand[]>(
     commands: T
+  ): Promise<InferBatchResult<T>> {
+    const p = this.promise.then(() => this._execute(commands));
+    this.promise = p.then(() => {});
+    return p;
+  }
+
+  async _execute<const T extends SqliteCommand[]>(
+    commands: T
   ): Promise<InferBatchResult<T>> {
     let results: SqliteCommandResponse[] = [];
 

packages/wa-sqlite-driver/test/src/concurrency.test.ts

@@ -42,7 +42,9 @@ describe('concurrency tests', () => {
     for (let i = 0; i < 5; i++) {
       const p = (async () => {
         const start = Date.now();
-        await using connection = await driver.reserveConnection();
+        await using connection = await driver.reserveConnection({
+          readonly: true
+        });
 
         using b = connection.prepare('begin immediate');
         await b.step();
@@ -59,5 +61,27 @@ describe('concurrency tests', () => {
       promises.push(p);
     }
     await Promise.all(promises);
+
+    for (let i = 0; i < 5; i++) {
+      const p = (async () => {
+        const start = Date.now();
+        await using connection = await driver.reserveConnection({
+          readonly: true
+        });
+
+        using b = connection.prepare('begin immediate');
+        await b.step();
+        using s = connection.prepare('select * from test_data');
+        const { rows } = await s.step();
+
+        expect(rows).toEqual([{ id: 1, data: 'test' }]);
+        await new Promise((resolve) => setTimeout(resolve, 500));
+
+        using e = connection.prepare('commit');
+        await e.step();
+        console.log('tx done in', Date.now() - start);
+      })();
+      promises.push(p);
+    }
   });
 });