Procesarea Paralelă a Volumelor Mari de Date cu Isolate Pools în Dart: Evitarea Blocării Event Loop-ului în Flutter și CLI în Dart

# Firul Principal (UI / Main Event Loop)
#         │
#         ▼ [Adăugare Sarcini în Coadă]
# ┌─────────────────────────────────────────────────────────────┐
# │                 IsolatePool Dispatcher                      │
# └──────┬──────────────────────┬──────────────────────┬────────┘
#        │ (Canal SendPort)     │ (Canal SendPort)     │ (Canal SendPort)
#        ▼                      ▼                      ▼
# ┌──────────────┐     ┌──────────────┐     ┌──────────────┐
# │ Worker #1    │     │ Worker #2    │     │ Worker #3    │
# │ Heap Izolat  │     │ Heap Izolat  │     │ Heap Izolat  │
# │ Core CPU #0  │     │ Core CPU #1  │     │ Core CPU #2  │
# └──────────────┘     └──────────────┘     └──────────────┘
#        │                      │                      │
#        └──────────────────────┼──────────────────────┘
#                               ▼ [Returnare Rezultate]
#             Firul Principal recepționează datele fără jank!

import 'dart:async';
import 'dart:io';
import 'dart:isolate';

typedef Task = (R Function(T), T);

class WorkerTask {
  final int taskId;
  final R Function(T) computation;
  final T data;
  final Completer completer;

  WorkerTask(this.taskId, this.computation, this.data) : completer = Completer();
}

class IsolatePool {
  final int workerCount;
  final List _workerSendPorts = [];
  final List _workerReceivePorts = [];
  final List _isolates = [];
  final Map> _pendingTasks = {};

  int _nextTaskId = 0;
  int _roundRobinIndex = 0;
  bool _isStarted = false;

  IsolatePool({int? workerCount})
      : workerCount = workerCount ?? Platform.numberOfProcessors;

  Future start() async {
    if (_isStarted) return;

    for (int i = 0; i < workerCount; i++) {
      final initReceive = ReceivePort();
      final isolate = await Isolate.spawn(_workerLoop, initReceive.sendPort);
      _isolates.add(isolate);

      final workerSendPort = await initReceive.first as SendPort;
      _workerSendPorts.add(workerSendPort);

      // Port pentru recepționarea rezultatelor de la worker:
      final resultReceive = ReceivePort();
      _workerReceivePorts.add(resultReceive);
      workerSendPort.send(resultReceive.sendPort);

      resultReceive.listen((message) {
        if (message is (int taskId, dynamic result, String? error)) {
          final completer = _pendingTasks.remove(taskId);
          if (completer != null) {
            if (error != null) {
              completer.completeError(Exception(error));
            } else {
              completer.complete(result);
            }
          }
        }
      });
    }

    _isStarted = true;
    print('[IsolatePool] Inițializat cu succes cu $workerCount workeri activi.');
  }

  Future execute(R Function(T) computation, T data) {
    if (!_isStarted) throw StateError('IsolatePool-ul nu este pornit!');

    final taskId = _nextTaskId++;
    final completer = Completer();
    _pendingTasks[taskId] = completer;

    // Distribuție uniformă a sarcinilor prin Round-Robin:
    final targetPort = _workerSendPorts[_roundRobinIndex];
    _roundRobinIndex = (_roundRobinIndex + 1) % workerCount;

    targetPort.send((taskId, computation, data));
    return completer.future;
  }

  static void _workerLoop(SendPort mainSendPort) async {
    final workerCommandPort = ReceivePort();
    mainSendPort.send(workerCommandPort.sendPort);

    SendPort? responsePort;

    await for (final message in workerCommandPort) {
      if (message is SendPort) {
        responsePort = message;
      } else if (message is (int, dynamic, dynamic)) {
        final (taskId, Function fn, dynamic data) = message;
        try {
          final result = fn(data);
          responsePort?.send((taskId, result, null));
        } catch (e) {
          responsePort?.send((taskId, null, e.toString()));
        }
      }
    }
  }

  Future shutdown() async {
    for (final port in _workerReceivePorts) {
      port.close();
    }
    for (final isolate in _isolates) {
      isolate.kill(priority: Isolate.immediate);
    }
    _pendingTasks.clear();
    _isStarted = false;
    print('[IsolatePool] Oprit curat.');
  }
}

// Exemplu de consum în producție:
Future main() async {
  final pool = IsolatePool(workerCount: 4);
  await pool.start();

  // Trimitem 10 sarcini intensive paralele către pool:
  final futures = List.generate(10, (i) {
    return pool.execute((data) {
      // Simulare calcul CPU intensiv pe isolate separat:
      int sum = 0;
      for (int k = 0; k < 5000000; k++) {
        sum += (k * data) % 997;
      }
      return sum;
    }, i + 1);
  });

  final results = await Future.wait(futures);
  print('Toate sarcinile s-au executat în paralel fără jank: $results');

  await pool.shutdown();
}