Construct PackageGraphs and run tools asynchronously (#1849)

* asynchronous PackageGraph construction and precaching

* Run tools more asynchronously

* Enhacements and dropping of Future.wait

* dartfmt

* Rate-limit the number of tools in flight

* dartfmt and review comments

Author: jcollins-g

analysis_options.yaml

@@ -17,4 +17,5 @@ linter:
     - package_api_docs
     - slash_for_doc_comments
     - prefer_final_fields
+    - unawaited_futures
 #    - unnecessary_brace_in_string_interps

lib/src/dartdoc_options.dart

@@ -20,6 +20,7 @@ import 'dart:io';
 import 'package:analyzer/dart/element/element.dart';
 import 'package:args/args.dart';
 import 'package:dartdoc/dartdoc.dart';
+import 'package:dartdoc/src/tuple.dart';
 import 'package:path/path.dart' as pathLib;
 import 'package:yaml/yaml.dart';
 
@@ -226,22 +227,23 @@ class DartToolDefinition extends ToolDefinition {
   /// to run. If no snapshot file existed, then create one and modify the args
   /// so that if they are executed with dart, will result in the snapshot being
   /// built.
-  String createSnapshotIfNeeded(List<String> args) {
-    assert(ToolDefinition.isDartExecutable(args[0]));
-    // Generate a new snapshot, if needed, and use the first run as the training
+  Future<Tuple2<String, Function()>> modifyArgsToCreateSnapshotIfNeeded(
+      List<String> args) async {
+    assert(args[0] == command.first);
+    // Set up flags to create a new snapshot, if needed, and use the first run as the training
     // run.
-    File snapshotPath = _snapshotPath;
-    snapshotPath ??= SnapshotCache.instance.getSnapshot(args[0]);
-    if (snapshotPath.existsSync()) {
+    File snapshotFile = await getSnapshotFile();
+    if (snapshotFile.existsSync()) {
       // replace the first argument with the path to the snapshot.
-      args[0] = snapshotPath.absolute.path;
+      args[0] = snapshotFile.absolute.path;
     } else {
       args.insertAll(0, [
-        '--snapshot=${snapshotPath.absolute.path}',
+        '--snapshot=${snapshotFile.absolute.path}',
         '--snapshot_kind=app-jit'
       ]);
     }
-    return Platform.resolvedExecutable;
+    return new Tuple2(Platform.resolvedExecutable,
+        _snapshotCompleter.isCompleted ? null : _snapshotCompleter.complete);
   }
 
   DartToolDefinition(
@@ -250,11 +252,23 @@ class DartToolDefinition extends ToolDefinition {
     // If the dart tool is already a snapshot, then we just use that.
     if (command[0].endsWith('.snapshot')) {
       _snapshotPath = File(command[0]);
+      _snapshotCompleter.complete();
     }
   }
 
+  final Completer _snapshotCompleter = new Completer();
+
   /// If the tool has a pre-built snapshot, it will be stored here.
   File _snapshotPath;
+
+  Future<File> getSnapshotFile() async {
+    if (_snapshotPath == null) {
+      _snapshotPath = SnapshotCache.instance.getSnapshot(command.first);
+    } else {
+      await _snapshotCompleter.future;
+    }
+    return _snapshotPath;
+  }
 }
 
 /// A configuration class that can interpret [ToolDefinition]s from a YAML map.

lib/src/io_utils.dart

@@ -75,25 +75,25 @@ class MultiFutureTracker<T> {
 
   MultiFutureTracker(this.parallel);
 
-  /// Adds a Future to the queue of outstanding Futures, and returns a Future
-  /// that completes only when the number of Futures outstanding is < [parallel]
-  /// (and so it is OK to start another).
-  ///
-  /// That can be extremely brief and there's no longer a guarantee after that
-  /// point that another async task has not added a Future to the list.
-  Future<void> addFuture(Future<T> future) async {
-    _queue.add(future);
-    future.then((f) => _queue.remove(future));
-    await _waitUntil(parallel - 1);
-  }
-
   /// Wait until fewer or equal to this many Futures are outstanding.
   Future<void> _waitUntil(int max) async {
     while (_queue.length > max) {
       await Future.any(_queue);
     }
   }
 
+  /// Generates a [Future] from the given closure and adds it to the queue,
+  /// once the queue is sufficiently empty.
+  Future<void> addFutureFromClosure(Future<T> Function() closure) async {
+    while (_queue.length > parallel - 1) {
+      await Future.any(_queue);
+    }
+    Future future = closure();
+    _queue.add(future);
+    // ignore: unawaited_futures
+    future.then((f) => _queue.remove(future));
+  }
+
   /// Wait until all futures added so far have completed.
   Future<void> wait() async => await _waitUntil(0);
 }