[flutter-frontend] Shard flutter-frontend bot

Speed up bot runtime by sharding the test-runs into - for now - a fixed
number of shards (4).

Change-Id: Id3a256db82a8fd7ed7902766d74c9f790adbc1cb
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/201821
Reviewed-by: Johnni Winther <johnniwinther@google.com>
Commit-Queue: Jens Johansen <jensj@google.com>

pkg/frontend_server/test/frontend_server_flutter.dart

@@ -43,17 +43,26 @@ Future<NnbdMode> _getNNBDMode(Uri script, Uri packagesFileUri) async {
 }
 
 Future compileTests(String flutterDir, String flutterPlatformDir, Logger logger,
-    {String filter}) async {
+    {String filter, int shards: 1, int shard: 0}) async {
   if (flutterDir == null || !(new Directory(flutterDir).existsSync())) {
     throw "Didn't get a valid flutter directory to work with.";
   }
-  Directory flutterDirectory = new Directory(flutterDir);
+  if (shards < 1) {
+    throw "Shards must be >= 1";
+  }
+  if (shard < 0) {
+    throw "Shard must be >= 0";
+  }
+  if (shard >= shards) {
+    throw "Shard must be < shards";
+  }
   // Ensure the path ends in a slash.
-  flutterDirectory = new Directory.fromUri(flutterDirectory.uri);
+  final Directory flutterDirectory =
+      new Directory.fromUri(new Directory(flutterDir).uri);
 
   List<FileSystemEntity> allFlutterFiles =
       flutterDirectory.listSync(recursive: true, followLinks: false);
-  Directory flutterPlatformDirectory;
+  Directory flutterPlatformDirectoryTmp;
 
   if (flutterPlatformDir == null) {
     List<File> platformFiles = new List<File>.from(allFlutterFiles.where((f) =>
@@ -63,16 +72,16 @@ Future compileTests(String flutterDir, String flutterPlatformDir, Logger logger,
     if (platformFiles.length < 1) {
       throw "Expected to find a flutter platform file but didn't.";
     }
-    flutterPlatformDirectory = platformFiles.first.parent;
+    flutterPlatformDirectoryTmp = platformFiles.first.parent;
   } else {
-    flutterPlatformDirectory = Directory(flutterPlatformDir);
+    flutterPlatformDirectoryTmp = Directory(flutterPlatformDir);
   }
-  if (!flutterPlatformDirectory.existsSync()) {
-    throw "$flutterPlatformDirectory doesn't exist.";
+  if (!flutterPlatformDirectoryTmp.existsSync()) {
+    throw "$flutterPlatformDirectoryTmp doesn't exist.";
   }
   // Ensure the path ends in a slash.
-  flutterPlatformDirectory =
-      new Directory.fromUri(flutterPlatformDirectory.uri);
+  final Directory flutterPlatformDirectory =
+      new Directory.fromUri(flutterPlatformDirectoryTmp.uri);
 
   if (!new File.fromUri(
           flutterPlatformDirectory.uri.resolve("platform_strong.dill"))
@@ -87,9 +96,11 @@ Future compileTests(String flutterDir, String flutterPlatformDir, Logger logger,
       f.uri.toString().endsWith("/.packages")));
 
   List<String> allCompilationErrors = [];
-  for (File dotPackage in dotPackagesFiles) {
-    Directory systemTempDir = Directory.systemTemp;
-    Directory tempDir;
+  final Directory systemTempDir = Directory.systemTemp;
+  List<_QueueEntry> queue = [];
+  int totalFiles = 0;
+  for (int i = 0; i < dotPackagesFiles.length; i++) {
+    File dotPackage = dotPackagesFiles[i];
     Directory testDir =
         new Directory.fromUri(dotPackage.parent.uri.resolve("test/"));
     if (!testDir.existsSync()) continue;
@@ -98,7 +109,6 @@ Future compileTests(String flutterDir, String flutterPlatformDir, Logger logger,
       // in a setup that can handle that.
       continue;
     }
-    logger.notice("Go for $testDir");
     List<File> testFiles =
         new List<File>.from(testDir.listSync(recursive: true).where((f) {
       if (!f.path.endsWith("_test.dart")) return false;
@@ -126,27 +136,46 @@ Future compileTests(String flutterDir, String flutterPlatformDir, Logger logger,
     }
     for (List<File> files in [weak, strong]) {
       if (files.isEmpty) continue;
-      tempDir = systemTempDir.createTempSync('flutter_frontend_test');
-      try {
-        List<String> compilationErrors = await attemptStuff(
-            files,
-            tempDir,
-            flutterPlatformDirectory,
-            dotPackage,
-            testDir,
-            flutterDirectory,
-            logger,
-            filter);
-        if (compilationErrors.isNotEmpty) {
-          logger.notice("Notice that we had ${compilationErrors.length} "
-              "compilation errors for $testDir");
-          allCompilationErrors.addAll(compilationErrors);
-        }
-      } finally {
-        tempDir.delete(recursive: true);
-      }
+      queue.add(new _QueueEntry(files, dotPackage, testDir));
+      totalFiles += files.length;
     }
   }
+
+  // Process queue, taking shards into account.
+  // This involves ignoring some queue entries and cutting others up to
+  // process exactly the files assigned to this shard.
+  int shardChunkSize = (totalFiles + shards - 1) ~/ shards;
+  int chunkStart = shard * shardChunkSize;
+  int chunkEnd = (shard + 1) * shardChunkSize;
+  int processedFiles = 0;
+
+  for (_QueueEntry queueEntry in queue) {
+    if (processedFiles < chunkEnd &&
+        processedFiles + queueEntry.files.length >= chunkStart) {
+      List<File> chunk = [];
+      for (File file in queueEntry.files) {
+        if (processedFiles >= chunkStart && processedFiles < chunkEnd) {
+          chunk.add(file);
+        }
+        processedFiles++;
+      }
+
+      await _processFiles(
+          systemTempDir,
+          chunk,
+          flutterPlatformDirectory,
+          queueEntry.dotPackage,
+          queueEntry.testDir,
+          flutterDirectory,
+          logger,
+          filter,
+          allCompilationErrors);
+    } else {
+      // None of these files are part of the chunk.
+      processedFiles += queueEntry.files.length;
+    }
+  }
+
   if (allCompilationErrors.isNotEmpty) {
     logger.notice(
         "Had a total of ${allCompilationErrors.length} compilation errors:");
@@ -155,6 +184,45 @@ Future compileTests(String flutterDir, String flutterPlatformDir, Logger logger,
   }
 }
 
+class _QueueEntry {
+  final List<File> files;
+  final File dotPackage;
+  final Directory testDir;
+
+  _QueueEntry(this.files, this.dotPackage, this.testDir);
+}
+
+Future<void> _processFiles(
+    Directory systemTempDir,
+    List<File> files,
+    Directory flutterPlatformDirectory,
+    File dotPackage,
+    Directory testDir,
+    Directory flutterDirectory,
+    Logger logger,
+    String filter,
+    List<String> allCompilationErrors) async {
+  Directory tempDir = systemTempDir.createTempSync('flutter_frontend_test');
+  try {
+    List<String> compilationErrors = await attemptStuff(
+        files,
+        tempDir,
+        flutterPlatformDirectory,
+        dotPackage,
+        testDir,
+        flutterDirectory,
+        logger,
+        filter);
+    if (compilationErrors.isNotEmpty) {
+      logger.notice("Notice that we had ${compilationErrors.length} "
+          "compilation errors for $testDir");
+      allCompilationErrors.addAll(compilationErrors);
+    }
+  } finally {
+    tempDir.delete(recursive: true);
+  }
+}
+
 Future<List<String>> attemptStuff(
     List<File> testFiles,
     Directory tempDir,

pkg/frontend_server/test/frontend_server_flutter_suite.dart

@@ -146,6 +146,9 @@ class SuiteConfiguration {
   final String configurationName;
   final String testFilter;
 
+  final int shard;
+  final int shards;
+
   final String flutterDir;
   final String flutterPlatformDir;
 
@@ -157,7 +160,9 @@ class SuiteConfiguration {
       this.configurationName,
       this.testFilter,
       this.flutterDir,
-      this.flutterPlatformDir);
+      this.flutterPlatformDir,
+      this.shard,
+      this.shards);
 }
 
 void runSuite(SuiteConfiguration configuration) async {
@@ -168,6 +173,8 @@ void runSuite(SuiteConfiguration configuration) async {
       configuration.flutterPlatformDir,
       logger,
       filter: configuration.testFilter,
+      shard: configuration.shard,
+      shards: configuration.shards,
     );
   } catch (e) {
     logger.logUnexpectedResult("startup");
@@ -188,50 +195,58 @@ main([List<String> arguments = const <String>[]]) async {
     ..listen((logEntry) => logs.add(logEntry));
   String filter = options.testFilter;
 
-  // Start the test suite in a new isolate.
-  ReceivePort exitPort = new ReceivePort();
-  ReceivePort errorPort = new ReceivePort();
-  SuiteConfiguration configuration = new SuiteConfiguration(
-    resultsPort.sendPort,
-    logsPort.sendPort,
-    options.verbose,
-    options.printFailureLog,
-    options.configurationName,
-    filter,
-    options.flutterDir,
-    options.flutterPlatformDir,
-  );
-  Future<bool> future = Future<bool>(() async {
-    Stopwatch stopwatch = Stopwatch()..start();
-    print("Running suite");
-    Isolate isolate = await Isolate.spawn<SuiteConfiguration>(
-        runSuite, configuration,
-        onExit: exitPort.sendPort, onError: errorPort.sendPort);
-    bool gotError = false;
-    StreamSubscription errorSubscription = errorPort.listen((message) {
-      print("Got error: $message!");
-      gotError = true;
-      logs.add("$message");
+  const int shards = 4;
+  List<Future<bool>> futures = [];
+  for (int shard = 0; shard < shards; shard++) {
+    // Start the test suite in a new isolate.
+    ReceivePort exitPort = new ReceivePort();
+    ReceivePort errorPort = new ReceivePort();
+    SuiteConfiguration configuration = new SuiteConfiguration(
+      resultsPort.sendPort,
+      logsPort.sendPort,
+      options.verbose,
+      options.printFailureLog,
+      options.configurationName,
+      filter,
+      options.flutterDir,
+      options.flutterPlatformDir,
+      shard,
+      shards,
+    );
+    Future<bool> future = Future<bool>(() async {
+      Stopwatch stopwatch = Stopwatch()..start();
+      print("Running suite shard $shard of $shards");
+      Isolate isolate = await Isolate.spawn<SuiteConfiguration>(
+          runSuite, configuration,
+          onExit: exitPort.sendPort, onError: errorPort.sendPort);
+      bool gotError = false;
+      StreamSubscription errorSubscription = errorPort.listen((message) {
+        print("Got error: $message!");
+        gotError = true;
+        logs.add("$message");
+      });
+      bool timedOut = false;
+      Timer timer = Timer(timeoutDuration, () {
+        timedOut = true;
+        print("Suite timed out after "
+            "${timeoutDuration.inMilliseconds}ms");
+        isolate.kill(priority: Isolate.immediate);
+      });
+      await exitPort.first;
+      errorSubscription.cancel();
+      timer.cancel();
+      if (!timedOut && !gotError) {
+        int seconds = stopwatch.elapsedMilliseconds ~/ 1000;
+        print("Suite finished (shard #$shard) (took ${seconds} seconds)");
+      }
+      return timedOut || gotError;
     });
-    bool timedOut = false;
-    Timer timer = Timer(timeoutDuration, () {
-      timedOut = true;
-      print("Suite timed out after "
-          "${timeoutDuration.inMilliseconds}ms");
-      isolate.kill(priority: Isolate.immediate);
-    });
-    await exitPort.first;
-    errorSubscription.cancel();
-    timer.cancel();
-    if (!timedOut && !gotError) {
-      int seconds = stopwatch.elapsedMilliseconds ~/ 1000;
-      print("Suite finished (took ${seconds} seconds)");
-    }
-    return timedOut || gotError;
-  });
+    futures.add(future);
+  }
 
   // Wait for isolate to terminate and clean up.
-  bool timeoutOrCrash = await future;
+  Iterable<bool> timeoutsOrCrashes = await Future.wait(futures);
+  bool timeoutOrCrash = timeoutsOrCrashes.any((timeout) => timeout);
   resultsPort.close();
   logsPort.close();