Make `flutter update-packages` run in parallel (#91006)

This modifies the flutter update-packages and flutter update-packages --force-upgrade commands so that the many invocations of "dart pub get" in each repo project run in parallel instead of in series.

packages/flutter_tools/lib/src/base/task_queue.dart

+// Copyright 2014 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:collection';
+
+import '../globals_null_migrated.dart' as globals;
+
+/// A closure type used by the [TaskQueue].
+typedef TaskQueueClosure<T> = Future<T> Function();
+
+/// A task queue of Futures to be completed in parallel, throttling
+/// the number of simultaneous tasks.
+///
+/// The tasks return results of type T.
+class TaskQueue<T> {
+  /// Creates a task queue with a maximum number of simultaneous jobs.
+  /// The [maxJobs] parameter defaults to the number of CPU cores on the
+  /// system.
+  TaskQueue({int? maxJobs})
+      : maxJobs = maxJobs ?? globals.platform.numberOfProcessors;
+
+  /// The maximum number of jobs that this queue will run simultaneously.
+  final int maxJobs;
+
+  final Queue<_TaskQueueItem<T>> _pendingTasks = Queue<_TaskQueueItem<T>>();
+  final Set<_TaskQueueItem<T>> _activeTasks = <_TaskQueueItem<T>>{};
+  final Set<Completer<void>> _completeListeners = <Completer<void>>{};
+
+  /// Returns a future that completes when all tasks in the [TaskQueue] are
+  /// complete.
+  Future<void> get tasksComplete {
+    // In case this is called when there are no tasks, we want it to
+    // signal complete immediately.
+    if (_activeTasks.isEmpty && _pendingTasks.isEmpty) {
+      return Future<void>.value();
+    }
+    final Completer<void> completer = Completer<void>();
+    _completeListeners.add(completer);
+    return completer.future;
+  }
+
+  /// Adds a single closure to the task queue, returning a future that
+  /// completes when the task completes.
+  Future<T> add(TaskQueueClosure<T> task) {
+    final Completer<T> completer = Completer<T>();
+    _pendingTasks.add(_TaskQueueItem<T>(task, completer));
+    if (_activeTasks.length < maxJobs) {
+      _processTask();
+    }
+    return completer.future;
+  }
+
+  // Process a single task.
+  void _processTask() {
+    if (_pendingTasks.isNotEmpty && _activeTasks.length <= maxJobs) {
+      final _TaskQueueItem<T> item = _pendingTasks.removeFirst();
+      _activeTasks.add(item);
+      item.onComplete = () {
+        _activeTasks.remove(item);
+        _processTask();
+      };
+      item.run();
+    } else {
+      _checkForCompletion();
+    }
+  }
+
+  void _checkForCompletion() {
+    if (_activeTasks.isEmpty && _pendingTasks.isEmpty) {
+      for (final Completer<void> completer in _completeListeners) {
+        if (!completer.isCompleted) {
+          completer.complete();
+        }
+      }
+      _completeListeners.clear();
+    }
+  }
+}
+
+class _TaskQueueItem<T> {
+  _TaskQueueItem(this._closure, this._completer, {this.onComplete});
+
+  final TaskQueueClosure<T> _closure;
+  final Completer<T> _completer;
+  void Function()? onComplete;
+
+  Future<void> run() async {
+    try {
+      _completer.complete(await _closure());
+    } catch (e) { // ignore: avoid_catches_without_on_clauses
+      _completer.completeError(e);
+    } finally {
+      onComplete?.call();
+    }
+  }
+}

packages/flutter_tools/lib/src/commands/update_packages.dart

@@ -13,6 +13,7 @@ import '../base/context.dart';
 import '../base/file_system.dart';
 import '../base/logger.dart';
 import '../base/net.dart';
+import '../base/task_queue.dart';
 import '../cache.dart';
 import '../dart/pub.dart';
 import '../globals_null_migrated.dart' as globals;
@@ -160,9 +161,6 @@ class UpdatePackagesCommand extends FlutterCommand {
   );
 
   Future<void> _downloadCoverageData() async {
-    final Status status = globals.logger.startProgress(
-      'Downloading lcov data for package:flutter...',
-    );
     final String urlBase = globals.platform.environment['FLUTTER_STORAGE_BASE_URL'] ?? 'https://storage.googleapis.com';
     final Uri coverageUri = Uri.parse('$urlBase/flutter_infra_release/flutter/coverage/lcov.info');
     final List<int> data = await _net.fetchUrl(coverageUri);
@@ -176,7 +174,6 @@ class UpdatePackagesCommand extends FlutterCommand {
     globals.fs.file(globals.fs.path.join(coverageDir, 'lcov.info'))
       ..createSync(recursive: true)
       ..writeAsBytesSync(data, flush: true);
-    status.stop();
   }
 
   @override
@@ -271,19 +268,24 @@ class UpdatePackagesCommand extends FlutterCommand {
       return FlutterCommandResult.success();
     }
 
-    if (upgrade || isPrintPaths || isPrintTransitiveClosure) {
-      globals.printStatus('Upgrading packages...');
+    final Map<String, PubspecDependency> dependencies = <String, PubspecDependency>{};
+    final bool doUpgrade = upgrade || isPrintPaths || isPrintTransitiveClosure;
+    if (doUpgrade) {
       // This feature attempts to collect all the packages used across all the
       // pubspec.yamls in the repo (including via transitive dependencies), and
       // find the latest version of each that can be used while keeping each
       // such package fixed at a single version across all the pubspec.yamls.
-      //
+      globals.printStatus('Upgrading packages...');
+    }
+
     // First, collect up the explicit dependencies:
     final List<PubspecYaml> pubspecs = <PubspecYaml>[];
-      final Map<String, PubspecDependency> dependencies = <String, PubspecDependency>{};
     final Set<String> specialDependencies = <String>{};
-      for (final Directory directory in packages) { // these are all the directories with pubspec.yamls we care about
+    // Visit all the directories with pubspec.yamls we care about.
+    for (final Directory directory in packages) {
+      if (doUpgrade) {
         globals.printTrace('Reading pubspec.yaml from: ${directory.path}');
+      }
       PubspecYaml pubspec;
       try {
         pubspec = PubspecYaml(directory); // this parses the pubspec.yaml
@@ -325,15 +327,23 @@ class UpdatePackagesCommand extends FlutterCommand {
     }
 
     // Now that we have all the dependencies we explicitly care about, we are
-      // going to create a fake package and then run "pub upgrade" on it. The
-      // pub tool will attempt to bring these dependencies up to the most recent
-      // possible versions while honoring all their constraints.
+    // going to create a fake package and then run either "pub upgrade" or "pub
+    // get" on it, depending on whether we are upgrading or not. If upgrading,
+    // the pub tool will attempt to bring these dependencies up to the most
+    // recent possible versions while honoring all their constraints. If not
+    // upgrading the pub tool will attempt to download any necessary package
+    // versions to the pub cache to warm the cache.
     final PubDependencyTree tree = PubDependencyTree(); // object to collect results
     final Directory tempDir = globals.fs.systemTempDirectory.createTempSync('flutter_update_packages.');
     try {
       final File fakePackage = _pubspecFor(tempDir);
       fakePackage.createSync();
-        fakePackage.writeAsStringSync(_generateFakePubspec(dependencies.values));
+      fakePackage.writeAsStringSync(
+        _generateFakePubspec(
+          dependencies.values,
+          useAnyVersion: doUpgrade,
+        ),
+      );
       // Create a synthetic flutter SDK so that transitive flutter SDK
       // constraints are not affected by this upgrade.
       Directory temporaryFlutterSdk;
@@ -346,11 +356,13 @@ class UpdatePackagesCommand extends FlutterCommand {
         );
       }
 
-        // Next we run "pub upgrade" on this generated package:
+      // Next we run "pub upgrade" on this generated package, if we're doing
+      // an upgrade. Otherwise, we just run a regular "pub get" on it in order
+      // to force the download of any needed packages to the pub cache.
       await pub.get(
         context: PubContext.updatePackages,
         directory: tempDir.path,
-          upgrade: true,
+        upgrade: doUpgrade,
         offline: offline,
         flutterRootOverride: upgrade
           ? temporaryFlutterSdk.path
@@ -364,11 +376,12 @@ class UpdatePackagesCommand extends FlutterCommand {
         // Failed to delete temporary SDK.
       }
 
-        // Then we run "pub deps --style=compact" on the result. We pipe all the
-        // output to tree.fill(), which parses it so that it can create a graph
-        // of all the dependencies so that we can figure out the transitive
-        // dependencies later. It also remembers which version was selected for
-        // each package.
+      if (doUpgrade) {
+        // If upgrading, we run "pub deps --style=compact" on the result. We
+        // pipe all the output to tree.fill(), which parses it so that it can
+        // create a graph of all the dependencies so that we can figure out the
+        // transitive dependencies later. It also remembers which version was
+        // selected for each package.
         await pub.batch(
           <String>['deps', '--style=compact'],
           context: PubContext.updatePackages,
@@ -376,10 +389,12 @@ class UpdatePackagesCommand extends FlutterCommand {
           filter: tree.fill,
           retry: false, // errors here are usually fatal since we're not hitting the network
         );
+      }
     } finally {
       tempDir.deleteSync(recursive: true);
     }
 
+    if (doUpgrade) {
       // The transitive dependency tree for the fake package does not contain
       // dependencies between Flutter SDK packages and pub packages. We add them
       // here.
@@ -429,20 +444,53 @@ class UpdatePackagesCommand extends FlutterCommand {
     final Stopwatch timer = Stopwatch()..start();
     int count = 0;
 
+    // Now we run pub get on each of the affected packages to update their
+    // pubspec.lock files with the right transitive dependencies.
+    //
+    // This can be expensive, so we run them in parallel. If we hadn't already
+    // warmed the cache above, running them in parallel could be dangerous due
+    // to contention when unpacking downloaded dependencies, but since we have
+    // downloaded all that we need, it is safe to run them in parallel.
+    final Status status = globals.logger.startProgress(
+      'Running "flutter pub get" in affected packages...',
+    );
+    try {
+      final TaskQueue<void> queue = TaskQueue<void>();
       for (final Directory dir in packages) {
+        unawaited(queue.add(() async {
+          final Stopwatch stopwatch = Stopwatch();
+          stopwatch.start();
           await pub.get(
             context: PubContext.updatePackages,
             directory: dir.path,
             offline: offline,
             generateSyntheticPackage: false,
+            printProgress: false,
           );
+          stopwatch.stop();
+          final double seconds = stopwatch.elapsedMilliseconds / 1000.0;
+          final String relativeDir = globals.fs.path.relative(dir.path, from: Cache.flutterRoot);
+          globals.printStatus('Ran pub get in $relativeDir in ${seconds.toStringAsFixed(1)}s...');
+        }));
         count += 1;
       }
-
+      unawaited(queue.add(() async {
+        final Stopwatch stopwatch = Stopwatch();
         await _downloadCoverageData();
+        stopwatch.stop();
+        final double seconds = stopwatch.elapsedMilliseconds / 1000.0;
+        globals.printStatus('Downloaded lcov data for package:flutter in ${seconds.toStringAsFixed(1)}s...');
+      }));
+      await queue.tasksComplete;
+      status?.stop();
+      // The exception is rethrown, so don't catch only Exceptions.
+    } catch (exception) { // ignore: avoid_catches_without_on_clauses
+      status?.cancel();
+      rethrow;
+    }
 
     final double seconds = timer.elapsedMilliseconds / 1000.0;
-    globals.printStatus("\nRan 'pub' $count time${count == 1 ? "" : "s"} and fetched coverage data in ${seconds.toStringAsFixed(1)}s.");
+    globals.printStatus("\nRan 'pub get' $count time${count == 1 ? "" : "s"} and fetched coverage data in ${seconds.toStringAsFixed(1)}s.");
 
     return FlutterCommandResult.success();
   }
@@ -1221,7 +1269,8 @@ class PubspecDependency extends PubspecLine {
 
   /// This generates the entry for this dependency for the pubspec.yaml for the
   /// fake package that we'll use to get the version numbers figured out.
-  void describeForFakePubspec(StringBuffer dependencies, StringBuffer overrides) {
+  void describeForFakePubspec(StringBuffer dependencies, StringBuffer overrides, { bool useAnyVersion = true}) {
+    final String versionToUse = useAnyVersion || version.isEmpty ? 'any' : version;
     switch (kind) {
       case DependencyKind.unknown:
       case DependencyKind.overridden:
@@ -1229,12 +1278,12 @@ class PubspecDependency extends PubspecLine {
         break;
       case DependencyKind.normal:
         if (!_kManuallyPinnedDependencies.containsKey(name)) {
-          dependencies.writeln('  $name: any');
+          dependencies.writeln('  $name: $versionToUse');
         }
         break;
       case DependencyKind.path:
         if (_lockIsOverride) {
-          dependencies.writeln('  $name: any');
+          dependencies.writeln('  $name: $versionToUse');
           overrides.writeln('  $name:');
           overrides.writeln('    path: $lockTarget');
         } else {
@@ -1244,7 +1293,7 @@ class PubspecDependency extends PubspecLine {
         break;
       case DependencyKind.sdk:
         if (_lockIsOverride) {
-          dependencies.writeln('  $name: any');
+          dependencies.writeln('  $name: $versionToUse');
           overrides.writeln('  $name:');
           overrides.writeln('    sdk: $lockTarget');
         } else {
@@ -1254,7 +1303,7 @@ class PubspecDependency extends PubspecLine {
         break;
       case DependencyKind.git:
         if (_lockIsOverride) {
-          dependencies.writeln('  $name: any');
+          dependencies.writeln('  $name: $versionToUse');
           overrides.writeln('  $name:');
           overrides.writeln(lockLine);
         } else {
@@ -1263,6 +1312,11 @@ class PubspecDependency extends PubspecLine {
         }
     }
   }
+
+  @override
+  String toString() {
+    return '$name: $version';
+  }
 }
 
 /// Generates the File object for the pubspec.yaml file of a given Directory.
@@ -1273,16 +1327,22 @@ File _pubspecFor(Directory directory) {
 
 /// Generates the source of a fake pubspec.yaml file given a list of
 /// dependencies.
-String _generateFakePubspec(Iterable<PubspecDependency> dependencies) {
+String _generateFakePubspec(
+  Iterable<PubspecDependency> dependencies, {
+  bool useAnyVersion = false
+}) {
   final StringBuffer result = StringBuffer();
   final StringBuffer overrides = StringBuffer();
+  final bool verbose = useAnyVersion;
   result.writeln('name: flutter_update_packages');
   result.writeln('environment:');
   result.writeln("  sdk: '>=2.10.0 <3.0.0'");
   result.writeln('dependencies:');
   overrides.writeln('dependency_overrides:');
   if (_kManuallyPinnedDependencies.isNotEmpty) {
+    if (verbose) {
       globals.printStatus('WARNING: the following packages use hard-coded version constraints:');
+    }
     final Set<String> allTransitive = <String>{
       for (final PubspecDependency dependency in dependencies)
         dependency.name,
@@ -1290,17 +1350,21 @@ String _generateFakePubspec(Iterable<PubspecDependency> dependencies) {
     for (final String package in _kManuallyPinnedDependencies.keys) {
       // Don't add pinned dependency if it is not in the set of all transitive dependencies.
       if (!allTransitive.contains(package)) {
+        if (verbose) {
           globals.printStatus('Skipping $package because it was not transitive');
+        }
         continue;
       }
       final String version = _kManuallyPinnedDependencies[package];
       result.writeln('  $package: $version');
+      if (verbose) {
         globals.printStatus('  - $package: $version');
       }
     }
+  }
   for (final PubspecDependency dependency in dependencies) {
     if (!dependency.pointsToSdk) {
-      dependency.describeForFakePubspec(result, overrides);
+      dependency.describeForFakePubspec(result, overrides, useAnyVersion: useAnyVersion);
     }
   }
   result.write(overrides.toString());

packages/flutter_tools/lib/src/dart/pub.dart

@@ -104,6 +104,7 @@ abstract class Pub {
     String flutterRootOverride,
     bool checkUpToDate = false,
     bool shouldSkipThirdPartyGenerator = true,
+    bool printProgress = true,
   });
 
   /// Runs pub in 'batch' mode.
@@ -179,6 +180,7 @@ class _DefaultPub implements Pub {
     String? flutterRootOverride,
     bool checkUpToDate = false,
     bool shouldSkipThirdPartyGenerator = true,
+    bool printProgress = true,
   }) async {
     directory ??= _fileSystem.currentDirectory.path;
     final File packageConfigFile = _fileSystem.file(
@@ -232,9 +234,9 @@ class _DefaultPub implements Pub {
     }
 
     final String command = upgrade ? 'upgrade' : 'get';
-    final Status status = _logger.startProgress(
+    final Status? status = printProgress ? _logger.startProgress(
       'Running "flutter pub $command" in ${_fileSystem.path.basename(directory)}...',
-    );
+    ) : null;
     final bool verbose = _logger.isVerbose;
     final List<String> args = <String>[
       if (verbose)
@@ -257,10 +259,10 @@ class _DefaultPub implements Pub {
         retry: !offline,
         flutterRootOverride: flutterRootOverride,
       );
-      status.stop();
+      status?.stop();
     // The exception is rethrown, so don't catch only Exceptions.
     } catch (exception) { // ignore: avoid_catches_without_on_clauses
-      status.cancel();
+      status?.cancel();
       rethrow;
     }
 

packages/flutter_tools/test/commands.shard/hermetic/drive_test.dart

@@ -119,5 +119,6 @@ class FakePub extends Fake implements Pub {
     String flutterRootOverride,
     bool checkUpToDate = false,
     bool shouldSkipThirdPartyGenerator = true,
+    bool printProgress = true,
   }) async { }
 }

packages/flutter_tools/test/commands.shard/hermetic/pub_get_test.dart

@@ -114,6 +114,7 @@ class FakePub extends Fake implements Pub {
     String flutterRootOverride,
     bool checkUpToDate = false,
     bool shouldSkipThirdPartyGenerator = true,
+    bool printProgress = true,
   }) async {
     fileSystem.currentDirectory
       .childDirectory('.dart_tool')

packages/flutter_tools/test/general.shard/base/task_queue_test.dart

+// Copyright 2014 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 'package:flutter_tools/src/base/task_queue.dart';
+
+import '../../src/common.dart';
+
+void main() {
+  group('TaskQueue', () {
+    /// A special test designed to check shared [TaskQueue]
+    /// behavior when exceptions occur after a delay in the passed closures to
+    /// [TaskQueue.add].
+    test('no deadlock when delayed exceptions fire in closures', () async {
+      final TaskQueue<void> sharedTracker = TaskQueue<void>(maxJobs: 2);
+      expect(() async {
+        final Future<void> t = Future<void>.delayed(const Duration(milliseconds: 10), () => throw TestException());
+        await sharedTracker.add(() => t);
+        return t;
+      }, throwsA(const TypeMatcher<TestException>()));
+      expect(() async {
+        final Future<void> t = Future<void>.delayed(const Duration(milliseconds: 10), () => throw TestException());
+        await sharedTracker.add(() => t);
+        return t;
+      }, throwsA(const TypeMatcher<TestException>()));
+      expect(() async {
+        final Future<void> t = Future<void>.delayed(const Duration(milliseconds: 10), () => throw TestException());
+        await sharedTracker.add(() => t);
+        return t;
+      }, throwsA(const TypeMatcher<TestException>()));
+      expect(() async {
+        final Future<void> t = Future<void>.delayed(const Duration(milliseconds: 10), () => throw TestException());
+        await sharedTracker.add(() => t);
+        return t;
+      }, throwsA(const TypeMatcher<TestException>()));
+
+      /// We deadlock here if the exception is not handled properly.
+      await sharedTracker.tasksComplete;
+    });
+
+    test('basic sequential processing works with no deadlock', () async {
+      final Set<int> completed = <int>{};
+      final TaskQueue<void> tracker = TaskQueue<void>(maxJobs: 1);
+      await tracker.add(() async => completed.add(1));
+      await tracker.add(() async => completed.add(2));
+      await tracker.add(() async => completed.add(3));
+      await tracker.tasksComplete;
+      expect(completed.length, equals(3));
+    });
+
+    test('basic sequential processing works on exceptions', () async {
+      final Set<int> completed = <int>{};
+      final TaskQueue<void> tracker = TaskQueue<void>(maxJobs: 1);
+      await tracker.add(() async => completed.add(0));
+      await tracker.add(() async => throw TestException()).catchError((Object _) {});
+      await tracker.add(() async => throw TestException()).catchError((Object _) {});
+      await tracker.add(() async => completed.add(3));
+      await tracker.tasksComplete;
+      expect(completed.length, equals(2));
+    });
+
+    /// Verify that if there are more exceptions than the maximum number
+    /// of in-flight [Future]s that there is no deadlock.
+    test('basic parallel processing works with no deadlock', () async {
+      final Set<int> completed = <int>{};
+      final TaskQueue<void> tracker = TaskQueue<void>(maxJobs: 10);
+      for (int i = 0; i < 100; i++) {
+        await tracker.add(() async => completed.add(i));
+      }
+      await tracker.tasksComplete;
+      expect(completed.length, equals(100));
+    });
+
+    test('basic parallel processing works on exceptions', () async {
+      final Set<int> completed = <int>{};
+      final TaskQueue<void> tracker = TaskQueue<void>(maxJobs: 10);
+      for (int i = 0; i < 50; i++) {
+        await tracker.add(() async => completed.add(i));
+      }
+      for (int i = 50; i < 65; i++) {
+        try {
+          await tracker.add(() async => throw TestException());
+        } on TestException {
+          // Ignore
+        }
+      }
+      for (int i = 65; i < 100; i++) {
+        await tracker.add(() async => completed.add(i));
+      }
+      await tracker.tasksComplete;
+      expect(completed.length, equals(85));
+    });
+  });
+}
+
+class TestException implements Exception {}
\ No newline at end of file

packages/flutter_tools/test/general.shard/cache_test.dart

@@ -1073,6 +1073,7 @@ class FakePub extends Fake implements Pub {
     String flutterRootOverride,
     bool checkUpToDate = false,
     bool shouldSkipThirdPartyGenerator = true,
+    bool printProgress = true,
   }) async {
     calledGet += 1;
   }

packages/flutter_tools/test/general.shard/runner/flutter_command_test.dart

@@ -723,5 +723,6 @@ class FakePub extends Fake implements Pub {
     String flutterRootOverride,
     bool checkUpToDate = false,
     bool shouldSkipThirdPartyGenerator = true,
+    bool printProgress = true,
   }) async { }
 }

packages/flutter_tools/test/src/throwing_pub.dart

@@ -31,6 +31,7 @@ class ThrowingPub implements Pub {
     String? flutterRootOverride,
     bool checkUpToDate = false,
     bool shouldSkipThirdPartyGenerator = true,
+    bool printProgress = true,
   }) {
     throw UnsupportedError('Attempted to invoke pub during test.');
   }