lib/src/process_pool.dart - third_party/process_runner - Git at Google

 // Copyright 2020 The Flutter Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 import 'dart:async';
 import 'dart:convert' show Encoding;
 import 'dart:io'
     show Directory, Platform, ProcessStartMode, SystemEncoding, stderr, stdout;

 import 'package:async/async.dart' show StreamGroup;
 import '../process_runner.dart';

 import 'process_runner.dart';

 /// A type of job that can depend on other jobs.
 ///
 /// This is the base type of [WorkerJob] and [WorkerJobGroup], which can all
 /// depend on each other.
 ///
 /// Jobs are not allowed to have a dependency cycle, meaning that they can't
 /// depend on themselves, directly or indirectly. Will throw a
 /// [ProcessRunnerException] if a cycle is detected.
 abstract class DependentJob {
   DependentJob({Iterable<DependentJob> dependsOn = const <DependentJob>{}})
       : _dependsOn = dependsOn.toSet();

   /// The name of the job.
   String get name;

   /// Other jobs that this job depends on.
   ///
   /// This job will not be scheduled until all of the jobs in this set have
   /// completed.
   ///
   /// To add a dependency, call [addDependency] or [addDependencies].
   ///
   /// To remove a dependency, call [removeDependency] or [removeDependencies].
   ///
   /// Modifying the returned set will not affect the dependencies of this job.
   ///
   /// Will throw if there is a dependency cycle, or if the given job has not
   /// been added to the pool.
   ///
   /// Defaults to an empty set.
   Set<DependentJob> get dependsOn => _dependsOn.toSet();
   final Set<DependentJob> _dependsOn;

   /// Add a dependency to this job.
   ///
   /// The given job must complete before this job will executed.
   ///
   /// See also:
   ///
   /// * [removeDependency] which removes a single job.
   /// * [addDependencies] which adds all of the given jobs as dependencies
   ///   of this job.
   void addDependency(DependentJob job) {
     if (job == this) {
       throw ProcessRunnerException('A job cannot depend on itself');
     }
     if (_dependsOn.contains(job)) {
       return;
     }
     if (job._dependsOn.contains(this)) {
       throw ProcessRunnerException(
           '$this is already a dependency of $job, no cycle allowed');
     }
     _dependsOn.add(job);
   }

   /// Remove a dependency to this job that was added with [addDependency].
   ///
   /// If the given job is not a dependency of this job, this will assert.
   ///
   /// See also:
   ///
   /// * [addDependency] which adds a single job.
   /// * [removeDependencies] which removes all of the given jobs as dependencies
   ///   of this job.
   void removeDependency(DependentJob job) {
     assert(_dependsOn.contains(job));
     assert(job != this);
     _dependsOn.remove(job);
   }

   /// Adds all of the [jobs] as dependencies of this job.
   ///
   /// See also:
   ///
   /// * [addDependency] which adds a single job.
   /// * [removeDependencies] which removes all of the given jobs as dependencies
   ///   of this job.
   void addDependencies(Iterable<DependentJob> jobs) {
     // don't just add it to _dependsOn so that subclass addDependency will be
     // called.
     jobs.forEach(addDependency);
   }

   /// Removes all of the given [jobs] as dependencies of this job.
   ///
   /// See also:
   ///
   /// * [removeDependency] which removes a single job.
   /// * [addDependencies] which adds all of the given jobs as dependencies
   ///   of this job.
   void removeDependencies(Iterable<DependentJob> jobs) {
     // don't just remove it from _dependsOn so that subclass removeDependency
     // will be called.
     jobs.forEach(removeDependency);
   }

   /// Adds this job, and any jobs it manages to the given [jobQueue].
   ///
   /// This is called by [ProcessPool.startWorkers] and
   /// [ProcessPool.runToCompletion] to expand the jobs for this worker into
   /// individual jobs.
   void addToQueue(List<DependentJob> jobQueue);
 }

 /// A class that represents a job to be done by a [ProcessPool].
 ///
 /// Create a list of these to pass to [ProcessPool.startWorkers] or
 /// [ProcessPool.runToCompletion].
 class WorkerJob extends DependentJob {
   WorkerJob(
     this.command, {
     String? name,
     this.workingDirectory,
     this.printOutput = false,
     this.stdin,
     this.stdinRaw,
     this.failOk = true,
     this.runInShell = false,
     Iterable<DependentJob>? dependsOn,
   })  : name = name ?? command.join(' '),
         super(dependsOn: dependsOn?.toSet() ?? <DependentJob>{});

   /// The name of the job.
   ///
   /// Defaults to the command args joined by spaces.
   @override
   final String name;

   /// The name and arguments for the process, including the command name as
   /// command[0].
   final List<String> command;

   /// The working directory that the command should be executed in.
   final Directory? workingDirectory;

   /// If set, the stream to read the stdin for this process from.
   ///
   /// It will be encoded using the [ProcessPool.encoding] before being sent to
   /// the process.
   ///
   /// If both [stdin] and [stdinRaw] are set, only [stdinRaw] will be used.
   final Stream<String>? stdin;

   /// If set, the stream to read the raw stdin for this process from.
   ///
   /// It will be used directly, and not encoded (as [stdin] would be).
   ///
   /// If both [stdin] and [stdinRaw] are set, only [stdinRaw] will be used.
   final Stream<List<int>>? stdinRaw;

   /// Whether or not this command should print it's stdout when it runs.
   final bool printOutput;

   /// Whether or not failure of this job should throw an exception.
   ///
   /// If `failOk` is false, and this job fails (returns a non-zero exit code, or
   /// otherwise fails to start), then a [ProcessRunnerException] will be thrown
   /// containing the details.
   ///
   /// Defaults to true, since the [result] will contain the exit code.
   final bool failOk;

   /// If set to true, the process will run be spawned through a system shell.
   ///
   /// Running in a shell is generally not recommended, as it provides worse
   /// performance, and some security risk, but is sometimes necessary for
   /// accessing the shell environment. Shell command line expansion and
   /// interpolation is not performed on the commands, but you can execute shell
   /// builtins. Use the shell builtin "eval" (on Unix systems) if you want to
   /// execute shell commands with expansion.
   ///
   /// On Linux and OS X, `/bin/sh` is used, while on Windows,
   /// `%WINDIR%\system32\cmd.exe` is used.
   ///
   /// Defaults to false.
   final bool runInShell;

   /// Once the job is complete, this contains the result of the job.
   ///
   /// The [stderr] and [stdout] accessors will decode their raw
   /// equivalents using the [ProcessRunner.decoder] that is set on the process
   /// runner for the pool that ran this job.
   ///
   /// If no process runner is supplied to the pool, then the decoder will be the
   /// same as the [ProcessPool.encoding] that was set on the pool.
   ///
   /// The initial value of this field is [ProcessRunnerResult.emptySuccess],
   /// and is updated when the job is complete.
   ProcessRunnerResult result = ProcessRunnerResult.emptySuccess;

   /// Once the job is complete, if it had an exception while running, this
   /// member contains the exception.
   Exception? exception;

   @override
   void addToQueue(List<DependentJob> jobQueue) {
     jobQueue.add(this);
   }

   @override
   String toString() => name;
 }

 /// A job that groups other jobs.
 ///
 /// The [jobs] will be run in the order given to the constructor.
 ///
 /// This group job finishes when all the workers finish.
 class WorkerJobGroup extends DependentJob {
   WorkerJobGroup(Iterable<DependentJob> jobs,
       {Iterable<DependentJob>? dependsOn, this.name = 'Group'})
       : assert(jobs.isNotEmpty),
         jobs = jobs.toList(),
         super(dependsOn: <DependentJob>{
           ...jobs.toSet(),
           if (dependsOn != null) ...dependsOn
         }) {
     // Make sure they run in series, and they depend on anything that the group
     // depends on.
     if (dependsOn != null) {
       this.jobs.first.addDependencies(dependsOn);
     }
     for (var i = 1; i < this.jobs.length; i++) {
       this.jobs[i].addDependency(this.jobs[i - 1]);
     }
   }

   @override
   final String name;

   /// The jobs that will run in order because they depend on each other.
   final List<DependentJob> jobs;

   @override
   void addDependency(DependentJob job) {
     for (final worker in jobs) {
       worker.addDependency(job);
     }
     super.addDependency(job);
   }

   @override
   void removeDependency(DependentJob job) {
     for (final worker in jobs) {
       worker.removeDependency(job);
     }
     super.removeDependency(job);
   }

   @override
   void addToQueue(List<DependentJob> jobQueue) {
     jobQueue.addAll(jobs);
     jobQueue.add(this);
   }

   @override
   String toString() =>
       '${name.isNotEmpty ? name : 'Group'} with ${jobs.length} members';
 }

 /// The type of the reporting function for [ProcessPool.printReport].
 typedef ProcessPoolProgressReporter = void Function(
   int totalJobs,
   int completed,
   int inProgress,
   int pending,
   int failed,
 );

 /// A pool of worker processes that will keep [numWorkers] busy until all of the
 /// (presumably single-threaded) processes are finished.
 class ProcessPool {
   ProcessPool({
     int? numWorkers,
     ProcessRunner? processRunner,
     this.printReport = defaultPrintReport,
     this.encoding = const SystemEncoding(),
   })  : processRunner = processRunner ?? ProcessRunner(decoder: encoding),
         numWorkers = numWorkers ?? Platform.numberOfProcessors;

   /// A function to be called periodically to update the progress on the pool.
   ///
   /// May be set to null if no progress report is desired.
   ///
   /// Defaults to [defaultPrintReport], which prints the progress report to
   /// stdout.
   final ProcessPoolProgressReporter? printReport;

   /// The decoder to use for decoding the stdout, stderr, and output of a
   /// process, and encoding the stdin from the job.
   ///
   /// Defaults to an instance of [SystemEncoding].
   final Encoding encoding;

   /// The process runner to use when running the jobs in the pool.
   ///
   /// Setting this allows for configuration of the process runnner.
   ///
   /// Be default, a default-constructed [ProcessRunner] is used.
   final ProcessRunner processRunner;

   /// The number of workers to use for this pool.
   ///
   /// Defaults to the number of processors the machine has.
   final int numWorkers;

   /// Returns the number of jobs currently in progress.
   int get inProgressJobs => _inProgressJobs;
   int _inProgressJobs = 0;

   /// Returns the number of jobs that have been completed
   int get completedJobs => _completedJobs.length;

   /// Returns the number of jobs that are pending.
   int get pendingJobs => _pendingJobs.length;

   /// Returns the number of jobs that have failed so far.
   int get failedJobs => _failedJobs.length;

   /// Returns the total number of jobs that have been given to this pool.
   int get totalJobs =>
       _completedJobs.length +
       _inProgressJobs +
       _pendingJobs.length +
       _failedJobs.length;

   final List<DependentJob> _pendingJobs = <DependentJob>[];
   final List<DependentJob> _failedJobs = <DependentJob>[];
   final List<DependentJob> _completedJobs = <DependentJob>[];

   void _printReportIfNeeded() {
     if (printReport == null) {
       return;
     }
     printReport?.call(
       totalJobs,
       _completedJobs.length,
       _inProgressJobs,
       _pendingJobs.length,
       _failedJobs.length,
     );
   }

   static String defaultReportToString(
     int total,
     int completed,
     int inProgress,
     int pending,
     int failed,
   ) {
     final percent = total == 0
         ? '100'
         : ((100 * (completed + failed)) ~/ total).toString().padLeft(3);
     final completedStr = completed.toString().padLeft(3);
     final totalStr = total.toString().padRight(3);
     final inProgressStr = inProgress.toString().padLeft(2);
     final pendingStr = pending.toString().padLeft(3);
     final failedStr = failed.toString().padLeft(3);
     return 'Jobs: $percent% done, $completedStr/$totalStr completed, $inProgressStr in progress, $pendingStr pending, $failedStr failed.    \r';
   }

   /// The default report printing function, if one is not supplied.
   static void defaultPrintReport(
     int total,
     int completed,
     int inProgress,
     int pending,
     int failed,
   ) {
     stdout.write(
         defaultReportToString(total, completed, inProgress, pending, failed));
   }

   Future<WorkerJob> _performJob(WorkerJob job) async {
     try {
       if (job.dependsOn.intersection(_failedJobs.toSet()).isNotEmpty) {
         // A dependent job has failed, so just immediately fail this one instead
         // of starting it.
         _addFailedJob(
           job,
           ProcessRunnerException(
             'One or more dependent jobs failed.',
             result: ProcessRunnerResult.failed,
           ),
         );
         return job;
       }
       job.result = await processRunner.runProcess(
         job.command,
         workingDirectory:
             job.workingDirectory ?? processRunner.defaultWorkingDirectory,
         printOutput: job.printOutput,
         stdin: job.stdinRaw ??
             encoding.encoder.bind(job.stdin ?? const Stream<String>.empty()),
         // Starting process pool jobs in any other mode makes no sense: they
         // would all just be immediately started and bring the machine to its
         // knees.
         startMode: ProcessStartMode.normal,
         runInShell: job.runInShell,
         failOk:
             false, // Must be false so that we can catch the exception below.
       );
       _completedJobs.add(job);
     } on ProcessRunnerException catch (e) {
       _addFailedJob(job, e);
       if (!job.failOk) {
         rethrow;
       }
     } finally {
       _inProgressJobs--;
       _printReportIfNeeded();
     }
     return job;
   }

   void _addFailedJob(WorkerJob job, ProcessRunnerException e) {
     job.result = e.result ?? ProcessRunnerResult.failed;
     job.exception = e;
     _failedJobs.add(job);
   }

   DependentJob? _getNextIndependentJob() {
     if (_pendingJobs.isEmpty) {
       return null;
     }
     if (inProgressJobs == 0 && _completedJobs.isEmpty && _failedJobs.isEmpty) {
       final firstIndependent = _pendingJobs
           .indexWhere((DependentJob element) => element.dependsOn.isEmpty);
       if (firstIndependent == -1) {
         throw ProcessRunnerException(
           'Nothing is in progress, and no pending jobs are without '
           'dependencies. At least one must have no dependencies so that '
           'something can start.',
         );
       }
       return _pendingJobs.removeAt(firstIndependent);
     }
     // Go through the list of jobs, looking for the first one where all of its
     // dependencies have been satisfied by appearing in the _completedJobs list.
     final allFinishedJobs = _completedJobs.toSet().union(_failedJobs.toSet());
     for (var i = 0; i < _pendingJobs.length; i += 1) {
       final job = _pendingJobs[i];
       if (job.dependsOn.isEmpty ||
           job.dependsOn.difference(allFinishedJobs.toSet()).isEmpty) {
         return _pendingJobs.removeAt(i);
       }
     }
     // This can be the case if all the dependent jobs are still running.
     return null;
   }

   Stream<WorkerJob> _startWorker() async* {
     while (_pendingJobs.isNotEmpty) {
       final newJob = _getNextIndependentJob();
       if (newJob == null && _inProgressJobs > 0) {
         // All the dependent jobs are still pending.
         // Small pause to let pending jobs complete, so we don't just spin.
         await Future<void>.delayed(const Duration(milliseconds: 10));
         continue;
       }
       if (newJob is WorkerJobGroup) {
         // Just finish up any groups immediately now that all of their workers
         // are done. We keep them until now just in case a job depends on a
         // group. Don't yield these jobs either, since we don't want groups in
         // the output, just completed WorkerJobs.
         _completedJobs.add(newJob);
       } else if (newJob is WorkerJob) {
         _inProgressJobs++;
         yield await _performJob(newJob);
       }
     }
   }

   /// Runs all of the jobs to completion, and returns a list of completed jobs
   /// when all have been completed.
   ///
   /// To listen to jobs as they are completed, use [startWorkers] instead.
   Future<List<WorkerJob>> runToCompletion(Iterable<DependentJob> jobs) async {
     final results = <WorkerJob>[];
     await startWorkers(jobs).forEach(results.add);
     return results;
   }

   /// Runs the `jobs` in parallel, with at most [numWorkers] jobs running
   /// simultaneously.
   ///
   /// If the supplied job is a [WorkerJobGroup], then the jobs in the task
   /// group will be run so that the tasks are executed in order (but still in
   /// parallel with other jobs).
   ///
   /// Returns the the jobs in a [Stream] as they are completed.
   Stream<WorkerJob> startWorkers(Iterable<DependentJob> jobs) async* {
     assert(_inProgressJobs == 0);
     _failedJobs.clear();
     _completedJobs.clear();
     if (jobs.isEmpty) {
       return;
     }
     for (final job in jobs) {
       job.addToQueue(_pendingJobs);
     }
     _verifyDependencies();
     final streams = <Stream<WorkerJob>>[];
     for (var i = 0; i < numWorkers; ++i) {
       if (_pendingJobs.isEmpty) {
         break;
       }
       streams.add(_startWorker());
     }
     await for (final WorkerJob job in StreamGroup.merge<WorkerJob>(streams)) {
       yield job;
     }
     assert(_pendingJobs.isEmpty);
     assert(_inProgressJobs == 0);
     _printReportIfNeeded();
     return;
   }

   bool _hasDependencyLoop(DependentJob job,
       {required Set<DependentJob> visited}) {
     if (visited.contains(job)) {
       return true;
     }
     visited.add(job);
     for (final dependentJob in job.dependsOn) {
       if (_hasDependencyLoop(dependentJob, visited: visited)) {
         return true;
       }
     }
     visited.remove(job);
     return false;
   }

   void _verifyDependencies() {
     // Dependencies for all jobs must also appear in the pending jobs.
     assert(_completedJobs.isEmpty && _inProgressJobs == 0,
         "Can't verify dependencies once started.");
     final pending = _pendingJobs.toSet();
     for (final job in pending) {
       final diff = job.dependsOn.difference(pending);
       if (diff.isNotEmpty) {
         final diffs =
             diff.map<String>((DependentJob item) => item.name).join('\n  ');
         throw ProcessRunnerException(
             "${job.name} has dependent jobs that aren't scheduled to be run:\n"
             '  $diffs');
       }
     }
     // Check for dependency loops.
     for (final job in pending) {
       final visited = <DependentJob>{};
       if (_hasDependencyLoop(job, visited: visited)) {
         throw ProcessRunnerException('Illegal dependency loop detected:\n'
             '  ${<DependentJob>[
           ...visited,
           job
         ].map((DependentJob item) => item.name).join('\n  ')}');
       }
     }
   }
 }
	// Copyright 2020 The Flutter Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style license that can be
	// found in the LICENSE file.

	import 'dart:async';
	import 'dart:convert' show Encoding;
	import 'dart:io'
	show Directory, Platform, ProcessStartMode, SystemEncoding, stderr, stdout;

	import 'package:async/async.dart' show StreamGroup;
	import '../process_runner.dart';

	import 'process_runner.dart';

	/// A type of job that can depend on other jobs.
	///
	/// This is the base type of [WorkerJob] and [WorkerJobGroup], which can all
	/// depend on each other.
	///
	/// Jobs are not allowed to have a dependency cycle, meaning that they can't
	/// depend on themselves, directly or indirectly. Will throw a
	/// [ProcessRunnerException] if a cycle is detected.
	abstract class DependentJob {
	DependentJob({Iterable<DependentJob> dependsOn = const <DependentJob>{}})
	: _dependsOn = dependsOn.toSet();

	/// The name of the job.
	String get name;

	/// Other jobs that this job depends on.
	///
	/// This job will not be scheduled until all of the jobs in this set have
	/// completed.
	///
	/// To add a dependency, call [addDependency] or [addDependencies].
	///
	/// To remove a dependency, call [removeDependency] or [removeDependencies].
	///
	/// Modifying the returned set will not affect the dependencies of this job.
	///
	/// Will throw if there is a dependency cycle, or if the given job has not
	/// been added to the pool.
	///
	/// Defaults to an empty set.
	Set<DependentJob> get dependsOn => _dependsOn.toSet();
	final Set<DependentJob> _dependsOn;

	/// Add a dependency to this job.
	///
	/// The given job must complete before this job will executed.
	///
	/// See also:
	///
	/// * [removeDependency] which removes a single job.
	/// * [addDependencies] which adds all of the given jobs as dependencies
	/// of this job.
	void addDependency(DependentJob job) {
	if (job == this) {
	throw ProcessRunnerException('A job cannot depend on itself');
	}
	if (_dependsOn.contains(job)) {
	return;
	}
	if (job._dependsOn.contains(this)) {
	throw ProcessRunnerException(
	'$this is already a dependency of $job, no cycle allowed');
	}
	_dependsOn.add(job);
	}

	/// Remove a dependency to this job that was added with [addDependency].
	///
	/// If the given job is not a dependency of this job, this will assert.
	///
	/// See also:
	///
	/// * [addDependency] which adds a single job.
	/// * [removeDependencies] which removes all of the given jobs as dependencies
	/// of this job.
	void removeDependency(DependentJob job) {
	assert(_dependsOn.contains(job));
	assert(job != this);
	_dependsOn.remove(job);
	}

	/// Adds all of the [jobs] as dependencies of this job.
	///
	/// See also:
	///
	/// * [addDependency] which adds a single job.
	/// * [removeDependencies] which removes all of the given jobs as dependencies
	/// of this job.
	void addDependencies(Iterable<DependentJob> jobs) {
	// don't just add it to _dependsOn so that subclass addDependency will be
	// called.
	jobs.forEach(addDependency);
	}

	/// Removes all of the given [jobs] as dependencies of this job.
	///
	/// See also:
	///
	/// * [removeDependency] which removes a single job.
	/// * [addDependencies] which adds all of the given jobs as dependencies
	/// of this job.
	void removeDependencies(Iterable<DependentJob> jobs) {
	// don't just remove it from _dependsOn so that subclass removeDependency
	// will be called.
	jobs.forEach(removeDependency);
	}

	/// Adds this job, and any jobs it manages to the given [jobQueue].
	///
	/// This is called by [ProcessPool.startWorkers] and
	/// [ProcessPool.runToCompletion] to expand the jobs for this worker into
	/// individual jobs.
	void addToQueue(List<DependentJob> jobQueue);
	}

	/// A class that represents a job to be done by a [ProcessPool].
	///
	/// Create a list of these to pass to [ProcessPool.startWorkers] or
	/// [ProcessPool.runToCompletion].
	class WorkerJob extends DependentJob {
	WorkerJob(
	this.command, {
	String? name,
	this.workingDirectory,
	this.printOutput = false,
	this.stdin,
	this.stdinRaw,
	this.failOk = true,
	this.runInShell = false,
	Iterable<DependentJob>? dependsOn,
	}) : name = name ?? command.join(' '),
	super(dependsOn: dependsOn?.toSet() ?? <DependentJob>{});

	/// The name of the job.
	///
	/// Defaults to the command args joined by spaces.
	@override
	final String name;

	/// The name and arguments for the process, including the command name as
	/// command[0].
	final List<String> command;

	/// The working directory that the command should be executed in.
	final Directory? workingDirectory;

	/// If set, the stream to read the stdin for this process from.
	///
	/// It will be encoded using the [ProcessPool.encoding] before being sent to
	/// the process.
	///
	/// If both [stdin] and [stdinRaw] are set, only [stdinRaw] will be used.
	final Stream<String>? stdin;

	/// If set, the stream to read the raw stdin for this process from.
	///
	/// It will be used directly, and not encoded (as [stdin] would be).
	///
	/// If both [stdin] and [stdinRaw] are set, only [stdinRaw] will be used.
	final Stream<List<int>>? stdinRaw;

	/// Whether or not this command should print it's stdout when it runs.
	final bool printOutput;

	/// Whether or not failure of this job should throw an exception.
	///
	/// If `failOk` is false, and this job fails (returns a non-zero exit code, or
	/// otherwise fails to start), then a [ProcessRunnerException] will be thrown
	/// containing the details.
	///
	/// Defaults to true, since the [result] will contain the exit code.
	final bool failOk;

	/// If set to true, the process will run be spawned through a system shell.
	///
	/// Running in a shell is generally not recommended, as it provides worse
	/// performance, and some security risk, but is sometimes necessary for
	/// accessing the shell environment. Shell command line expansion and
	/// interpolation is not performed on the commands, but you can execute shell
	/// builtins. Use the shell builtin "eval" (on Unix systems) if you want to
	/// execute shell commands with expansion.
	///
	/// On Linux and OS X, `/bin/sh` is used, while on Windows,
	/// `%WINDIR%\system32\cmd.exe` is used.
	///
	/// Defaults to false.
	final bool runInShell;

	/// Once the job is complete, this contains the result of the job.
	///
	/// The [stderr] and [stdout] accessors will decode their raw
	/// equivalents using the [ProcessRunner.decoder] that is set on the process
	/// runner for the pool that ran this job.
	///
	/// If no process runner is supplied to the pool, then the decoder will be the
	/// same as the [ProcessPool.encoding] that was set on the pool.
	///
	/// The initial value of this field is [ProcessRunnerResult.emptySuccess],
	/// and is updated when the job is complete.
	ProcessRunnerResult result = ProcessRunnerResult.emptySuccess;

	/// Once the job is complete, if it had an exception while running, this
	/// member contains the exception.
	Exception? exception;

	@override
	void addToQueue(List<DependentJob> jobQueue) {
	jobQueue.add(this);
	}

	@override
	String toString() => name;
	}

	/// A job that groups other jobs.
	///
	/// The [jobs] will be run in the order given to the constructor.
	///
	/// This group job finishes when all the workers finish.
	class WorkerJobGroup extends DependentJob {
	WorkerJobGroup(Iterable<DependentJob> jobs,
	{Iterable<DependentJob>? dependsOn, this.name = 'Group'})
	: assert(jobs.isNotEmpty),
	jobs = jobs.toList(),
	super(dependsOn: <DependentJob>{
	...jobs.toSet(),
	if (dependsOn != null) ...dependsOn
	}) {
	// Make sure they run in series, and they depend on anything that the group
	// depends on.
	if (dependsOn != null) {
	this.jobs.first.addDependencies(dependsOn);
	}
	for (var i = 1; i < this.jobs.length; i++) {
	this.jobs[i].addDependency(this.jobs[i - 1]);
	}
	}

	@override
	final String name;

	/// The jobs that will run in order because they depend on each other.
	final List<DependentJob> jobs;

	@override
	void addDependency(DependentJob job) {
	for (final worker in jobs) {
	worker.addDependency(job);
	}
	super.addDependency(job);
	}

	@override
	void removeDependency(DependentJob job) {
	for (final worker in jobs) {
	worker.removeDependency(job);
	}
	super.removeDependency(job);
	}

	@override
	void addToQueue(List<DependentJob> jobQueue) {
	jobQueue.addAll(jobs);
	jobQueue.add(this);
	}

	@override
	String toString() =>
	'${name.isNotEmpty ? name : 'Group'} with ${jobs.length} members';
	}

	/// The type of the reporting function for [ProcessPool.printReport].
	typedef ProcessPoolProgressReporter = void Function(
	int totalJobs,
	int completed,
	int inProgress,
	int pending,
	int failed,
	);

	/// A pool of worker processes that will keep [numWorkers] busy until all of the
	/// (presumably single-threaded) processes are finished.
	class ProcessPool {
	ProcessPool({
	int? numWorkers,
	ProcessRunner? processRunner,
	this.printReport = defaultPrintReport,
	this.encoding = const SystemEncoding(),
	}) : processRunner = processRunner ?? ProcessRunner(decoder: encoding),
	numWorkers = numWorkers ?? Platform.numberOfProcessors;

	/// A function to be called periodically to update the progress on the pool.
	///
	/// May be set to null if no progress report is desired.
	///
	/// Defaults to [defaultPrintReport], which prints the progress report to
	/// stdout.
	final ProcessPoolProgressReporter? printReport;

	/// The decoder to use for decoding the stdout, stderr, and output of a
	/// process, and encoding the stdin from the job.
	///
	/// Defaults to an instance of [SystemEncoding].
	final Encoding encoding;

	/// The process runner to use when running the jobs in the pool.
	///
	/// Setting this allows for configuration of the process runnner.
	///
	/// Be default, a default-constructed [ProcessRunner] is used.
	final ProcessRunner processRunner;

	/// The number of workers to use for this pool.
	///
	/// Defaults to the number of processors the machine has.
	final int numWorkers;

	/// Returns the number of jobs currently in progress.
	int get inProgressJobs => _inProgressJobs;
	int _inProgressJobs = 0;

	/// Returns the number of jobs that have been completed
	int get completedJobs => _completedJobs.length;

	/// Returns the number of jobs that are pending.
	int get pendingJobs => _pendingJobs.length;

	/// Returns the number of jobs that have failed so far.
	int get failedJobs => _failedJobs.length;

	/// Returns the total number of jobs that have been given to this pool.
	int get totalJobs =>
	_completedJobs.length +
	_inProgressJobs +
	_pendingJobs.length +
	_failedJobs.length;

	final List<DependentJob> _pendingJobs = <DependentJob>[];
	final List<DependentJob> _failedJobs = <DependentJob>[];
	final List<DependentJob> _completedJobs = <DependentJob>[];

	void _printReportIfNeeded() {
	if (printReport == null) {
	return;
	}
	printReport?.call(
	totalJobs,
	_completedJobs.length,
	_inProgressJobs,
	_pendingJobs.length,
	_failedJobs.length,
	);
	}

	static String defaultReportToString(
	int total,
	int completed,
	int inProgress,
	int pending,
	int failed,
	) {
	final percent = total == 0
	? '100'
	: ((100 * (completed + failed)) ~/ total).toString().padLeft(3);
	final completedStr = completed.toString().padLeft(3);
	final totalStr = total.toString().padRight(3);
	final inProgressStr = inProgress.toString().padLeft(2);
	final pendingStr = pending.toString().padLeft(3);
	final failedStr = failed.toString().padLeft(3);
	return 'Jobs: $percent% done, $completedStr/$totalStr completed, $inProgressStr in progress, $pendingStr pending, $failedStr failed. \r';
	}

	/// The default report printing function, if one is not supplied.
	static void defaultPrintReport(
	int total,
	int completed,
	int inProgress,
	int pending,
	int failed,
	) {
	stdout.write(
	defaultReportToString(total, completed, inProgress, pending, failed));
	}

	Future<WorkerJob> _performJob(WorkerJob job) async {
	try {
	if (job.dependsOn.intersection(_failedJobs.toSet()).isNotEmpty) {
	// A dependent job has failed, so just immediately fail this one instead
	// of starting it.
	_addFailedJob(
	job,
	ProcessRunnerException(
	'One or more dependent jobs failed.',
	result: ProcessRunnerResult.failed,
	),
	);
	return job;
	}
	job.result = await processRunner.runProcess(
	job.command,
	workingDirectory:
	job.workingDirectory ?? processRunner.defaultWorkingDirectory,
	printOutput: job.printOutput,
	stdin: job.stdinRaw ??
	encoding.encoder.bind(job.stdin ?? const Stream<String>.empty()),
	// Starting process pool jobs in any other mode makes no sense: they
	// would all just be immediately started and bring the machine to its
	// knees.
	startMode: ProcessStartMode.normal,
	runInShell: job.runInShell,
	failOk:
	false, // Must be false so that we can catch the exception below.
	);
	_completedJobs.add(job);
	} on ProcessRunnerException catch (e) {
	_addFailedJob(job, e);
	if (!job.failOk) {
	rethrow;
	}
	} finally {
	_inProgressJobs--;
	_printReportIfNeeded();
	}
	return job;
	}

	void _addFailedJob(WorkerJob job, ProcessRunnerException e) {
	job.result = e.result ?? ProcessRunnerResult.failed;
	job.exception = e;
	_failedJobs.add(job);
	}

	DependentJob? _getNextIndependentJob() {
	if (_pendingJobs.isEmpty) {
	return null;
	}
	if (inProgressJobs == 0 && _completedJobs.isEmpty && _failedJobs.isEmpty) {
	final firstIndependent = _pendingJobs
	.indexWhere((DependentJob element) => element.dependsOn.isEmpty);
	if (firstIndependent == -1) {
	throw ProcessRunnerException(
	'Nothing is in progress, and no pending jobs are without '
	'dependencies. At least one must have no dependencies so that '
	'something can start.',
	);
	}
	return _pendingJobs.removeAt(firstIndependent);
	}
	// Go through the list of jobs, looking for the first one where all of its
	// dependencies have been satisfied by appearing in the _completedJobs list.
	final allFinishedJobs = _completedJobs.toSet().union(_failedJobs.toSet());
	for (var i = 0; i < _pendingJobs.length; i += 1) {
	final job = _pendingJobs[i];
	if (job.dependsOn.isEmpty \|\|
	job.dependsOn.difference(allFinishedJobs.toSet()).isEmpty) {
	return _pendingJobs.removeAt(i);
	}
	}
	// This can be the case if all the dependent jobs are still running.
	return null;
	}

	Stream<WorkerJob> _startWorker() async* {
	while (_pendingJobs.isNotEmpty) {
	final newJob = _getNextIndependentJob();
	if (newJob == null && _inProgressJobs > 0) {
	// All the dependent jobs are still pending.
	// Small pause to let pending jobs complete, so we don't just spin.
	await Future<void>.delayed(const Duration(milliseconds: 10));
	continue;
	}
	if (newJob is WorkerJobGroup) {
	// Just finish up any groups immediately now that all of their workers
	// are done. We keep them until now just in case a job depends on a
	// group. Don't yield these jobs either, since we don't want groups in
	// the output, just completed WorkerJobs.
	_completedJobs.add(newJob);
	} else if (newJob is WorkerJob) {
	_inProgressJobs++;
	yield await _performJob(newJob);
	}
	}
	}

	/// Runs all of the jobs to completion, and returns a list of completed jobs
	/// when all have been completed.
	///
	/// To listen to jobs as they are completed, use [startWorkers] instead.
	Future<List<WorkerJob>> runToCompletion(Iterable<DependentJob> jobs) async {
	final results = <WorkerJob>[];
	await startWorkers(jobs).forEach(results.add);
	return results;
	}

	/// Runs the `jobs` in parallel, with at most [numWorkers] jobs running
	/// simultaneously.
	///
	/// If the supplied job is a [WorkerJobGroup], then the jobs in the task
	/// group will be run so that the tasks are executed in order (but still in
	/// parallel with other jobs).
	///
	/// Returns the the jobs in a [Stream] as they are completed.
	Stream<WorkerJob> startWorkers(Iterable<DependentJob> jobs) async* {
	assert(_inProgressJobs == 0);
	_failedJobs.clear();
	_completedJobs.clear();
	if (jobs.isEmpty) {
	return;
	}
	for (final job in jobs) {
	job.addToQueue(_pendingJobs);
	}
	_verifyDependencies();
	final streams = <Stream<WorkerJob>>[];
	for (var i = 0; i < numWorkers; ++i) {
	if (_pendingJobs.isEmpty) {
	break;
	}
	streams.add(_startWorker());
	}
	await for (final WorkerJob job in StreamGroup.merge<WorkerJob>(streams)) {
	yield job;
	}
	assert(_pendingJobs.isEmpty);
	assert(_inProgressJobs == 0);
	_printReportIfNeeded();
	return;
	}

	bool _hasDependencyLoop(DependentJob job,
	{required Set<DependentJob> visited}) {
	if (visited.contains(job)) {
	return true;
	}
	visited.add(job);
	for (final dependentJob in job.dependsOn) {
	if (_hasDependencyLoop(dependentJob, visited: visited)) {
	return true;
	}
	}
	visited.remove(job);
	return false;
	}

	void _verifyDependencies() {
	// Dependencies for all jobs must also appear in the pending jobs.
	assert(_completedJobs.isEmpty && _inProgressJobs == 0,
	"Can't verify dependencies once started.");
	final pending = _pendingJobs.toSet();
	for (final job in pending) {
	final diff = job.dependsOn.difference(pending);
	if (diff.isNotEmpty) {
	final diffs =
	diff.map<String>((DependentJob item) => item.name).join('\n ');
	throw ProcessRunnerException(
	"${job.name} has dependent jobs that aren't scheduled to be run:\n"
	' $diffs');
	}
	}
	// Check for dependency loops.
	for (final job in pending) {
	final visited = <DependentJob>{};
	if (_hasDependencyLoop(job, visited: visited)) {
	throw ProcessRunnerException('Illegal dependency loop detected:\n'
	' ${<DependentJob>[
	...visited,
	job
	].map((DependentJob item) => item.name).join('\n ')}');
	}
	}
	}
	}