Refactor devicelab bin/run.dart and other cleanup (#96320)

Author: Hixie

dev/devicelab/bin/run.dart

@@ -10,6 +10,7 @@ import 'dart:isolate';
 
 import 'package:logging/logging.dart';
 import 'package:path/path.dart' as path;
+import 'package:process/process.dart';
 import 'package:stack_trace/stack_trace.dart';
 
 import 'devices.dart';
@@ -41,35 +42,43 @@ bool _isTaskRegistered = false;
 ///
 /// It is OK for a [task] to perform many things. However, only one task can be
 /// registered per Dart VM.
-Future<TaskResult> task(TaskFunction task) async {
+///
+/// If no `processManager` is provided, a default [LocalProcessManager] is created
+/// for the task.
+Future<TaskResult> task(TaskFunction task, { ProcessManager? processManager }) async {
   if (_isTaskRegistered)
     throw StateError('A task is already registered');
-
   _isTaskRegistered = true;
 
+  processManager ??= const LocalProcessManager();
+
   // TODO(ianh): allow overriding logging.
   Logger.root.level = Level.ALL;
   Logger.root.onRecord.listen((LogRecord rec) {
     print('${rec.level.name}: ${rec.time}: ${rec.message}');
   });
 
-  final _TaskRunner runner = _TaskRunner(task);
+  final _TaskRunner runner = _TaskRunner(task, processManager);
   runner.keepVmAliveUntilTaskRunRequested();
   return runner.whenDone;
 }
 
 class _TaskRunner {
-  _TaskRunner(this.task) {
+  _TaskRunner(this.task, this.processManager) {
     registerExtension('ext.cocoonRunTask',
         (String method, Map<String, String> parameters) async {
       final Duration? taskTimeout = parameters.containsKey('timeoutInMinutes')
         ? Duration(minutes: int.parse(parameters['timeoutInMinutes']!))
         : null;
-      // This is only expected to be passed in unit test runs so they do not
-      // kill the Dart process that is running them and waste time running config.
-      final bool runFlutterConfig = parameters['runFlutterConfig'] != 'false';
+      final bool runFlutterConfig = parameters['runFlutterConfig'] != 'false'; // used by tests to avoid changing the configuration
       final bool runProcessCleanup = parameters['runProcessCleanup'] != 'false';
-      final TaskResult result = await run(taskTimeout, runProcessCleanup: runProcessCleanup, runFlutterConfig: runFlutterConfig);
+      final String? localEngine = parameters['localEngine'];
+      final TaskResult result = await run(
+        taskTimeout,
+        runProcessCleanup: runProcessCleanup,
+        runFlutterConfig: runFlutterConfig,
+        localEngine: localEngine,
+      );
       return ServiceExtensionResponse.result(json.encode(result.toJson()));
     });
     registerExtension('ext.cocoonRunnerReady',
@@ -79,6 +88,7 @@ class _TaskRunner {
   }
 
   final TaskFunction task;
+  final ProcessManager processManager;
 
   Future<Device?> _getWorkingDeviceIfAvailable() async {
     try {
@@ -103,6 +113,7 @@ class _TaskRunner {
   Future<TaskResult> run(Duration? taskTimeout, {
     bool runFlutterConfig = true,
     bool runProcessCleanup = true,
+    required String? localEngine,
   }) async {
     try {
       _taskStarted = true;
@@ -113,34 +124,31 @@ class _TaskRunner {
         section('Checking running Dart$exe processes');
         beforeRunningDartInstances = await getRunningProcesses(
           processName: 'dart$exe',
-        ).toSet();
-        final Set<RunningProcessInfo> allProcesses = await getRunningProcesses().toSet();
+          processManager: processManager,
+        );
+        final Set<RunningProcessInfo> allProcesses = await getRunningProcesses(processManager: processManager);
         beforeRunningDartInstances.forEach(print);
         for (final RunningProcessInfo info in allProcesses) {
           if (info.commandLine.contains('iproxy')) {
             print('[LEAK]: ${info.commandLine} ${info.creationDate} ${info.pid} ');
           }
         }
-      } else {
-        section('Skipping check running Dart$exe processes');
       }
 
       if (runFlutterConfig) {
-        print('enabling configs for macOS, Linux, Windows, and Web...');
+        print('Enabling configs for macOS, Linux, Windows, and Web...');
         final int configResult = await exec(path.join(flutterDirectory.path, 'bin', 'flutter'), <String>[
           'config',
           '-v',
           '--enable-macos-desktop',
           '--enable-windows-desktop',
           '--enable-linux-desktop',
           '--enable-web',
-          if (localEngine != null) ...<String>['--local-engine', localEngine!],
+          if (localEngine != null) ...<String>['--local-engine', localEngine],
         ], canFail: true);
         if (configResult != 0) {
           print('Failed to enable configuration, tasks may not run.');
         }
-      } else {
-        print('Skipping enabling configs for macOS, Linux, Windows, and Web');
       }
 
       final Device? device = await _getWorkingDeviceIfAvailable();
@@ -169,26 +177,24 @@ class _TaskRunner {
       }
 
       if (runProcessCleanup) {
-        section('Checking running Dart$exe processes after task...');
-        final List<RunningProcessInfo> afterRunningDartInstances = await getRunningProcesses(
+        section('Terminating lingering Dart$exe processes after task...');
+        final Set<RunningProcessInfo> afterRunningDartInstances = await getRunningProcesses(
           processName: 'dart$exe',
-        ).toList();
+          processManager: processManager,
+        );
         for (final RunningProcessInfo info in afterRunningDartInstances) {
           if (!beforeRunningDartInstances.contains(info)) {
             print('$info was leaked by this test.');
             if (result is TaskResultCheckProcesses) {
               result = TaskResult.failure('This test leaked dart processes');
             }
-            final bool killed = await killProcess(info.pid);
-            if (!killed) {
-              print('Failed to kill process ${info.pid}.');
-            } else {
+            if (await info.terminate(processManager: processManager)) {
               print('Killed process id ${info.pid}.');
+            } else {
+              print('Failed to kill process ${info.pid}.');
             }
           }
         }
-      } else {
-        print('Skipping check running Dart$exe processes after task');
       }
       _completer.complete(result);
       return result;

dev/devicelab/lib/framework/framework.dart

@@ -4,7 +4,6 @@
 
 import 'dart:async';
 import 'dart:convert';
-// import 'dart:core' as core;
 import 'dart:io';
 
 import 'package:meta/meta.dart';
@@ -29,6 +28,10 @@ import 'utils.dart';
 Future<void> runTasks(
   List<String> taskNames, {
   bool exitOnFirstTestFailure = false,
+  // terminateStrayDartProcesses defaults to false so that tests don't have to specify it.
+  // It is set based on the --terminate-stray-dart-processes command line argument in
+  // normal execution, and that flag defaults to true.
+  bool terminateStrayDartProcesses = false,
   bool silent = false,
   String? deviceId,
   String? gitBranch,
@@ -50,6 +53,7 @@ Future<void> runTasks(
         deviceId: deviceId,
         localEngine: localEngine,
         localEngineSrcPath: localEngineSrcPath,
+        terminateStrayDartProcesses: terminateStrayDartProcesses,
         silent: silent,
         taskArgs: taskArgs,
         resultsPath: resultsPath,
@@ -58,23 +62,26 @@ Future<void> runTasks(
         isolateParams: isolateParams,
       );
 
-      section('Flaky status for "$taskName"');
       if (!result.succeeded) {
-        retry++;
+        retry += 1;
       } else {
+        section('Flaky status for "$taskName"');
         if (retry > 0) {
-          print('Total ${retry+1} executions: $retry failures and 1 success');
+          print('Total ${retry+1} executions: $retry failures and 1 false positive.');
           print('flaky: true');
+          // TODO(ianh): stop ignoring this failure. We should set exitCode=1, and quit
+          // if exitOnFirstTestFailure is true.
         } else {
-          print('Total ${retry+1} executions: 1 success');
+          print('Test passed on first attempt.');
           print('flaky: false');
         }
         break;
       }
     }
 
     if (!result.succeeded) {
-      print('Total $retry executions: 0 success');
+      section('Flaky status for "$taskName"');
+      print('Consistently failed across all $retry executions.');
       print('flaky: false');
       exitCode = 1;
       if (exitOnFirstTestFailure) {
@@ -92,6 +99,7 @@ Future<TaskResult> rerunTask(
   String? deviceId,
   String? localEngine,
   String? localEngineSrcPath,
+  bool terminateStrayDartProcesses = false,
   bool silent = false,
   List<String>? taskArgs,
   String? resultsPath,
@@ -105,6 +113,7 @@ Future<TaskResult> rerunTask(
     deviceId: deviceId,
     localEngine: localEngine,
     localEngineSrcPath: localEngineSrcPath,
+    terminateStrayDartProcesses: terminateStrayDartProcesses,
     silent: silent,
     taskArgs: taskArgs,
     isolateParams: isolateParams,
@@ -138,11 +147,12 @@ Future<TaskResult> rerunTask(
 /// [taskArgs] are passed to the task executable for additional configuration.
 Future<TaskResult> runTask(
   String taskName, {
+  bool terminateStrayDartProcesses = false,
   bool silent = false,
   String? localEngine,
   String? localEngineSrcPath,
   String? deviceId,
-  List<String> ?taskArgs,
+  List<String>? taskArgs,
   @visibleForTesting Map<String, String>? isolateParams,
 }) async {
   final String taskExecutable = 'bin/tasks/$taskName.dart';
@@ -198,17 +208,26 @@ Future<TaskResult> runTask(
 
   try {
     final ConnectionResult result = await _connectToRunnerIsolate(await uri.future);
+    print('[$taskName] Connected to VM server.');
+    isolateParams = isolateParams == null ? <String, String>{} : Map<String, String>.of(isolateParams);
+    isolateParams['runProcessCleanup'] = terminateStrayDartProcesses.toString();
     final Map<String, dynamic> taskResultJson = (await result.vmService.callServiceExtension(
       'ext.cocoonRunTask',
       args: isolateParams,
       isolateId: result.isolate.id,
     )).json!;
     final TaskResult taskResult = TaskResult.fromJson(taskResultJson);
-    await runner.exitCode;
+    final int exitCode = await runner.exitCode;
+    print('[$taskName] Process terminated with exit code $exitCode.');
     return taskResult;
+  } catch (error, stack) {
+    print('[$taskName] Task runner system failed with exception!\n$error\n$stack');
+    rethrow;
   } finally {
-    if (!runnerFinished)
+    if (!runnerFinished) {
+      print('[$taskName] Terminating process...');
       runner.kill(ProcessSignal.sigkill);
+    }
     await stdoutSub.cancel();
     await stderrSub.cancel();
   }