fix: memory leak due to un-limited process usage

README.md

@@ -55,6 +55,7 @@ Options:
                                      "--no-stash".
   --diff-filter [string]             override the default "--diff-filter=ACMR" flag of "git diff" to get list of files
   --no-stash                         disable the backup stash, and do not revert in case of errors
+  --processes [int]                  limit the maximum processes can run at the same time.
 ```
 
 - **`--allow-empty`**: By default, when linter tasks undo all staged changes, lint_staged will exit with an error and abort the commit. Use this flag to allow creating empty git commits.

bin/lint_staged.dart

@@ -16,19 +16,27 @@ void main(List<String> arguments) async {
     ..addFlag('stash',
         defaultsTo: true,
         negatable: true,
-        help: 'Enable the backup stash, and revert in case of errors');
+        help: 'Enable the backup stash, and revert in case of errors')
+    ..addOption('processes',
+        defaultsTo: null,
+        abbr: 'p',
+        help:
+            'The maximum processes can run at the same time, limits it can reduce the memory usage');
   final argResults = argParser.parse(arguments);
   final allowEmpty = argResults['allow-empty'] == true;
   final diff = argResults['diff'];
   final diffFilter = argResults['diff-filter'];
   final stash = argResults['stash'] == true;
+  final numberOfProcessors = argResults['processes'] != null
+      ? int.parse(argResults['processes'])
+      : null;
   final passed = await lintStaged(
-    allowEmpty: allowEmpty,
-    diff: diff,
-    diffFilter: diffFilter,
-    stash: stash,
-    maxArgLength: _maxArgLength ~/ 2,
-  );
+      allowEmpty: allowEmpty,
+      diff: diff,
+      diffFilter: diffFilter,
+      stash: stash,
+      maxArgLength: _maxArgLength ~/ 2,
+      numOfProcesses: numberOfProcessors);
   exit(passed ? 0 : 1);
 }
 

lib/lint_staged.dart

@@ -21,15 +21,18 @@ Future<bool> lintStaged({
   bool stash = true,
   String? workingDirectory,
   int maxArgLength = 0,
+  int? numOfProcesses,
 }) async {
   try {
     final ctx = await runAll(
-        allowEmpty: allowEmpty,
-        diff: diff,
-        diffFilter: diffFilter,
-        stash: stash,
-        maxArgLength: maxArgLength,
-        workingDirectory: workingDirectory);
+      allowEmpty: allowEmpty,
+      diff: diff,
+      diffFilter: diffFilter,
+      stash: stash,
+      maxArgLength: maxArgLength,
+      workingDirectory: workingDirectory,
+      numOfProcesses: numOfProcesses,
+    );
     _printTaskOutput(ctx);
     return true;
   } catch (e) {

lib/src/processes_pool.dart

@@ -0,0 +1,128 @@
+import 'dart:async';
+import 'dart:collection';
+import 'dart:convert';
+import 'dart:io';
+
+typedef OnCompleted = void Function(ProcessResult result);
+
+class ProcessTask {
+  final String executable;
+  final List<String> arguments;
+  final String? workingDirectory;
+  final Map<String, String>? environment;
+  final bool includeParentEnvironment;
+  final bool runInShell;
+  final Encoding? stdoutEncoding;
+  final Encoding? stderrEncoding;
+  final OnCompleted? onCompleted;
+
+  /// The copy of the parameter of Process.run()
+  const ProcessTask(
+    this.executable,
+    this.arguments, {
+    this.workingDirectory,
+    this.environment,
+    this.includeParentEnvironment = true,
+    this.runInShell = false,
+    this.stdoutEncoding = systemEncoding,
+    this.stderrEncoding = systemEncoding,
+    this.onCompleted,
+  });
+
+  Future<ProcessResult> run() async {
+    ProcessResult result = await Process.run(
+      executable,
+      arguments,
+      workingDirectory: workingDirectory,
+      environment: environment,
+      includeParentEnvironment: includeParentEnvironment,
+      runInShell: runInShell,
+      stdoutEncoding: stdoutEncoding,
+      stderrEncoding: stderrEncoding,
+    );
+    Process.killPid(result.pid);
+    return result;
+  }
+}
+
+class ProcessEntity {
+  final ProcessTask task;
+  final Future<ProcessResult> process;
+  const ProcessEntity({
+    required this.task,
+    required this.process,
+  });
+}
+
+class ProcessesPool {
+  final int? size;
+  final Queue<ProcessTask> _tasks = Queue();
+  final List<ProcessEntity?> _processes = [];
+  bool isStarted = false;
+
+  int get tasksNumber => _tasks.length;
+
+  ProcessesPool({
+    this.size,
+  });
+
+  void addAll({
+    List<ProcessTask> tasks = const [],
+  }) async {
+    _tasks.addAll(tasks);
+  }
+
+  void addTask(ProcessTask task) {
+    _tasks.add(task);
+  }
+
+  Future<void> start({
+    OnCompleted? onCompleted,
+  }) async {
+    if (isStarted) {
+      throw Exception('You have already started');
+    }
+    isStarted = true;
+    if (size == null) {
+      await Future.wait(_tasks.map((task) async {
+        ProcessResult result = await task.run();
+        (onCompleted ?? task.onCompleted)?.call(result);
+        _tasks.remove(task);
+      }).toList());
+      isStarted = false;
+      return;
+    }
+    _processes.addAll(List.filled(size!, null));
+    await Future.wait(List.generate(size!, (int index) async {
+      return _runTaskSync(
+        index: index,
+        onCompleted: onCompleted,
+      );
+    }));
+    isStarted = false;
+  }
+
+  Future<ProcessResult?> _runTaskSync({
+    required int index,
+    OnCompleted? onCompleted,
+  }) async {
+    if (_tasks.isEmpty) return null;
+    ProcessTask task = _tasks.removeFirst();
+    Future<ProcessResult> process = task.run();
+    ProcessEntity entity = ProcessEntity(process: process, task: task);
+    _processes[index] = entity;
+    ProcessResult result = await process;
+    _processes[index] = null;
+    (onCompleted ?? task.onCompleted)?.call(result);
+
+    return _runTaskSync(
+      index: index,
+      onCompleted: onCompleted ?? task.onCompleted,
+    );
+  }
+
+  void close() {
+    _tasks.clear();
+    _processes.clear(); // Process.run does not return process instance
+  }
+}

lib/src/run.dart

@@ -8,6 +8,7 @@ import 'chunk.dart';
 import 'config.dart';
 import 'git.dart';
 import 'group.dart';
+import 'processes_pool.dart';
 import 'workflow.dart';
 import 'logging.dart';
 import 'message.dart';
@@ -23,6 +24,7 @@ Future<Context> runAll({
   bool stash = true,
   String? workingDirectory,
   int maxArgLength = 0,
+  int? numOfProcesses,
 }) async {
   final ctx = getInitialContext();
   final fs = FileSystem(workingDirectory);
@@ -87,28 +89,40 @@ Future<Context> runAll({
     spinner.skipped('Hide unstaged changes');
   }
   spinner.progress('Run tasks for staged files...');
-  await Future.wait(groups.values.map((group) async {
-    await Future.wait(group.scripts.map((script) async {
+  ProcessesPool processesPool = ProcessesPool(size: numOfProcesses);
+  for (var group in groups.values) {
+    for (var script in group.scripts) {
       final args = script.split(' ');
       final exe = args.removeAt(0);
-      await Future.wait(group.files.map((file) async {
-        final result = await Process.run(exe, [...args, file],
-            workingDirectory: workingDirectory);
-        final messsages = ['$script $file'];
-        if (result.stderr.toString().trim().isNotEmpty) {
-          messsages.add(red(result.stderr.toString().trim()));
-        }
-        if (result.stdout.toString().trim().isNotEmpty) {
-          messsages.add(result.stdout.toString().trim());
-        }
-        _verbose(messsages.join('\n'));
-        if (result.exitCode != 0) {
-          ctx.output.add(messsages.join('\n'));
-          ctx.errors.add(kTaskError);
-        }
-      }));
-    }));
-  }));
+      for (var file in group.files) {
+        processesPool.addTask(
+          ProcessTask(
+            exe,
+            [...args, file],
+            workingDirectory: workingDirectory,
+            onCompleted: (result) {
+              final messsages = ['$script $file'];
+              if (result.stderr.toString().trim().isNotEmpty) {
+                messsages.add(red(result.stderr.toString().trim()));
+              }
+              if (result.stdout.toString().trim().isNotEmpty) {
+                messsages.add(result.stdout.toString().trim());
+              }
+              _verbose(messsages.join('\n'));
+              if (result.exitCode != 0) {
+                ctx.output.add(messsages.join('\n'));
+                ctx.errors.add(kTaskError);
+              }
+            },
+          ),
+        );
+      }
+    }
+  }
+
+  await processesPool.start();
+  processesPool.close();
+
   spinner.success('Run tasks for staged files');
   if (!applyModifationsSkipped(ctx)) {
     spinner.progress('Apply modifications...');