Isolate pool sharing a connection

[kascote]

Hi, I'm working with isolatePool2 to have some workers and each job receive a db connection.
This code is working just great in plain Dart. Now I moved it to Flutter and I see the that db connection never happens, the print line 'db connected' is never show.

What I'm doing wrong? is there some differences between Dart and Flutter that need to take into account?

TIA
:: nelson

class AppDatabase extends _$AppDatabase {
  AppDatabase([QueryExecutor? executor]) : super(executor ?? _openConnection());
  
  static LazyDatabase _openConnection() {
    return LazyDatabase(() async {
      return NativeDatabase.createInBackground(
        File(await _getDatabasePath()),
        logStatements: false,
        readPool: 4,
        setup: (db) async {
          db
            ..execute('PRAGMA journal_mode = WAL;')
            ..execute('PRAGMA foreign_keys = ON;')
            ..execute('PRAGMA encoding = "UTF-8";');
        },
      );
    });
  }
}

=======================================

Future<void> worker(AppDatabase db) async {
  final pool = IsolatePool<int>(4);

  try {
    await pool.start();
    await processData(pool, db);
  } catch (e, s) {
    print('ERROR dispatching job', e, s);
  } finally {
    pool.stop();
  }
}

Future<void> processData(IsolatePool<int> pool, AppDatabase db) async {
  final dbConn = await db.serializableConnection();
  final workers = <Future<int>>[];

  final reports = await db.getReportsToProcess();
  for (final report in reports) {
    workers.add(pool.scheduleJob(ProcessReport<int>(report.id, dbConn)));
  }

  await workers.wait;
}

=======================================

class ProcessReport<T> extends PooledJob<T> {
  final String reportId;
  final DriftIsolate dbConn;

  ProcessReport(this.reportId, this.dbConn);

  @override
  Future<T> job() async {
    print('Processing report: $reportId');
    final db = AppDatabase(await dbConn.connect());
    print('db connected');

    try {
      final report = await db.getReportById(reportId);
      if (report == null) throw Exception('report not found: $feedId');
    
      // Process Report

    } finally {
      await db.close();
    }

    return 1;
  }
}

[simolus3]

I'm not aware of any difference between Dart and Flutter that might cause this. I see however that serializableConnection only supports a single connection when the database you're calling this on is not directly connected to a drift isolate itself. NativeDatabase.createInBackground manages a DriftIsolate, but the LazyDatabase drops that information.

Two possible solutions are:

1. Giving each scheduleJob its own serializableConnection(). That should fix the issue you've seen, but something worth mentioning is that this is less efficient because the original database now acts as a proxy, so the flow is ProcessReport isolate <-> main isolate <-> drift isolates.
2. Using DatabaseConnection.delayed, which wraps a full connection, instead of LazyDatabase. This one forwards the information that the database is connected to a DriftIsolate and will make the spawned isolates connect to that one too. _openConnection would have to return a DatabaseConnection here, and you'll need to use createBackgroundConnection.

return DatabaseConnection.delayed(Future(() async {
  return NativeDatabase.createBackgroundConnection(
    File(await _getDatabasePath()),
    logStatements: false,
    readPool: 4,
    setup: (db) async {
      db
        ..execute('PRAGMA journal_mode = WAL;')
        ..execute('PRAGMA foreign_keys = ON;')
        ..execute('PRAGMA encoding = "UTF-8";');
    },
  );
}));

Something worth mentioning is that createBackgroundConnection is not really meant to be shared across isolates. When using serializableConnection() here, an isolate calling connect() on the other side only receives a single DriftIsolate (the writer), not the whole pool. We don't currently have an API to share a read pool (although you could take a look at sqlite_async which uses a pool by default that can be shared after opening it with isolateConnectionFactory() and supports drift through drift_sqlite_async).
You can also write this functionality yourself, drift provides all the basic building blocks:

final class DriftIsolatePool {
  final DriftIsolate writer;
  final List<DriftIsolate> readers;

  DriftIsolatePool({required this.writer, required this.readers});

  Future<DatabaseConnection> connect() async {
    final writingConnection = await writer.connect();
    final readConnections = await Future.wait([
      for (final readIsolate in readers) readIsolate.connect(),
    ]);

    return DatabaseConnection(
      MultiExecutor.withReadPool(
        reads: readConnections,
        write: writingConnection,
      ),
      streamQueries: writingConnection.streamQueries,
      connectionData: writingConnection.connectionData,
    );
  }

  Future<void> close() async {
    await writer.shutdownAll();
    await Future.wait([for (final reader in readers) reader.shutdownAll()]);
  }

  static Future<DriftIsolatePool> spawn(File file, {int readSize = 4}) async {
    final writer = await _spawnForPool(file.path, false);
    final readers = await Future.wait(
      [for (var i = 0; i < readSize; i++) _spawnForPool(file.path, true)],
    );

    return DriftIsolatePool(writer: writer, readers: readers);
  }

  static Future<DriftIsolate> _spawnForPool(String path, bool reader) {
    return DriftIsolate.spawn(() {
      return NativeDatabase(
        File(path),
        enableMigrations: !reader,
        setup: (db) {
          db
            ..execute('PRAGMA journal_mode = WAL;')
            ..execute('PRAGMA foreign_keys = ON;')
            ..execute('PRAGMA encoding = "UTF-8";');
        },
      );
    });
  }
}

final _pool = Future(() async {
  final path = 'test.db'; //_getDatabasePath();
  return DriftIsolatePool.spawn(File(path));
});

To connect the main isolate, use

static QueryExecutor _openConnection() {
  return DatabaseConnection.delayed(Future(() async {
    final pool = await _pool;
    return await pool.connect();
  }));
}

And to connect other isolates:

Future<void> processData(IsolatePool<int> pool) async {
  final driftPool = await _pool;
  final workers = <Future<int>>[];

  final reports = await db.getReportsToProcess();
  for (final report in reports) {
    workers.add(pool.scheduleJob(ProcessReport<int>(report.id, driftPool)));
  }

  await workers.wait;
}

class ProcessReport<T> extends PooledJob<T> {
  final String reportId;
  final DriftIsolatePool pool;

  ProcessReport(this.reportId, this.pool);

  @override
  Future<T> job() async {
    print('Processing report: $reportId');
    final db = AppDatabase(await pool.connect());
    print('db connected');

    try {
      final report = await db.getReportById(reportId);
      if (report == null) throw Exception('report not found: $feedId');
    
      // Process Report

    } finally {
      await db.close();
    }

    return 1;
  }
}

[kascote]

WOW, thanks for take the time for the explanation and the code sample!!... I tried your DraftIsolatePool and works perfect now.
I need to review a bit more your example, may be I'll came back with some more questions, there are some things that don't click yet for me :p

Thanks!!