VM: [Kernel] Switch to DFE for testing -c dartk configuration.

tools/testing/dart/test_runner.dart

 // Copyright (c) 2012, 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.
 
 /**
  * Classes and methods for executing tests.
  *
  * This module includes:
  * - Managing parallel execution of tests, including timeout checks.
  * - Evaluating the output of each test as pass/fail/crash/timeout.
@@ skipping 347 matching lines @@
   void _buildHashCode(HashCodeBuilder builder) {
     super._buildHashCode(builder);
     builder.addJson(flavor);
   }
 
   bool _equal(AnalysisCommand other) =>
       super._equal(other) && flavor == other.flavor;
 }
 
 class VmCommand extends ProcessCommand {
+  final bool needsDFERunner;
   VmCommand._(String executable, List<String> arguments,
-      Map<String, String> environmentOverrides)
+      Map<String, String> environmentOverrides,
+      bool this.needsDFERunner)
       : super._("vm", executable, arguments, environmentOverrides);
+
+  bool _equal(VmCommand other) =>
+      super._equal(other) && needsDFERunner == other.needsDFERunner;

[kustermann, 2017/01/31 15:39:00]

whenever you overwrite _equal, also overwrite _buildHashCode

[Vyacheslav Egorov (Google), 2017/01/31 17:06:56]

Done.

 }
 
 class VmBatchCommand extends ProcessCommand implements VmCommand {
   final String dartFile;
   final bool checked;
+  final needsDFERunner = false;
 
   VmBatchCommand._(String executable, String dartFile, List<String> arguments,
       Map<String, String> environmentOverrides, {this.checked: true})
       : this.dartFile = dartFile,
         super._('vm-batch', executable, arguments, environmentOverrides);
 
   @override
   List<String> get batchArguments => checked
       ? ['--checked', dartFile]
       : [dartFile];
@@ skipping 340 matching lines @@
 
   AnalysisCommand getAnalysisCommand(
       String displayName, executable, arguments, environmentOverrides,
       {String flavor: 'dart2analyzer'}) {
     var command = new AnalysisCommand._(
         flavor, displayName, executable, arguments, environmentOverrides);
     return _getUniqueCommand(command);
   }
 
   VmCommand getVmCommand(String executable, List<String> arguments,
-      Map<String, String> environmentOverrides) {
-    var command = new VmCommand._(executable, arguments, environmentOverrides);
+      Map<String, String> environmentOverrides, {bool needsDFERunner: false}) {
+    var command = new VmCommand._(executable, arguments, environmentOverrides, needsDFERunner);
     return _getUniqueCommand(command);
   }
 
   VmBatchCommand getVmBatchCommand(String executable, String tester,
       List<String> arguments, Map<String, String> environmentOverrides,
       {bool checked: true}) {
     var command =
         new VmBatchCommand._(executable, tester, arguments, environmentOverrides,
             checked: checked);
     return _getUniqueCommand(command);
@@ skipping 837 matching lines @@
           outWarnings.add(fields[FORMATTED_ERROR]);
         }
         // OK to Skip error output that doesn't match the machine format
       }
     }
   }
 }
 
 class VmCommandOutputImpl extends CommandOutputImpl
     with UnittestSuiteMessagesMixin {
+  static const DART_VM_EXITCODE_DFE_ERROR = 252;
   static const DART_VM_EXITCODE_COMPILE_TIME_ERROR = 254;
   static const DART_VM_EXITCODE_UNCAUGHT_EXCEPTION = 255;
 
   VmCommandOutputImpl(Command command, int exitCode, bool timedOut,
       List<int> stdout, List<int> stderr, Duration time, int pid)
       : super(command, exitCode, timedOut, stdout, stderr, time, false, pid);
 
   Expectation result(TestCase testCase) {
     // Handle crashes and timeouts first
+    if (exitCode == DART_VM_EXITCODE_DFE_ERROR) return Expectation.DARTK_CRASH;
     if (hasCrashed) return Expectation.CRASH;
     if (hasTimedOut) return Expectation.TIMEOUT;
 
     // Multitests are handled specially
     if (testCase.expectCompileError) {
       if (exitCode == DART_VM_EXITCODE_COMPILE_TIME_ERROR) {
         return Expectation.PASS;
       }
       return Expectation.MISSING_COMPILETIME_ERROR;
     }
@@ skipping 294 matching lines @@
   int timeout;
   bool timedOut = false;
   DateTime startTime;
   Timer timeoutTimer;
   int pid;
   OutputLog stdout = new OutputLog();
   OutputLog stderr = new OutputLog();
   List<String> diagnostics = <String>[];
   bool compilationSkipped = false;
   Completer<CommandOutput> completer;
+  List<String> preArguments;
 
-  RunningProcess(this.command, this.timeout);
+  RunningProcess(this.command, this.timeout, {this.preArguments});
 
   Future<CommandOutput> run() {
     completer = new Completer<CommandOutput>();
     startTime = new DateTime.now();
     _runCommand();
     return completer.future;
   }
 
   void _runCommand() {
     command.outputIsUpToDate.then((bool isUpToDate) {
       if (isUpToDate) {
         compilationSkipped = true;
         _commandComplete(0);
       } else {
         var processEnvironment = _createProcessEnvironment();
+        var args = command.arguments;
+        if (preArguments != null) {
+          args = []..addAll(preArguments)..addAll(args);
+        }
         Future processFuture = io.Process.start(
-            command.executable, command.arguments,
+            command.executable, args,
             environment: processEnvironment,
             workingDirectory: command.workingDirectory);
         processFuture.then((io.Process process) {
           StreamSubscription stdoutSubscription =
               _drainStream(process.stdout, stdout);
           StreamSubscription stderrSubscription =
               _drainStream(process.stderr, stderr);
 
           var stdoutCompleter = new Completer();
           var stderrCompleter = new Completer();
@@ skipping 133 matching lines @@
       }
     }
     for (var excludedEnvironmentVariable in EXCLUDED_ENVIRONMENT_VARIABLES) {
       environment.remove(excludedEnvironmentVariable);
     }
 
     return environment;
   }
 }
 
-class BatchRunnerProcess {
+

[kustermann, 2017/01/31 15:39:00]

one new line is enough

[Vyacheslav Egorov (Google), 2017/01/31 17:06:56]

Done.

+class BatchRunnerProcess  {
   Completer<CommandOutput> _completer;
   ProcessCommand _command;
   List<String> _arguments;
   String _runnerType;
 
   io.Process _process;
   Map _processEnvironmentOverrides;
   Completer _stdoutCompleter;
   Completer _stderrCompleter;
   StreamSubscription<String> _stdoutSubscription;
   StreamSubscription<String> _stderrSubscription;
   Function _processExitHandler;
 
   bool _currentlyRunning = false;
   OutputLog _testStdout;
   OutputLog _testStderr;
   String _status;
   DateTime _startTime;
   Timer _timer;
 
-  BatchRunnerProcess();
-
   Future<CommandOutput> runCommand(String runnerType, ProcessCommand command,
       int timeout, List<String> arguments) {
     assert(_completer == null);
     assert(!_currentlyRunning);
 
     _completer = new Completer<CommandOutput>();
     bool sameRunnerType = _runnerType == runnerType &&
         _dictEquals(_processEnvironmentOverrides, command.environmentOverrides);
     _runnerType = runnerType;
     _currentlyRunning = true;
@@ skipping 176 matching lines @@
   bool _dictEquals(Map a, Map b) {
     if (a == null) return b == null;
     if (b == null) return false;
     for (var key in a.keys) {
       if (a[key] != b[key]) return false;
     }
     return true;
   }
 }
 
+class BatchDFEProcess  {
+  io.Process _process;
+  int _port = -1;
+  Function _processExitHandler;
+
+  Completer terminating = null;
+
+  bool locked = false;
+
+  Future<int> acquire() async {
+    try {
+      assert(!locked);
+      locked = true;
+      if (_process == null) {
+        await _startProcess();
+      }
+      return _port;
+    } catch(e) {
+      locked = false;
+      rethrow;
+    }
+  }
+
+  void release() {
+    locked = false;
+  }
+
+  Future terminate() {
+    locked = true;
+    if (_process == null) {
+      return new Future.value(true);
+    }
+    if (terminating == null) {
+      terminating = new Completer();
+      _process.kill();
+    }
+    return terminating.future;
+  }
+
+  _onExit(exitCode) {
+    if (terminating != null) {
+      terminating.complete();
+      return;
+    }
+
+    _process = null;
+    locked = false;
+    _port = -1;
+  }
+
+  _startProcess() async {

[kustermann, 2017/01/31 15:39:00]

Future 

[Vyacheslav Egorov (Google), 2017/01/31 17:06:56]

Done.

+    final executable = 'tools/dartk_wrappers/dfe_worker';
+    final arguments = [];
+
+    try {
+      _port = -1;
+      _process = await io.Process.start(executable, arguments);

[kustermann, 2017/01/31 15:39:00]

You can just use Platform.executable and 'tools/dartk_wrappers/dfe_worker' here.
The reason we have a wrapper script for dartk is different and not necessary
here.

[Vyacheslav Egorov (Google), 2017/01/31 17:06:56]

Done.

+      _process.exitCode.then(_onExit);
+
+      final readyMsg = await _process.stdout.transform(UTF8.decoder).transform(new LineSplitter()).first;

[kustermann, 2017/01/31 15:39:00]

long line

this will potentially crash the subprocess if it prints more than one line
(because the write causes SIGPIPE / broken pipe).

The .first will cancel the subscription after the first line.

[Vyacheslav Egorov (Google), 2017/01/31 17:06:56]

Done.

[Vyacheslav Egorov (Google), 2017/01/31 17:06:56]

Done.

+      final data = readyMsg.split(' ');
+      assert(data[0] == 'READY');

[kustermann, 2017/01/31 15:39:00]

Consider draining _process.stderr as well (maybe just using DebugLogger or
io.stderr to report it)

[Vyacheslav Egorov (Google), 2017/01/31 17:06:56]

Done.

+
+      _port = int.parse(data[1]);
+    } catch (e) {
+      // TODO(floitsch): should we try to report the stacktrace?

[kustermann, 2017/01/31 15:39:00]

floitsch :)

[Vyacheslav Egorov (Google), 2017/01/31 17:06:56]

Done.

+      print("Process error:");
+      print("  Command: $executable ${arguments.join(' ')}");
+      print("  Error: $e");
+      // If there is an error starting a batch process, chances are that
+      // it will always fail. So rather than re-trying a 1000+ times, we
+      // exit.
+      io.exit(1);
+      return true;
+    }
+  }
+}
+
 /**
  * [TestCaseEnqueuer] takes a list of TestSuites, generates TestCases and
  * builds a dependency graph of all commands in every TestSuite.
  *
  * It will maintain three helper data structures
  *  - command2node: A mapping from a [Command] to a node in the dependency graph
  *  - command2testCases: A mapping from [Command] to all TestCases that it is
  *    part of.
  *  - remainingTestCases: A set of TestCases that were enqueued but are not
  *    finished
@@ skipping 288 matching lines @@
   final int maxProcesses;
   final int maxBrowserProcesses;
   AdbDevicePool adbDevicePool;
 
   // For dart2js and analyzer batch processing,
   // we keep a list of batch processes.
   final _batchProcesses = new Map<String, List<BatchRunnerProcess>>();
   // We keep a BrowserTestRunner for every configuration.
   final _browserTestRunners = new Map<Map, BrowserTestRunner>();
 
+  List<BatchDFEProcess> _dfeProcesses = null;
+
   bool _finishing = false;
 
   CommandExecutorImpl(
       this.globalConfiguration, this.maxProcesses, this.maxBrowserProcesses,
       {this.adbDevicePool});
 
   Future cleanup() {
     assert(!_finishing);
     _finishing = true;
 
     Future _terminateBatchRunners() {
       var futures = [];
       for (var runners in _batchProcesses.values) {
         futures.addAll(runners.map((runner) => runner.terminate()));
       }
       return Future.wait(futures);
     }
 
     Future _terminateBrowserRunners() {
       var futures =
           _browserTestRunners.values.map((runner) => runner.terminate());
       return Future.wait(futures);
     }
 
-    return Future.wait([_terminateBatchRunners(), _terminateBrowserRunners()]);
+    Future _terminateDFEWorkers() =>
+      Future.wait((_dfeProcesses ?? []).map((p) => p.terminate()));
+
+    return Future.wait([_terminateBatchRunners(),
+    _terminateBrowserRunners(),
+    _terminateDFEWorkers()]);

[kustermann, 2017/01/31 15:39:00]

indentation

[Vyacheslav Egorov (Google), 2017/01/31 17:06:56]

Done.

   }
 
   Future<CommandOutput> runCommand(node, Command command, int timeout) {
     assert(!_finishing);
 
     Future<CommandOutput> runCommand(int retriesLeft) {
       return _runCommand(command, timeout).then((CommandOutput output) {
         if (retriesLeft > 0 && shouldRetryCommand(output)) {
           DebugLogger.warning("Rerunning Command: ($retriesLeft "
               "attempt(s) remains) [cmd: $command]");
@@ skipping 27 matching lines @@
     } else if (command is ScriptCommand) {
       return command.run();
     } else if (command is AdbPrecompilationCommand) {
       assert(adbDevicePool != null);
       return adbDevicePool.acquireDevice().then((AdbDevice device) {
         return _runAdbPrecompilationCommand(device, command, timeout)
             .whenComplete(() {
           adbDevicePool.releaseDevice(device);
         });
       });
+    } else if (command is VmCommand && command.needsDFERunner) {
+      final runner = _getDFEProcess();
+      return runner.acquire().then((port) {
+        return new RunningProcess(command, timeout, preArguments: ['-DDFE_WORKER_PORT=${port}']).run();
+      }).whenComplete(() => runner.release());
     } else if (command is VmBatchCommand) {
       var name = command.displayName;
       return _getBatchRunner(command.displayName + command.dartFile)
           .runCommand(name, command, timeout, command.arguments);
     } else {
       return new RunningProcess(command, timeout).run();
     }
   }
 
   Future<CommandOutput> _runAdbPrecompilationCommand(
@@ skipping 82 matching lines @@
       }
       _batchProcesses[identifier] = runners;
     }
 
     for (var runner in runners) {
       if (!runner._currentlyRunning) return runner;
     }
     throw new Exception('Unable to find inactive batch runner.');
   }
 
+  BatchDFEProcess _getDFEProcess() {
+    _dfeProcesses ??= new List<BatchDFEProcess>.generate(maxProcesses,
+        (_) => new BatchDFEProcess());
+    return _dfeProcesses.firstWhere((runner) => !runner.locked);
+  }
+
   Future<CommandOutput> _startBrowserControllerTest(
       BrowserTestCommand browserCommand, int timeout) {
     var completer = new Completer<CommandOutput>();
 
     var callback = (BrowserTestOutput output) {
       completer
           .complete(new BrowserControllerTestOutcome(browserCommand, output));
     };
 
     BrowserTest browserTest;
@@ skipping 411 matching lines @@
     }
   }
 
   void eventAllTestsDone() {
     for (var listener in _eventListener) {
       listener.allDone();
     }
     _allDone();
   }
 }