Use AnalysisDriverScheduler to schedule work across multiple AnalysisDriver(s).

pkg/analyzer/lib/src/dart/analysis/driver.dart

 // Copyright (c) 2016, 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 'dart:typed_data';
 
 import 'package:analyzer/dart/ast/ast.dart';
 import 'package:analyzer/error/error.dart';
 import 'package:analyzer/error/listener.dart';
 import 'package:analyzer/file_system/file_system.dart';
 import 'package:analyzer/src/context/context.dart';
 import 'package:analyzer/src/dart/analysis/byte_store.dart';
 import 'package:analyzer/src/dart/analysis/file_state.dart';
 import 'package:analyzer/src/generated/engine.dart'
     show AnalysisContext, AnalysisEngine, AnalysisOptions, ChangeSet;
 import 'package:analyzer/src/generated/source.dart';
 import 'package:analyzer/src/summary/api_signature.dart';
 import 'package:analyzer/src/summary/format.dart';
 import 'package:analyzer/src/summary/idl.dart';
 import 'package:analyzer/src/summary/link.dart';
 import 'package:analyzer/src/summary/package_bundle_reader.dart';
 import 'package:analyzer/src/summary/summarize_elements.dart';
+import 'package:meta/meta.dart';
 
 /**
  * This class computes [AnalysisResult]s for Dart files.
  *
  * Let the set of "explicitly analyzed files" denote the set of paths that have
  * been passed to [addFile] but not subsequently passed to [removeFile]. Let
  * the "current analysis results" denote the map from the set of explicitly
  * analyzed files to the most recent [AnalysisResult] delivered to [results]
  * for each file. Let the "current file state" represent a map from file path
  * to the file contents most recently read from that file, or fetched from the
@@ skipping 26 matching lines @@
  * TODO(scheglov) Clean up the list of implicitly analyzed files.
  *
  * TODO(scheglov) Handle not existing 'dart:x' URIs (while user is typing).
  */
 class AnalysisDriver {
   /**
    * The version of data format, should be incremented on every format change.
    */
   static const int DATA_VERSION = 1;
 
+  /**
+   * The name of the driver, e.g. the name of the folder.
+   */
   String name;
+
+  /**
+   * The scheduler that schedules analysis work in this, and possible other

[Paul Berry, 2016/11/08 22:43:26]

s/possible/possibly/

[scheglov, 2016/11/09 01:54:50]

Done.

+   * analysis drivers.
+   */
+  final AnalysisDriverScheduler _scheduler;
+
+  /**
+   * The logger to write performed operations and performance to.
+   */
   final PerformanceLog _logger;
 
   /**
    * The resource provider for working with files.
    */
   final ResourceProvider _resourceProvider;
 
   /**
    * The byte storage to get and put serialized data.
    *
@@ skipping 60 matching lines @@
    * The set of files that are currently scheduled for analysis.
    */
   final _filesToAnalyze = new LinkedHashSet<String>();
 
   /**
    * Mapping from library URIs to the dependency signature of the library.
    */
   final _dependencySignatureMap = <Uri, String>{};
 
   /**
-   * The monitor that is signalled when there is work to do.
+   * The controller for the [results] stream.
    */
-  final _Monitor _hasWork = new _Monitor();
+  final _resultController = new StreamController<AnalysisResult>();
 
   /**
    * The controller for the [status] stream.
    */
   final _statusController = new StreamController<AnalysisStatus>();
 
   /**
    * The last status sent to the [status] stream.
    */
   AnalysisStatus _currentStatus = AnalysisStatus.IDLE;
 
   /**
    * Create a new instance of [AnalysisDriver].
    *
    * The given [SourceFactory] is cloned to ensure that it does not contain a
    * reference to a [AnalysisContext] in which it could have been used.
    */
-  AnalysisDriver(this._logger, this._resourceProvider, this._byteStore,
-      this._contentOverlay, SourceFactory sourceFactory, this._analysisOptions)
+  AnalysisDriver(
+      this._scheduler,
+      this._logger,
+      this._resourceProvider,
+      this._byteStore,
+      this._contentOverlay,
+      SourceFactory sourceFactory,
+      this._analysisOptions)
       : _sourceFactory = sourceFactory.clone() {
     _fillSalt();
     _sdkBundle = sourceFactory.dartSdk.getLinkedBundle();
     _fsState = new FileSystemState(_logger, _byteStore, _contentOverlay,
         _resourceProvider, _sourceFactory, _analysisOptions, _salt);
+    _scheduler._add(this);
   }
 
   /**
    * Set the list of files that the driver should try to analyze sooner.
    *
    * Every path in the list must be absolute and normalized.
    *
    * The driver will produce the results through the [results] stream. The
    * exact order in which results are produced is not defined, neither
    * between priority files, nor between priority and non-priority files.
    */
   void set priorityFiles(List<String> priorityPaths) {
     _priorityFiles.clear();
     _priorityFiles.addAll(priorityPaths);
     _transitionToAnalyzing();
-    _hasWork.notify();
+    _scheduler._notify(this);
   }
 
   /**
    * Return the [Stream] that produces [AnalysisResult]s for added files.
    *
-   * Analysis starts when the client starts listening to the stream, and stops
-   * when the client cancels the subscription. Note that the stream supports
-   * only one single subscriber.
+   * Note that the stream supports only one single subscriber.
    *
-   * When the client starts listening, the analysis state transitions to
-   * "analyzing" and an analysis result is produced for every added file prior
-   * to the next time the analysis state transitions to "idle".
+   * Analysis starts when the [AnalysisDriverScheduler] is started and the
+   * driver is added to it. The analysis state transitions to "analyzing" and
+   * an analysis result is produced for every added file prior to the next time
+   * the analysis state transitions to "idle".
    *
    * At least one analysis result is produced for every file passed to
    * [addFile] or [changeFile] prior to the next time the analysis state
    * transitions to "idle", unless the file is later removed from analysis
    * using [removeFile]. Analysis results for other files are produced only if
    * the changes affect analysis results of other files.
    *
    * More than one result might be produced for the same file, even if the
    * client does not change the state of the files.
    *
    * Results might be produced even for files that have never been added
    * using [addFile], for example when [getResult] was called for a file.
    */
-  Stream<AnalysisResult> get results async* {
-    try {
-      PerformanceLogSection analysisSection = null;
-      while (true) {
-        // Pump the event queue to allow IO and other asynchronous data
-        // processing while analysis is active. For example Analysis Server
-        // needs to be able to process `updateContent` or `setPriorityFiles`
-        // requests while background analysis is in progress.
-        //
-        // The number of pumpings is arbitrary, might be changed if we see that
-        // analysis or other data processing tasks are starving. Ideally we
-        // would need to be able to set priority of (continuous) asynchronous
-        // tasks.
-        await _pumpEventQueue(128);
-
-        await _hasWork.signal;
-
-        if (analysisSection == null) {
-          analysisSection = _logger.enter('Analyzing');
-        }
-
-        // Verify all changed files one at a time.
-        if (_changedFiles.isNotEmpty) {
-          String path = _removeFirst(_changedFiles);
-          _verifyApiSignature(path);
-          // Repeat the processing loop.
-          _hasWork.notify();
-          continue;
-        }
-
-        // Analyze a requested file.
-        if (_requestedFiles.isNotEmpty) {
-          String path = _requestedFiles.keys.first;
-          AnalysisResult result = _computeAnalysisResult(path, withUnit: true);
-          // Notify the completers.
-          _requestedFiles.remove(path).forEach((completer) {
-            completer.complete(result);
-          });
-          // Remove from to be analyzed and produce it now.
-          _filesToAnalyze.remove(path);
-          yield result;
-          // Repeat the processing loop.
-          _hasWork.notify();
-          continue;
-        }
-
-        // Analyze a priority file.
-        if (_priorityFiles.isNotEmpty) {
-          bool analyzed = false;
-          for (String path in _priorityFiles) {
-            if (_filesToAnalyze.remove(path)) {
-              analyzed = true;
-              AnalysisResult result =
-                  _computeAnalysisResult(path, withUnit: true);
-              yield result;
-              break;
-            }
-          }
-          // Repeat the processing loop.
-          if (analyzed) {
-            _hasWork.notify();
-            continue;
-          }
-        }
-
-        // Analyze a general file.
-        if (_filesToAnalyze.isNotEmpty) {
-          String path = _removeFirst(_filesToAnalyze);
-          AnalysisResult result = _computeAnalysisResult(path, withUnit: false);
-          yield result;
-          // Repeat the processing loop.
-          _hasWork.notify();
-          continue;
-        }
-
-        // There is nothing to do.
-        analysisSection.exit();
-        analysisSection = null;
-        _transitionToIdle();
-      }
-    } finally {
-      print('The stream was cancelled.');
-    }
-  }
+  Stream<AnalysisResult> get results => _resultController.stream;
 
   /**
    * Return the stream that produces [AnalysisStatus] events.
    */
   Stream<AnalysisStatus> get status => _statusController.stream;
 
   /**
+   * Return the priority of work that the driver needs to perform.
+   */
+  AnalysisDriverPriority get _workPriority {
+    if (_requestedFiles.isNotEmpty) {
+      return AnalysisDriverPriority.interactive;
+    }
+    if (_priorityFiles.isNotEmpty) {
+      for (String path in _priorityFiles) {
+        if (_filesToAnalyze.contains(path)) {
+          return AnalysisDriverPriority.priority;
+        }
+      }
+    }
+    if (_filesToAnalyze.isNotEmpty) {
+      return AnalysisDriverPriority.general;
+    }
+    if (_changedFiles.isNotEmpty) {
+      return AnalysisDriverPriority.general;
+    }
+    _transitionToIdle();
+    return AnalysisDriverPriority.nothing;
+  }
+
+  /**
    * Add the file with the given [path] to the set of files to analyze.
    *
    * The [path] must be absolute and normalized.
    *
    * The results of analysis are eventually produced by the [results] stream.
    */
   void addFile(String path) {
     if (AnalysisEngine.isDartFileName(path)) {
       _explicitFiles.add(path);
       _filesToAnalyze.add(path);
     }
     _transitionToAnalyzing();
-    _hasWork.notify();
+    _scheduler._notify(this);
   }
 
   /**
    * The file with the given [path] might have changed - updated, added or
    * removed. Or not, we don't know. Or it might have, but then changed back.
    *
    * The [path] must be absolute and normalized.
    *
    * The [path] can be any file - explicitly or implicitly analyzed, or neither.
    *
    * Causes the analysis state to transition to "analyzing" (if it is not in
    * that state already). Schedules the file contents for [path] to be read
    * into the current file state prior to the next time the analysis state
    * transitions to "idle".
    *
    * Invocation of this method will not prevent a [Future] returned from
    * [getResult] from completing with a result, but the result is not
    * guaranteed to be consistent with the new current file state after this
    * [changeFile] invocation.
    */
   void changeFile(String path) {
     if (AnalysisEngine.isDartFileName(path)) {
       _changedFiles.add(path);
       if (_explicitFiles.contains(path)) {
         _filesToAnalyze.add(path);
       }
     }
     _transitionToAnalyzing();
-    _hasWork.notify();
+    _scheduler._notify(this);
   }
 
   /**
+   * Notify the driver that the client is going to stop using it.
+   */
+  void dispose() {
+    _scheduler._remove(this);
+  }
+
+  /**
    * Return the [Future] that completes with a [AnalysisResult] for the file
    * with the given [path].
    *
    * The [path] must be absolute and normalized.
    *
    * The [path] can be any file - explicitly or implicitly analyzed, or neither.
    *
    * Causes the analysis state to transition to "analyzing" (if it is not in
    * that state already), the driver will read the file and produce the analysis
    * result for it, which is consistent with the current file state (including
    * the new state of the file), prior to the next time the analysis state
    * transitions to "idle".
    */
   Future<AnalysisResult> getResult(String path) {
     var completer = new Completer<AnalysisResult>();
     _requestedFiles
         .putIfAbsent(path, () => <Completer<AnalysisResult>>[])
         .add(completer);
     _transitionToAnalyzing();
-    _hasWork.notify();
+    _scheduler._notify(this);
     return completer.future;
   }
 
   /**
    * Return `true` if the file with the given [path] was explicitly added
    * to analysis using [addFile].
    */
   bool isAddedFile(String path) {
     return _explicitFiles.contains(path);
   }
@@ skipping 282 matching lines @@
       ApiSignature signature = new ApiSignature();
       signature.addUint32List(_salt);
       signature.addString(dependencyHash);
       signature.addString(file.contentHash);
       return '${signature.toHex()}.resolved';
     }
     return null;
   }
 
   /**
+   * Perform a single chunk of work and produce [results].
+   */
+  Future<Null> _performWork() async {
+    // Verify all changed files one at a time.
+    if (_changedFiles.isNotEmpty) {
+      String path = _removeFirst(_changedFiles);
+      _verifyApiSignature(path);
+      return;
+    }
+
+    // Analyze a requested file.
+    if (_requestedFiles.isNotEmpty) {
+      String path = _requestedFiles.keys.first;
+      AnalysisResult result = _computeAnalysisResult(path, withUnit: true);
+      // Notify the completers.
+      _requestedFiles.remove(path).forEach((completer) {
+        completer.complete(result);
+      });
+      // Remove from to be analyzed and produce it now.
+      _filesToAnalyze.remove(path);
+      _resultController.add(result);
+      return;
+    }
+
+    // Analyze a priority file.
+    if (_priorityFiles.isNotEmpty) {
+      for (String path in _priorityFiles) {
+        if (_filesToAnalyze.remove(path)) {
+          AnalysisResult result = _computeAnalysisResult(path, withUnit: true);
+          _resultController.add(result);
+          return;
+        }
+      }
+    }
+
+    // Analyze a general file.
+    if (_filesToAnalyze.isNotEmpty) {
+      String path = _removeFirst(_filesToAnalyze);
+      AnalysisResult result = _computeAnalysisResult(path, withUnit: false);
+      _resultController.add(result);
+      return;
+    }
+  }
+
+  /**
    * Send a notifications to the [status] stream that the driver started
    * analyzing.
    */
   void _transitionToAnalyzing() {
     if (_currentStatus != AnalysisStatus.ANALYZING) {
       _currentStatus = AnalysisStatus.ANALYZING;
       _statusController.add(AnalysisStatus.ANALYZING);
     }
   }
 
@@ skipping 26 matching lines @@
       if (anyApiChanged) {
         _logger.writeln('API signatures mismatch found for $path');
         _dependencySignatureMap.clear();
         _filesToAnalyze.addAll(_explicitFiles);
       }
       return files[0];
     });
   }
 
   /**
+   * Remove and return the first item in the given [set].
+   */
+  static Object/*=T*/ _removeFirst/*<T>*/(LinkedHashSet<Object/*=T*/ > set) {
+    Object/*=T*/ element = set.first;
+    set.remove(element);
+    return element;
+  }
+}
+
+/**
+ * Priorities of [AnalysisDriver] work. The closer a priority to the beginning
+ * of the list, the earlier the corresponding [AnalysisDriver] should be asked
+ * to perform work.
+ */
+@visibleForTesting
+enum AnalysisDriverPriority { interactive, priority, general, nothing }
+
+/**
+ * Instances of this class schedule work in multiple [AnalysisDriver]s so that
+ * work with the lowest priority is performed first.
+ */
+class AnalysisDriverScheduler {
+  final PerformanceLog _logger;
+  final List<AnalysisDriver> _drivers = [];
+  final _Monitor _hasWork = new _Monitor();
+
+  bool _started = false;
+
+  AnalysisDriverScheduler(this._logger);
+
+  /**
+   * Start the scheduler, so that any [AnalysisDriver] created before or
+   * after will be asked to perform work.
+   */
+  void start() {
+    if (_started) {
+      throw new StateError('The scheduler has already been started.');
+    }
+    _started = true;
+    _run();
+  }
+
+  /**
+   * Add the given [driver] and schedule it to perform its work.
+   */
+  void _add(AnalysisDriver driver) {
+    _drivers.add(driver);
+    _hasWork.notify();
+  }
+
+  /**
+   * Notify that there is a change to the [driver], it it might need to
+   * perform some work.
+   */
+  void _notify(AnalysisDriver driver) {
+    _hasWork.notify();
+  }
+
+  /**
+   * Remove the given [driver] from the scheduler, so that it will not be
+   * asked to perform any new work.
+   */
+  void _remove(AnalysisDriver driver) {
+    _drivers.remove(driver);
+    _hasWork.notify();
+  }
+
+  /**
+   * Run infinitely analysis cycle, selecting the drivers with the lowest
+   * priority first.
+   */
+  Future<Null> _run() async {
+    PerformanceLogSection analysisSection;
+    while (true) {
+      // Pump the event queue to allow IO and other asynchronous data
+      // processing while analysis is active. For example Analysis Server
+      // needs to be able to process `updateContent` or `setPriorityFiles`
+      // requests while background analysis is in progress.
+      //
+      // The number of pumpings is arbitrary, might be changed if we see that
+      // analysis or other data processing tasks are starving. Ideally we
+      // would need to be able to set priority of (continuous) asynchronous
+      // tasks.
+      await _pumpEventQueue(128);
+
+      await _hasWork.signal;
+
+      if (analysisSection == null) {
+        analysisSection = _logger.enter('Analyzing');
+      }
+
+      // Find the driver with the most priority work.

[Paul Berry, 2016/11/08 22:43:26]

s/most/highest/

[scheglov, 2016/11/09 01:54:50]

Done.

+      AnalysisDriver bestDriver;
+      AnalysisDriverPriority bestDriverPriority;
+      for (AnalysisDriver driver in _drivers) {
+        AnalysisDriverPriority priority = driver._workPriority;
+        if (bestDriverPriority == null ||
+            priority.index < bestDriverPriority.index) {
+          bestDriver = driver;
+          bestDriverPriority = priority;
+        }
+      }
+
+      // Continue to sleeping if no work to do.
+      if (bestDriverPriority == null ||
+          bestDriverPriority == AnalysisDriverPriority.nothing) {
+        analysisSection.exit();
+        analysisSection = null;
+        continue;
+      }
+
+      // Ask the driver to perform a chunk of work.
+      await bestDriver._performWork();
+
+      // Schedule one more cycle.
+      _hasWork.notify();
+    }
+  }
+
+  /**
    * Returns a [Future] that completes after performing [times] pumpings of
    * the event queue.
    */
   static Future _pumpEventQueue(int times) {
     if (times == 0) {
       return new Future.value();
     }
     return new Future.delayed(Duration.ZERO, () => _pumpEventQueue(times - 1));
   }
-
-  /**
-   * Remove and return the first item in the given [set].
-   */
-  static Object/*=T*/ _removeFirst/*<T>*/(LinkedHashSet<Object/*=T*/ > set) {
-    Object/*=T*/ element = set.first;
-    set.remove(element);
-    return element;
-  }
 }
 
 /**
  * The result of analyzing of a single file.
  *
  * These results are self-consistent, i.e. [content], [contentHash], the
  * resolved [unit] correspond to each other. All referenced elements, even
  * external ones, are also self-consistent. But none of the results is
  * guaranteed to be consistent with the state of the files.
  *
@@ skipping 270 matching lines @@
   /**
    * Complete the [signal] future if it is not completed yet. It is safe to
    * call this method multiple times, but the [signal] will complete only once.
    */
   void notify() {
     if (!_completer.isCompleted) {
       _completer.complete(null);
     }
   }
 }