Convert to using dependencies instead of group lists
diff --git a/example/main.dart b/example/main.dart index 2d37287..a1dc36c 100644 --- a/example/main.dart +++ b/example/main.dart
@@ -121,11 +121,9 @@ int completed, int inProgress, int pending, - int groupsPending, int failed, ) { - stderr.write(ProcessPool.defaultReportToString( - total, completed, inProgress, pending, groupsPending, failed)); + stderr.write(ProcessPool.defaultReportToString(total, completed, inProgress, pending, failed)); } Future<void> main(List<String> args) async { @@ -311,7 +309,7 @@ runInShell: runInShell, failOk: failOk, )); - final Iterable<Job> jobs = parallelJobs.cast<Job>().followedBy(groupedJobs); + final Iterable<DependentJob> jobs = parallelJobs.cast<DependentJob>().followedBy(groupedJobs); try { await for (final WorkerJob done in pool.startWorkers(jobs)) { if (printStdout) {
diff --git a/lib/src/process_pool.dart b/lib/src/process_pool.dart index 53fb419..f0c5599 100644 --- a/lib/src/process_pool.dart +++ b/lib/src/process_pool.dart
@@ -10,29 +10,29 @@ import 'process_runner.dart'; -/// Base class for all job types. -/// -/// Defines an API for getting at the sub-tasks of a job, and for getting the -/// number of sub-tasks. -abstract class Job { - /// Return the stream of sub-tasks for this job. - /// - /// These tasks will be executed in order, but in parallel with tasks from - /// other jobs. - Stream<WorkerJob> get tasks; +abstract class DependentJob { + /// The name of the job. + String get name; - /// Return the number of tasks in this job. + /// Other jobs that this job depends on. /// - /// May be calculated asynchronously, but must complete before the first - /// task will be started. - Future<int> get numTasks; + /// This job will not be scheduled until all of the jobs in this set have + /// completed. + /// + /// 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; + + void addToQueue(List<DependentJob> jobs); } -/// A class that represents a single task to be performed by a [ProcessPool]. +/// 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 Job { +class WorkerJob extends DependentJob { WorkerJob( this.command, { String? name, @@ -42,11 +42,14 @@ this.stdinRaw, this.failOk = true, this.runInShell = false, - }) : name = name ?? command.join(' '); + Iterable<DependentJob>? dependsOn, + }) : name = name ?? command.join(' '), + dependsOn = dependsOn?.toSet() ?? <DependentJob>{}; /// The name of the job. /// - /// Defaults to the args joined by a space. + /// Defaults to the command args joined by spaces. + @override final String name; /// The name and arguments for the process, including the command name as @@ -116,35 +119,49 @@ Exception? exception; @override - Stream<WorkerJob> get tasks async* { - yield this; + void addToQueue(List<DependentJob> jobs) { + jobs.add(this); } @override - Future<int> get numTasks async => 1; + Set<DependentJob> dependsOn; @override - String toString() => command.join(' '); + String toString() => '${command.join(' ')} with ${dependsOn.length} dependencies'; } -class WorkerJobGroup extends Job { - WorkerJobGroup(this.workers); - - final Iterable<WorkerJob> workers; - - @override - Stream<WorkerJob> get tasks async* { - yield* StreamGroup.merge(workers.map((WorkerJob worker) => worker.tasks)); - } - - @override - Future<int> get numTasks async { - int numTasks = 0; - for (final WorkerJob job in workers) { - numTasks += await job.numTasks; +class WorkerJobGroup extends DependentJob { + WorkerJobGroup( + this.workers, { + this.name = '<unknown>', + bool setDependencies = true, + }) : assert(workers.isNotEmpty) { + // Make sure they run in series. + if (setDependencies) { + for (int i = 1; i < workers.length; i++) { + workers[i].dependsOn.add(workers[i - 1]); + } } - return numTasks; } + + @override + final String name; + + // The workers that will run in order because + // They depend on each other. + final List<DependentJob> workers; + + @override + Set<DependentJob> get dependsOn => workers.toSet(); + + @override + void addToQueue(List<DependentJob> jobs) { + jobs.addAll(workers); + jobs.add(this); + } + + @override + String toString() => '${name.isNotEmpty ? name : 'Group'} with ${workers.length} members'; } /// The type of the reporting function for [ProcessPool.printReport]. @@ -153,7 +170,6 @@ int completed, int inProgress, int pending, - int groupsPending, int failed, ); @@ -195,38 +211,30 @@ final int numWorkers; /// Returns the number of jobs currently in progress. - int get inProgressJobs => _inProgressTasks; - int _inProgressTasks = 0; + int get inProgressJobs => _inProgressJobs; + int _inProgressJobs = 0; /// Returns the number of jobs that have been completed - int get completedJobs => _completedTasks.length; + int get completedJobs => _completedJobs.length; /// Returns the number of jobs that are pending. - int get pendingJobs => _pendingTasksCount; - - /// Returns the number of groups that are pending. - int get pendingTaskGroups => _pendingJobGroups.length; + int get pendingJobs => _pendingJobs.length; /// Returns the number of jobs that have failed so far. - int get failedJobs => _failedTasks.length; + int get failedJobs => _failedJobs.length; /// Returns the total number of jobs that have been given to this pool. - int get totalJobs { - return _completedTasks.length + _inProgressTasks + _pendingTasksCount + _failedTasks.length; - } + int get totalJobs => _completedJobs.length + _inProgressJobs + _pendingJobs.length + _failedJobs.length; - int _pendingTasksCount = 0; - - final List<Job> _pendingJobGroups = <Job>[]; - final List<WorkerJob> _failedTasks = <WorkerJob>[]; - final List<WorkerJob> _completedTasks = <WorkerJob>[]; + final List<DependentJob> _pendingJobs = <DependentJob>[]; + final List<DependentJob> _failedJobs = <DependentJob>[]; + final List<DependentJob> _completedJobs = <DependentJob>[]; void _printReportIfNeeded() { if (printReport == null) { return; } - printReport?.call(totalJobs, _completedTasks.length, _inProgressTasks, _pendingTasksCount, - _pendingJobGroups.length, _failedTasks.length); + printReport?.call(totalJobs, _completedJobs.length, _inProgressJobs, _pendingJobs.length, _failedJobs.length); } static String defaultReportToString( @@ -234,18 +242,15 @@ int completed, int inProgress, int pending, - int groupsPending, int failed, ) { - final String percent = - total == 0 ? '100' : ((100 * (completed + failed)) ~/ total).toString().padLeft(3); + final String percent = total == 0 ? '100' : ((100 * (completed + failed)) ~/ total).toString().padLeft(3); final String completedStr = completed.toString().padLeft(3); final String totalStr = total.toString().padRight(3); final String inProgressStr = inProgress.toString().padLeft(2); final String pendingStr = pending.toString().padLeft(3); - final String pendingGroupsStr = groupsPending.toString().padLeft(2); final String failedStr = failed.toString().padLeft(3); - return 'Jobs: $percent% done, $completedStr/$totalStr completed, $inProgressStr in progress, $pendingStr pending (in $pendingGroupsStr groups), $failedStr failed. \r'; + 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. @@ -254,56 +259,104 @@ int completed, int inProgress, int pending, - int groupsPending, int failed, ) { - stdout - .write(defaultReportToString(total, completed, inProgress, pending, groupsPending, failed)); + stdout.write(defaultReportToString(total, completed, inProgress, pending, failed)); } - Stream<WorkerJob> _performTasks(Job job) async* { - await for (final WorkerJob job in job.tasks) { - try { - _inProgressTasks++; - _printReportIfNeeded(); - 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. + 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, + ), ); - _completedTasks.add(job); - } on ProcessRunnerException catch (e) { - job.result = e.result ?? ProcessRunnerResult.failed; - job.exception = e; - _failedTasks.add(job); - if (!job.failOk) { - rethrow; - } - } finally { - _inProgressTasks--; - _printReportIfNeeded(); - yield job; + 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? _getNextInependentJob() { + if (_pendingJobs.isEmpty) { + return null; + } + if (inProgressJobs == 0 && _completedJobs.isEmpty && _failedJobs.isEmpty) { + final int 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 Set<DependentJob> allFinishedJobs = _completedJobs.toSet().union(_failedJobs.toSet()); + for (int i = 0; i < _pendingJobs.length; i += 1) { + final DependentJob 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 (_pendingJobGroups.isNotEmpty) { - final Job newJob = _pendingJobGroups.removeAt(0); - _pendingTasksCount = 0; - final List<Job> jobs = _pendingJobGroups.toList(); - for (final Job job in jobs) { - _pendingTasksCount += await job.numTasks; + while (_pendingJobs.isNotEmpty) { + final DependentJob? newJob = _getNextInependentJob(); + 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; } - _printReportIfNeeded(); - yield* _performTasks(newJob); + if (newJob is! WorkerJob) { + // 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. + if (newJob is WorkerJobGroup) { + _completedJobs.add(newJob); + } + continue; + } + _inProgressJobs++; + yield await _performJob(newJob); } } @@ -311,7 +364,7 @@ /// when all have been completed. /// /// To listen to jobs as they are completed, use [startWorkers] instead. - Future<List<WorkerJob>> runToCompletion(List<Job> jobs) async { + Future<List<WorkerJob>> runToCompletion(Iterable<DependentJob> jobs) async { final List<WorkerJob> results = <WorkerJob>[]; await startWorkers(jobs).forEach(results.add); return results; @@ -325,23 +378,20 @@ /// parallel with other jobs). /// /// Returns the the jobs in a [Stream] as they are completed. - Stream<WorkerJob> startWorkers(Iterable<Job> jobs) async* { - assert(_inProgressTasks == 0); - _failedTasks.clear(); - _completedTasks.clear(); - _pendingTasksCount = 0; - _pendingJobGroups.clear(); + Stream<WorkerJob> startWorkers(Iterable<DependentJob> jobs) async* { + assert(_inProgressJobs == 0); + _failedJobs.clear(); + _completedJobs.clear(); if (jobs.isEmpty) { return; } - _pendingJobGroups.addAll(jobs); - for (final Job job in jobs) { - _pendingTasksCount += await job.numTasks; + for (final DependentJob job in jobs) { + job.addToQueue(_pendingJobs); } - _printReportIfNeeded(); + _verifyDependencies(); final List<Stream<WorkerJob>> streams = <Stream<WorkerJob>>[]; for (int i = 0; i < numWorkers; ++i) { - if (_pendingJobGroups.isEmpty) { + if (_pendingJobs.isEmpty) { break; } streams.add(_startWorker()); @@ -349,8 +399,58 @@ await for (final WorkerJob job in StreamGroup.merge<WorkerJob>(streams)) { yield job; } - assert(_inProgressTasks == 0); - assert(_pendingJobGroups.isEmpty); + assert(_pendingJobs.isEmpty); + assert(_inProgressJobs == 0); + _printReportIfNeeded(); return; } + + bool _hasDependencyLoop(DependentJob job, {required Set<DependentJob> visited}) { + // Check if the job has already been visited, indicating a potential loop + if (visited.contains(job)) { + return true; + } + + // Add the current job to visited set for tracking + visited.add(job); + + // Loop through each dependency of the current job + for (final DependentJob dependentJob in job.dependsOn) { + // Recursively check for loops in dependent jobs + if (_hasDependencyLoop(dependentJob, visited: visited)) { + return true; + } + } + + // Remove the current job from visited set after processing its dependencies + visited.remove(job); + + // No loop found after checking all dependencies + 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 Set<DependentJob> pending = _pendingJobs.toSet(); + for (final DependentJob job in pending) { + final Set<DependentJob> diff = job.dependsOn.difference(pending); + if (diff.isNotEmpty) { + throw ProcessRunnerException("${job.name} has dependent jobs that aren't scheduled to be run:\n" + " ${diff.map<String>((DependentJob item) => item.name).join('\n ')}"); + } + } + // Check for dependency loops. + for (final DependentJob job in pending) { + final Set<DependentJob> visited = <DependentJob>{}; + if (_hasDependencyLoop(job, visited: visited)) { + throw ProcessRunnerException('Dependency loop detected in:\n' + ' ${job.name}: $job\n' + 'Which depends on:\n' + " ${job.dependsOn.map((DependentJob item) => item.name).join('\n ')}" + 'Dependency loop:\n' + " ${visited.map((DependentJob item) => item.name).join('\n ')}"); + } + } + } }
diff --git a/test/src/process_pool_test.dart b/test/src/process_pool_test.dart index 2f3941f..b552d64 100644 --- a/test/src/process_pool_test.dart +++ b/test/src/process_pool_test.dart
@@ -5,6 +5,7 @@ import 'dart:io'; import 'package:process_runner/process_runner.dart'; +import 'package:test/expect.dart'; import 'package:test/test.dart'; import 'fake_process_manager.dart'; @@ -25,8 +26,7 @@ }); test('startWorkers works', () async { - final Map<FakeInvocationRecord, List<ProcessResult>> calls = - <FakeInvocationRecord, List<ProcessResult>>{ + final Map<FakeInvocationRecord, List<ProcessResult>> calls = <FakeInvocationRecord, List<ProcessResult>>{ FakeInvocationRecord(<String>['command', 'arg1', 'arg2'], testPath): <ProcessResult>[ ProcessResult(0, 0, 'output1', ''), ], @@ -39,8 +39,7 @@ fakeProcessManager.verifyCalls(calls.keys); }); test('runToCompletion works', () async { - final Map<FakeInvocationRecord, List<ProcessResult>> calls = - <FakeInvocationRecord, List<ProcessResult>>{ + final Map<FakeInvocationRecord, List<ProcessResult>> calls = <FakeInvocationRecord, List<ProcessResult>>{ FakeInvocationRecord(<String>['command', 'arg1', 'arg2'], testPath): <ProcessResult>[ ProcessResult(0, 0, 'output1', ''), ], @@ -53,8 +52,7 @@ fakeProcessManager.verifyCalls(calls.keys); }); test('failed tests report results', () async { - final Map<FakeInvocationRecord, List<ProcessResult>> calls = - <FakeInvocationRecord, List<ProcessResult>>{ + final Map<FakeInvocationRecord, List<ProcessResult>> calls = <FakeInvocationRecord, List<ProcessResult>>{ FakeInvocationRecord(<String>['command', 'arg1', 'arg2'], testPath): <ProcessResult>[ ProcessResult(0, -1, 'output1', 'stderr1'), ], @@ -70,8 +68,7 @@ expect(completed.first.result.output, equals('output1stderr1')); }); test('failed tests throw when failOk is false', () async { - final Map<FakeInvocationRecord, List<ProcessResult>> calls = - <FakeInvocationRecord, List<ProcessResult>>{ + final Map<FakeInvocationRecord, List<ProcessResult>> calls = <FakeInvocationRecord, List<ProcessResult>>{ FakeInvocationRecord(<String>['command', 'arg1', 'arg2'], testPath): <ProcessResult>[ ProcessResult(0, -1, 'output1', 'stderr1'), ], @@ -88,8 +85,7 @@ fakeProcessManager = FakeProcessManager((String value) {}, commandsThrow: true); processRunner = ProcessRunner(processManager: fakeProcessManager); processPool = ProcessPool(processRunner: processRunner, printReport: null); - final Map<FakeInvocationRecord, List<ProcessResult>> calls = - <FakeInvocationRecord, List<ProcessResult>>{ + final Map<FakeInvocationRecord, List<ProcessResult>> calls = <FakeInvocationRecord, List<ProcessResult>>{ FakeInvocationRecord(<String>['command', 'arg1', 'arg2'], testPath): <ProcessResult>[ ProcessResult(0, -1, 'output1', 'stderr1'), ], @@ -104,38 +100,38 @@ }); test('Commands in task groups run in order, but parallel with other groups', () async { - fakeProcessManager = FakeProcessManager((String value) {}, commandsThrow: true); + fakeProcessManager = FakeProcessManager((String value) {}); processRunner = ProcessRunner(processManager: fakeProcessManager); processPool = ProcessPool(processRunner: processRunner, printReport: null); - final Map<FakeInvocationRecord, List<ProcessResult>> calls = - <FakeInvocationRecord, List<ProcessResult>>{ + final Map<FakeInvocationRecord, List<ProcessResult>> calls = <FakeInvocationRecord, List<ProcessResult>>{ FakeInvocationRecord(<String>['commandA1', 'arg1', 'arg2'], testPath): <ProcessResult>[ - ProcessResult(0, -1, 'output1', 'stderr1'), + ProcessResult(0, 0, 'outputA1', 'stderrA1'), ], FakeInvocationRecord(<String>['commandB1', 'arg1', 'arg2'], testPath): <ProcessResult>[ - ProcessResult(0, -1, 'output1', 'stderr1'), + ProcessResult(0, 0, 'outputB1', 'stderrB1'), ], FakeInvocationRecord(<String>['commandA2', 'arg1', 'arg2'], testPath): <ProcessResult>[ - ProcessResult(0, -1, 'output1', 'stderr1'), + ProcessResult(0, 0, 'outputA2', 'stderrA2'), ], FakeInvocationRecord(<String>['commandB2', 'arg1', 'arg2'], testPath): <ProcessResult>[ - ProcessResult(0, -1, 'output1', 'stderr1'), + ProcessResult(0, -1, 'outputB2', 'stderrB2'), ], FakeInvocationRecord(<String>['commandA3', 'arg1', 'arg2'], testPath): <ProcessResult>[ - ProcessResult(0, -1, 'output1', 'stderr1'), + ProcessResult(0, 0, 'outputA3', 'stderrA3'), ], FakeInvocationRecord(<String>['commandB3', 'arg1', 'arg2'], testPath): <ProcessResult>[ - ProcessResult(0, -1, 'output1', 'stderr1'), + ProcessResult(0, 0, 'outputB3', 'stderrB3'), ], }; fakeProcessManager.fakeResults = calls; - final List<Job> jobs = <Job>[ + final List<WorkerJobGroup> jobs = <WorkerJobGroup>[ WorkerJobGroup( <WorkerJob>[ WorkerJob(<String>['commandA1', 'arg1', 'arg2'], name: 'job A1'), WorkerJob(<String>['commandA2', 'arg1', 'arg2'], name: 'job A2'), WorkerJob(<String>['commandA3', 'arg1', 'arg2'], name: 'job A3'), ], + name: 'Group A', ), WorkerJobGroup( <WorkerJob>[ @@ -143,10 +139,24 @@ WorkerJob(<String>['commandB2', 'arg1', 'arg2'], name: 'job B2'), WorkerJob(<String>['commandB3', 'arg1', 'arg2'], name: 'job B3'), ], + name: 'Group B', ), ]; final List<WorkerJob> completed = await processPool.runToCompletion(jobs); expect(completed.length, equals(6)); + // Command B2 failed with -1, so B3 should also fail. + expect( + completed.where((WorkerJob job) => job.result.exitCode != 0).map((WorkerJob job) => job.name), + unorderedEquals(<String>['job B2', 'job B3']), + ); + expect( + completed.where((WorkerJob job) => job.exception == null).map((WorkerJob job) => job.name), + unorderedEquals(<String>['job A1', 'job B1', 'job A2', 'job A3']), + ); + expect( + completed.where((WorkerJob job) => job.result.exitCode == 0).map((WorkerJob job) => job.name), + unorderedEquals(<String>['job B1', 'job A1', 'job A2', 'job A3']), + ); // Either group A or B can come first, but the individual group tasks should // be in order. expect(