Connection pool with workers; initial concurrency test.

Author: rkistner

packages/driver-tests/src/index.ts

@@ -1 +1,2 @@
 export * from './driver-tests.js';
+export * from './test.js';

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

@@ -1,5 +1,13 @@
 import { SqliteDriverConnectionPool } from '@sqlite-js/driver';
-import { LazyConnectionPool } from '@sqlite-js/driver/util';
+import {
+  LazyConnectionPool,
+  MultiConnectionPool
+} from '@sqlite-js/driver/util';
+import {
+  ReserveConnectionOptions,
+  SqliteDriverConnection
+} from '@sqlite-js/driver';
+
 import { WorkerDriverConnection } from './worker_threads';
 // import { WaSqliteConnection } from './wa-sqlite-driver';
 
@@ -9,7 +17,7 @@ import { WorkerDriverConnection } from './worker_threads';
 //   });
 // }
 
-export function waSqliteWorkerPool(path: string): SqliteDriverConnectionPool {
+export function waSqliteSingleWorker(path: string): SqliteDriverConnectionPool {
   return new LazyConnectionPool(async () => {
     const connection = new WorkerDriverConnection(
       new Worker(new URL('./wa-sqlite-worker.js', import.meta.url), {
@@ -22,3 +30,23 @@ export function waSqliteWorkerPool(path: string): SqliteDriverConnectionPool {
     // return await WaSqliteConnection.open(path);
   });
 }
+
+export function waSqliteWorkerPool(path: string): SqliteDriverConnectionPool {
+  return new MultiConnectionPool(
+    {
+      async openConnection(
+        options?: ReserveConnectionOptions & { connectionName?: string }
+      ): Promise<SqliteDriverConnection> {
+        const connection = new WorkerDriverConnection(
+          new Worker(new URL('./wa-sqlite-worker.js', import.meta.url), {
+            type: 'module'
+          }),
+          { path }
+        );
+        await connection.open();
+        return connection;
+      }
+    },
+    {}
+  );
+}

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

@@ -0,0 +1,63 @@
+import { beforeEach, describe, expect, test } from '@sqlite-js/driver-tests';
+import { waSqliteSingleWorker, waSqliteWorkerPool } from '../../lib/index.js';
+
+describe('concurrency tests', () => {
+  let dbPath: string;
+
+  const open = async () => {
+    const db = await waSqliteWorkerPool(dbPath);
+    return db;
+  };
+
+  beforeEach((context) => {
+    let testNameSanitized = context.fullName.replaceAll(
+      /[\s\/\\>\.\-\:]+/g,
+      '_'
+    );
+
+    if (testNameSanitized.length > 10) {
+      testNameSanitized =
+        testNameSanitized.substring(testNameSanitized.length - 7) +
+        String(Math.random()).substring(1, 4);
+    }
+    dbPath = `test-db/${testNameSanitized}.db`;
+  });
+
+  test('concurrent select', async () => {
+    await using driver = await open();
+    {
+      await using connection = await driver.reserveConnection();
+      using s1 = connection.prepare(
+        'create table test_data(id integer primary key, data text)'
+      );
+      await s1.step();
+      using s2 = connection.prepare(
+        "insert into test_data(data) values('test')"
+      );
+      await s2.step();
+    }
+
+    let promises: Promise<void>[] = [];
+
+    for (let i = 0; i < 5; i++) {
+      const p = (async () => {
+        const start = Date.now();
+        await using connection = await driver.reserveConnection();
+
+        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);
+    }
+    await Promise.all(promises);
+  });
+});

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

@@ -1,11 +1,11 @@
 import { describeDriverTests } from '@sqlite-js/driver-tests';
-import { waSqliteWorkerPool } from '../../lib/index.js';
+import { waSqliteSingleWorker } from '../../lib/index.js';
 
 describeDriverTests(
   'wa-sqlite',
   { getColumns: true, rawResults: true, allowsMissingParameters: false },
   async (path) => {
     console.log('open', path);
-    return waSqliteWorkerPool(path);
+    return waSqliteSingleWorker(path);
   }
 );