dart-concurrency-isolates

Dart の並行処理と Isolate の管理

目次

• 基本ガイドライン
• 適切な Isolate 戦略の選択
• 単発タスクの実装
• 長時間実行ワーカーの実装
• ワークフロー
• 例

基本ガイドライン

• Isolate のメモリ: Isolate 間で共有メモリは一切ないものとして扱います。Isolate 間の通信はメッセージパッシングのみで行います。
• データ転送: Isolate 間で大きな可変オブジェクトを渡すことは避けます。シリアライズのオーバーヘッドを抑えるため、シンプルなデータ型または不変レコードを優先します。
• リソース管理: メモリリークを防ぐため、不要になった Isolate とポートは必ず終了させます。
• プラットフォームの制約: Dart Web プラットフォームでは Isolate を使用しないでください。Web は JavaScript にコンパイルされ、JavaScript では代わりに Web Workers が使用されます。
• 関連スキル: Main Isolate で実行する標準的な非同期処理（Future、Stream、async／await）については、dart-async-programming を参照してください。

適切な Isolate 戦略の選択

条件分岐を適用して、適切な Isolate の実装方法を判断します。

• 単純な単発のバックグラウンドタスクを実行する場合（例：大きな JSON ペイロードを 1 つ解析する、ファイルを圧縮する）：Isolate.run() を使用します。
• 複雑で長時間実行するバックグラウンドワーカーを実行する場合（例：継続的なデータ処理、データベース接続の維持）：Isolate.spawn() を使用し、SendPort と ReceivePort を手動で管理します。

単発タスクの実装

Isolate.run() を使用すると、Isolate を起動して関数を実行し、結果を取得した後、自動的に Isolate を終了できます。

1. トップレベル関数、静的メソッド、またはクロージャを Isolate.run() に渡します。
2. Main Isolate で結果を await します。

Future<Map<String, dynamic>> parseLargeJson(String jsonString) async {
  // Spawns isolate, runs decode, returns result, and terminates automatically.
  return await Isolate.run(() => jsonDecode(jsonString) as Map<String, dynamic>);
}

長時間実行ワーカーの実装

長時間実行するワーカー Isolate との双方向通信を確立するには、SendPort と ReceivePort を手動で管理します。

1. RawReceivePort で初期化する: Main Isolate で RawReceivePort を使用し、起動時のロジックと継続的なメッセージ処理を分離します。
2. 双方向通信を確立する: Isolate.spawn() を介して、Main Isolate の SendPort を Worker Isolate に渡します。Worker Isolate は独自の ReceivePort を作成し、その SendPort を Main Isolate に送り返す必要があります。
3. リクエストとレスポンスを対応付ける: 一意の ID を使用する Completer マップで、Worker Isolate に送信した非同期リクエストを追跡し、レスポンスの受信時に完了させます。
4. エラーを処理する: Worker Isolate で例外をキャッチし、RemoteError オブジェクトとして送り返します。
5. 終了処理: Worker Isolate に専用のシャットダウンメッセージを送信して ReceivePort を閉じ、すべてのリクエストが完了したら Main Isolate の ReceivePort を閉じます。

ワークフロー

タスクの進捗：長時間実行ワーカーのセットアップ

長時間実行ワーカー Isolate を実装する際の進捗を追跡するには、このチェックリストをコピーしてください。

• Isolate の管理をカプセル化する Worker クラスを作成します。
• RawReceivePort を使用してワーカーから最初の SendPort を取得する静的な spawn() メソッドを実装します。
• RawReceivePort.sendPort とワーカーのエントリーポイントメソッドを渡して Isolate.spawn() を呼び出します。
• ワーカーのエントリーポイントメソッド（_startRemoteIsolate）を実装します。
• ワーカー内に ReceivePort を作成し、その SendPort を Main Isolate に送り返します。
• ワーカーの ReceivePort にリスナーを設定し、受信コマンドを処理します。
• Main Isolate の ReceivePort にリスナーを設定し、レスポンスを処理して Completer インスタンスを完了させます。
• シャットダウンコマンドを送信してすべてのポートを閉じる close() メソッドを実装します。
• バリデーターを実行 → エラーを確認 → 修正します（メモリリークがなく、すべてのポートが適切に閉じることを確認します）。

例

堅牢な長時間実行ワーカーの実装

長時間実行するワーカー Isolate と堅牢な双方向通信を行うには、このパターンを使用します。

import 'dart:async';
import 'dart:convert';
import 'dart:isolate';

class JsonWorker {
  final SendPort _commands;
  final ReceivePort _responses;
  final Map<int, Completer<Object?>> _activeRequests = {};
  int _idCounter = 0;
  bool _closed = false;

  JsonWorker._(this._responses, this._commands) {
    _responses.listen(_handleResponsesFromIsolate);
  }

  static Future<JsonWorker> spawn() async {
    final initPort = RawReceivePort();
    final connection = Completer<(ReceivePort, SendPort)>.sync();
    
    initPort.handler = (initialMessage) {
      final commandPort = initialMessage as SendPort;
      connection.complete((
        ReceivePort.fromRawReceivePort(initPort),
        commandPort,
      ));
    };

    try {
      await Isolate.spawn(_startRemoteIsolate, initPort.sendPort);
    } catch (e) {
      initPort.close();
      rethrow;
    }

    final (ReceivePort receivePort, SendPort sendPort) = await connection.future;
    return JsonWorker._(receivePort, sendPort);
  }

  Future<Object?> parseJson(String message) async {
    if (_closed) throw StateError('Worker is closed');
    final completer = Completer<Object?>.sync();
    final id = _idCounter++;
    _activeRequests[id] = completer;
    _commands.send((id, message));
    return await completer.future;
  }

  void _handleResponsesFromIsolate(dynamic message) {
    final (int id, Object? response) = message as (int, Object?);
    final completer = _activeRequests.remove(id)!;

    if (response is RemoteError) {
      completer.completeError(response);
    } else {
      completer.complete(response);
    }

    if (_closed && _activeRequests.isEmpty) _responses.close();
  }

  static void _startRemoteIsolate(SendPort sendPort) {
    final receivePort = ReceivePort();
    sendPort.send(receivePort.sendPort);

    receivePort.listen((message) {
      if (message == 'shutdown') {
        receivePort.close();
        return;
      }
      
      final (int id, String jsonText) = message as (int, String);
      try {
        final jsonData = jsonDecode(jsonText);
        sendPort.send((id, jsonData));
      } catch (e) {
        sendPort.send((id, RemoteError(e.toString(), '')));
      }
    });
  }

  void close() {
    if (!_closed) {
      _closed = true;
      _commands.send('shutdown');
      if (_activeRequests.isEmpty) _responses.close();
    }
  }
}