Parallel fetching of available versions (#2280)

Implements rate-limited parallel fetching of version information when solving.
Also does speculative pre-fetching of version information of dependencies of the newest versions of each package.

In informal benchmarks of pub get for a package whose only direct dependency is package:test the resolution time goes from 7.8 seconds to 1.5 seconds.

Also adds package:pedantic to our direct runtime dependencies

Author: sigurdm

doc/repository-spec-v2.md

@@ -62,7 +62,9 @@ repository.
 }
 ```
 
-### Inspect a specific version of a package
+### (Deprecated) Inspect a specific version of a package
+
+**Deprecated** as of Dart 2.8, use "List all versions of a package" instead.
 
 **GET** `<PUB_HOSTED_URL>/api/packages/<PACKAGE>/versions/<VERSION>`
 
@@ -99,4 +101,3 @@ MUST support redirects.
 
 The API for authenticating and publishing packages is not formalized yet, see
 [#1381](https://github.com/dart-lang/pub/issues/1381).
-

lib/src/rate_limited_scheduler.dart

@@ -0,0 +1,142 @@
+// Copyright (c) 2019, the Dart project authors.  Please see the AUTHORS file
+// for details. 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 'package:pool/pool.dart';
+import 'package:pedantic/pedantic.dart';
+
+/// Handles rate-limited scheduling of tasks.
+///
+/// Tasks are identified by a jobId of type [J] (should be useful as a Hash-key)
+/// and run with a supplied async function.
+///
+/// Designed to allow speculatively running tasks that will likely be needed
+/// later with [withPrescheduling].
+///
+/// Errors thrown by tasks scheduled with the `preschedule` callback will only
+/// be triggered when you await the [Future] returned by [schedule].
+///
+/// The operation will run in the [Zone] that the task was in when enqueued.
+///
+/// If a task if [preschedule]d and later [schedule]d before the operation is
+/// started, the task will go in front of the queue with the zone of the
+/// [schedule] operation.
+///
+/// Example:
+///
+/// ```dart
+/// // A scheduler that, given a uri, gets that page and returns the body
+/// final scheduler = RateLimitedScheduler(http.read);
+///
+/// scheduler.withPresceduling((preschedule) {
+///   // Start fetching `pub.dev` and `dart.dev` in the background.
+///   scheduler.preschedule(Uri.parse('https://pub.dev/'));
+///   scheduler.preschedule(Uri.parse('https://dart.dev/'));
+///   // ... do time-consuming task.
+///   // Now we actually need `pub.dev`.
+///   final pubDevBody =
+///       await scheduler.schedule(Uri.parse('https://pub.dev/'));
+///   // if the `dart.dev` task has not started yet, it will be canceled when
+///   // leaving `withPresceduling`.
+/// });
+/// ```
+class RateLimitedScheduler<J, V> {
+  final Future<V> Function(J) _runJob;
+
+  /// The results of ongoing and finished jobs.
+  final Map<J, Completer<V>> _cache = <J, Completer<V>>{};
+
+  /// Tasks that are waiting to be run.
+  final Queue<_Task<J>> _queue = Queue<_Task<J>>();
+
+  /// Rate limits the number of concurrent jobs.
+  final Pool _pool;
+
+  /// Jobs that have started running.
+  final Set<J> _started = {};
+
+  RateLimitedScheduler(Future<V> Function(J) runJob,
+      {maxConcurrentOperations = 10})
+      : _runJob = runJob,
+        _pool = Pool(maxConcurrentOperations);
+
+  /// Pick the next task in [_queue] and run it.
+  ///
+  /// If the task is already in [_started] it will not be run again.
+  Future<void> _processNextTask() async {
+    if (_queue.isEmpty) {
+      return;
+    }
+    final task = _queue.removeFirst();
+    final completer = _cache[task.jobId];
+
+    if (!_started.add(task.jobId)) {
+      return;
+    }
+
+    // Use an async function to catch sync exceptions from _runJob.
+    Future<V> runJob() async {
+      return await task.zone.runUnary(_runJob, task.jobId);
+    }
+
+    completer.complete(runJob());
+    // Listen to errors on the completer:
+    // this will make errors thrown by [_run] not
+    // become uncaught.
+    //
+    // They will still show up for other listeners of the future.
+    await completer.future.catchError((_) {});
+  }
+
+  /// Calls [callback] with a function that can pre-schedule jobs.
+  ///
+  /// When [callback] returns, all jobs that where prescheduled by [callback]
+  /// that have not started running will be removed from the work queue
+  /// (if they have been added seperately by [schedule] they will still be
+  /// executed).
+  Future<R> withPrescheduling<R>(
+    FutureOr<R> Function(void Function(J) preschedule) callback,
+  ) async {
+    final prescheduled = <_Task>{};
+    try {
+      return await callback((jobId) {
+        if (_started.contains(jobId)) return;
+        final task = _Task(jobId, Zone.current);
+        _cache.putIfAbsent(jobId, () => Completer());
+        _queue.addLast(task);
+        prescheduled.add(task);
+
+        unawaited(_pool.withResource(_processNextTask));
+      });
+    } finally {
+      _queue.removeWhere(prescheduled.contains);
+    }
+  }
+
+  /// Returns a future that completed with the result of running [jobId].
+  ///
+  /// If [jobId] has already run, the cached result will be returned.
+  /// If [jobId] is not yet running, it will go to the front of the work queue
+  /// to be scheduled next when there are free resources.
+  Future<V> schedule(J jobId) {
+    final completer = _cache.putIfAbsent(jobId, () => Completer());
+    if (!_started.contains(jobId)) {
+      final task = _Task(jobId, Zone.current);
+      _queue.addFirst(task);
+      scheduleMicrotask(() => _pool.withResource(_processNextTask));
+    }
+    return completer.future;
+  }
+}
+
+class _Task<J> {
+  final J jobId;
+  final Zone zone;
+  _Task(this.jobId, this.zone);
+
+  @override
+  String toString() => jobId.toString();
+}

lib/src/solver/version_solver.dart

@@ -91,13 +91,15 @@ class VersionSolver {
         [Term(PackageRange.root(_root), false)], IncompatibilityCause.root));
 
     try {
-      var next = _root.name;
-      while (next != null) {
-        _propagate(next);
-        next = await _choosePackageVersion();
-      }
+      return await _systemCache.hosted.withPrefetching(() async {
+        var next = _root.name;
+        while (next != null) {
+          _propagate(next);
+          next = await _choosePackageVersion();
+        }
 
-      return await _result();
+        return await _result();
+      });
     } finally {
       // Gather some solving metrics.
       log.solver('Version solving took ${stopwatch.elapsed} seconds.\n'

lib/src/source/git.dart

@@ -6,6 +6,7 @@ import 'dart:async';
 import 'dart:io';
 
 import 'package:path/path.dart' as p;
+import 'package:pool/pool.dart';
 import 'package:pub_semver/pub_semver.dart';
 
 import '../git.dart' as git;
@@ -189,6 +190,10 @@ class GitSource extends Source {
 
 /// The [BoundSource] for [GitSource].
 class BoundGitSource extends CachedSource {
+  /// Limit the number of concurrent git operations to 1.
+  // TODO(sigurdm): Use RateLimitedScheduler.
+  final Pool _pool = Pool(1);
+
   @override
   final GitSource source;
 
@@ -221,27 +226,31 @@ class BoundGitSource extends CachedSource {
 
   @override
   Future<List<PackageId>> doGetVersions(PackageRef ref) async {
-    await _ensureRepoCache(ref);
-    var path = _repoCachePath(ref);
-    var revision = await _firstRevision(path, ref.description['ref']);
-    var pubspec =
-        await _describeUncached(ref, revision, ref.description['path']);
-
-    return [
-      PackageId(ref.name, source, pubspec.version, {
-        'url': ref.description['url'],
-        'ref': ref.description['ref'],
-        'resolved-ref': revision,
-        'path': ref.description['path']
-      })
-    ];
+    return await _pool.withResource(() async {
+      await _ensureRepoCache(ref);
+      var path = _repoCachePath(ref);
+      var revision = await _firstRevision(path, ref.description['ref']);
+      var pubspec =
+          await _describeUncached(ref, revision, ref.description['path']);
+
+      return [
+        PackageId(ref.name, source, pubspec.version, {
+          'url': ref.description['url'],
+          'ref': ref.description['ref'],
+          'resolved-ref': revision,
+          'path': ref.description['path']
+        })
+      ];
+    });
   }
 
   /// Since we don't have an easy way to read from a remote Git repo, this
   /// just installs [id] into the system cache, then describes it from there.
   @override
-  Future<Pubspec> describeUncached(PackageId id) => _describeUncached(
-      id.toRef(), id.description['resolved-ref'], id.description['path']);
+  Future<Pubspec> describeUncached(PackageId id) {
+    return _pool.withResource(() => _describeUncached(
+        id.toRef(), id.description['resolved-ref'], id.description['path']));
+  }
 
   /// Like [describeUncached], but takes a separate [ref] and Git [revision]
   /// rather than a single ID.
@@ -284,28 +293,32 @@ class BoundGitSource extends CachedSource {
   /// in `cache/`.
   @override
   Future<Package> downloadToSystemCache(PackageId id) async {
-    var ref = id.toRef();
-    if (!git.isInstalled) {
-      fail("Cannot get ${id.name} from Git (${ref.description['url']}).\n"
-          'Please ensure Git is correctly installed.');
-    }
-
-    ensureDir(p.join(systemCacheRoot, 'cache'));
-    await _ensureRevision(ref, id.description['resolved-ref']);
-
-    var revisionCachePath = _revisionCachePath(id);
-    await _revisionCacheClones.putIfAbsent(revisionCachePath, () async {
-      if (!entryExists(revisionCachePath)) {
-        await _clone(_repoCachePath(ref), revisionCachePath);
-        await _checkOut(revisionCachePath, id.description['resolved-ref']);
-        _writePackageList(revisionCachePath, [id.description['path']]);
-      } else {
-        _updatePackageList(revisionCachePath, id.description['path']);
+    return await _pool.withResource(() async {
+      var ref = id.toRef();
+      if (!git.isInstalled) {
+        fail("Cannot get ${id.name} from Git (${ref.description['url']}).\n"
+            'Please ensure Git is correctly installed.');
       }
-    });
 
-    return Package.load(id.name,
-        p.join(revisionCachePath, id.description['path']), systemCache.sources);
+      ensureDir(p.join(systemCacheRoot, 'cache'));
+      await _ensureRevision(ref, id.description['resolved-ref']);
+
+      var revisionCachePath = _revisionCachePath(id);
+      await _revisionCacheClones.putIfAbsent(revisionCachePath, () async {
+        if (!entryExists(revisionCachePath)) {
+          await _clone(_repoCachePath(ref), revisionCachePath);
+          await _checkOut(revisionCachePath, id.description['resolved-ref']);
+          _writePackageList(revisionCachePath, [id.description['path']]);
+        } else {
+          _updatePackageList(revisionCachePath, id.description['path']);
+        }
+      });
+
+      return Package.load(
+          id.name,
+          p.join(revisionCachePath, id.description['path']),
+          systemCache.sources);
+    });
   }
 
   /// Returns the path to the revision-specific cache of [id].