[vm/concurrency] Update IsolateSpawnMemory benchmark

Ensure to count heap sizes of all isolate groups, since
AOT works with --enable-isolate-groups, JIT without.

Make the benchmark work in phases to make numbers more stable.

Add a max-process-rss metric to see peak memory during duration
of the entire benchmark.

Change-Id: I481b2cb8b666885b5c2b9c53fff1177accd01830
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/126724
Commit-Queue: Martin Kustermann <[email protected]>
Reviewed-by: Alexander Aprelev <[email protected]>

Author: mkustermann

benchmarks/IsolateSpawnMemory/dart/IsolateSpawnMemory.dart

@@ -6,11 +6,14 @@ import 'dart:async';
 import 'dart:developer';
 import 'dart:io';
 import 'dart:isolate';
+import 'dart:math' as math;
 
 import 'package:compiler/src/dart2js.dart' as dart2js_main;
 import 'package:vm_service/vm_service.dart' as vm_service;
 import 'package:vm_service/vm_service_io.dart' as vm_service_io;
 
+const String compilerIsolateName = 'isolate-compiler';
+
 class Result {
   const Result(
       this.rssOnStart, this.rssOnEnd, this.heapOnStart, this.heapOnEnd);
@@ -22,108 +25,167 @@ class Result {
 }
 
 class StartMessage {
-  const StartMessage(this.wsUri, this.groupRefId, this.sendPort);
+  const StartMessage(this.wsUri, this.sendPort);
 
   final String wsUri;
-  final String groupRefId;
   final SendPort sendPort;
 }
 
 class SpawnMemory {
-  SpawnMemory(this.name, this.wsUri, this.groupRefId);
+  SpawnMemory(this.name, this.wsUri);
 
   Future<void> report() async {
-    const numberOfRuns = 3;
+    int maxProcessRss = 0;
+    final timer = Timer.periodic(const Duration(microseconds: 100), (_) {
+      maxProcessRss = math.max(maxProcessRss, ProcessInfo.currentRss);
+    });
+
+    const numberOfBenchmarks = 3;
+
+    final beforeRss = ProcessInfo.currentRss;
+    final beforeHeap = await currentHeapUsage(wsUri);
 
-    int sumDeltaRssOnStart = 0;
-    int sumDeltaRssOnEnd = 0;
-    int sumDeltaHeapOnStart = 0;
-    int sumDeltaHeapOnEnd = 0;
-    for (int i = 0; i < numberOfRuns; i++) {
+    final iterators = <StreamIterator>[];
+    final continuations = <SendPort>[];
+
+    // Start all isolates & make them wait.
+    for (int i = 0; i < numberOfBenchmarks; i++) {
       final receivePort = ReceivePort();
-      final beforeRss = ProcessInfo.currentRss;
-      final beforeHeap = await currentHeapUsage(wsUri, groupRefId);
-
-      final startMessage =
-          StartMessage(wsUri, groupRefId, receivePort.sendPort);
-      await Isolate.spawn(isolateCompiler, startMessage);
-
-      final Result result = await receivePort.first;
-      sumDeltaRssOnStart += result.rssOnStart - beforeRss;
-      sumDeltaRssOnEnd += result.rssOnEnd - beforeRss;
-      sumDeltaHeapOnStart += result.heapOnStart - beforeHeap;
-      sumDeltaHeapOnEnd += result.heapOnEnd - beforeHeap;
+      final startMessage = StartMessage(wsUri, receivePort.sendPort);
+      await Isolate.spawn(isolateCompiler, startMessage,
+          debugName: compilerIsolateName);
+      final iterator = StreamIterator(receivePort);
+
+      if (!await iterator.moveNext()) throw 'failed';
+      continuations.add(iterator.current as SendPort);
+
+      iterators.add(iterator);
+    }
+
+    final readyRss = ProcessInfo.currentRss;
+    final readyHeap = await currentHeapUsage(wsUri);
+
+    // Let all isolates do the dart2js compilation.
+    for (int i = 0; i < numberOfBenchmarks; i++) {
+      final iterator = iterators[i];
+      final continuation = continuations[i];
+      continuation.send(null);
+      if (!await iterator.moveNext()) throw 'failed';
+      if (iterator.current != 'done') throw 'failed';
+    }
+
+    final doneRss = ProcessInfo.currentRss;
+    final doneHeap = await currentHeapUsage(wsUri);
+
+    // Shut down helper isolates
+    for (int i = 0; i < numberOfBenchmarks; i++) {
+      final iterator = iterators[i];
+      final continuation = continuations[i];
+      continuation.send(null);
+      if (!await iterator.moveNext()) throw 'failed';
+      if (iterator.current != 'shutdown') throw 'failed';
+      await iterator.cancel();
     }
-    print(
-        "${name}RssOnStart(MemoryUse): ${sumDeltaRssOnStart ~/ numberOfRuns}");
-    print("${name}RssOnEnd(MemoryUse): ${sumDeltaRssOnEnd ~/ numberOfRuns}");
-    print(
-        "${name}HeapOnStart(MemoryUse): ${sumDeltaHeapOnStart ~/ numberOfRuns}");
-    print("${name}HeapOnEnd(MemoryUse): ${sumDeltaHeapOnEnd ~/ numberOfRuns}");
+    timer.cancel();
+
+    final readyDiffRss =
+        math.max(0, readyRss - beforeRss) ~/ numberOfBenchmarks;
+    final readyDiffHeap =
+        math.max(0, readyHeap - beforeHeap) ~/ numberOfBenchmarks;
+    final doneDiffRss = math.max(0, doneRss - beforeRss) ~/ numberOfBenchmarks;
+    final doneDiffHeap =
+        math.max(0, doneHeap - beforeHeap) ~/ numberOfBenchmarks;
+
+    print("${name}RssOnStart(MemoryUse): $readyDiffRss");
+    print("${name}RssOnEnd(MemoryUse): $doneDiffRss");
+    print("${name}HeapOnStart(MemoryUse): $readyDiffHeap");
+    print("${name}HeapOnEnd(MemoryUse): $doneDiffHeap");
+    print("${name}PeakProcessRss(MemoryUse): $maxProcessRss");
   }
 
   final String name;
   final String wsUri;
-  final String groupRefId;
-  RawReceivePort receivePort;
 }
 
 Future<void> isolateCompiler(StartMessage startMessage) async {
-  final rssOnStart = ProcessInfo.currentRss;
-  final heapOnStart =
-      await currentHeapUsage(startMessage.wsUri, startMessage.groupRefId);
+  final port = ReceivePort();
+  final iterator = StreamIterator(port);
+
+  // Let main isolate know we're ready.
+  startMessage.sendPort.send(port.sendPort);
+  await iterator.moveNext();
+
   await runZoned(
       () => dart2js_main.internalMain(<String>[
             "benchmarks/IsolateSpawnMemory/dart/helloworld.dart",
             '--libraries-spec=sdk/lib/libraries.json'
           ]),
       zoneSpecification: ZoneSpecification(
           print: (Zone self, ZoneDelegate parent, Zone zone, String line) {}));
-  startMessage.sendPort.send(Result(
-      rssOnStart,
-      ProcessInfo.currentRss,
-      heapOnStart,
-      await currentHeapUsage(startMessage.wsUri, startMessage.groupRefId)));
-  ReceivePort(); // prevent isolate from exiting to ensure Rss monotonically grows
+
+  // Let main isolate know we're done.
+  startMessage.sendPort.send('done');
+  await iterator.moveNext();
+
+  // Closes the port.
+  startMessage.sendPort.send('shutdown');
+  await iterator.cancel();
 }
 
-Future<int> currentHeapUsage(String wsUri, String groupRefId) async {
+Future<int> currentHeapUsage(String wsUri) async {
   final vm_service.VmService vmService =
       await vm_service_io.vmServiceConnectUri(wsUri);
-  final vm_service.MemoryUsage usage =
-      await vmService.getIsolateGroupMemoryUsage(groupRefId);
+  final groupIds = await getGroupIds(vmService);
+  int sum = 0;
+  for (final groupId in groupIds) {
+    final vm_service.MemoryUsage usage =
+        await vmService.getIsolateGroupMemoryUsage(groupId);
+    sum += usage.heapUsage + usage.externalUsage;
+  }
   vmService.dispose();
-  return usage.heapUsage + usage.externalUsage;
+  return sum;
 }
 
 Future<void> main() async {
+  // Only if we successfully reach the end will we set 0 exit code.
+  exitCode = 255;
+
   final ServiceProtocolInfo info = await Service.controlWebServer(enable: true);
   final Uri observatoryUri = info.serverUri;
   final String wsUri =
       'ws://${observatoryUri.authority}${observatoryUri.path}ws';
-  final vm_service.VmService vmService =
-      await vm_service_io.vmServiceConnectUri(wsUri);
-  final String mainGroupRefId = await getMainGroupRefId(vmService);
-  vmService.dispose();
+  await SpawnMemory("IsolateSpawnMemory.Dart2JSDelta", wsUri).report();
 
-  await SpawnMemory("IsolateSpawnMemory.Dart2JSDelta", wsUri, mainGroupRefId)
-      .report();
+  // Only if we successfully reach the end will we set 0 exit code.
+  exitCode = 0;
 }
 
-Future<String> getMainGroupRefId(vm_service.VmService vmService) async {
+// Returns the set of isolate groups for which we should count the heap usage.
+//
+// We have two cases
+//
+//   a) --enable-isolate-groups: All isolates will be within the same isolate
+//   group.
+//
+//   b) --no-enable-isolate-groups: All isolates will be within their own,
+//   separate isolate group.
+//
+// In both cases we want to sum up the heap sizes of all isolate groups.
+Future<List<String>> getGroupIds(vm_service.VmService vmService) async {
+  final groupIds = <String>{};
   final vm = await vmService.getVM();
   for (vm_service.IsolateGroupRef groupRef in vm.isolateGroups) {
     final vm_service.IsolateGroup group =
         await vmService.getIsolateGroup(groupRef.id);
     for (vm_service.IsolateRef isolateRef in group.isolates) {
       final isolateOrSentinel = await vmService.getIsolate(isolateRef.id);
       if (isolateOrSentinel is vm_service.Isolate) {
-        final vm_service.Isolate isolate = isolateOrSentinel;
-        if (isolate.name == 'main') {
-          return groupRef.id;
-        }
+        groupIds.add(groupRef.id);
       }
     }
   }
-  throw "Could not find main isolate";
+  if (groupIds.isEmpty) {
+    throw "Could not find main isolate";
+  }
+  return groupIds.toList();
 }