// Copyright (c) 2022, 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:convert'; import 'dart:io'; import 'dart:isolate'; import 'dart:typed_data'; import 'package:_macros/src/executor/message_grouper.dart'; import 'package:_macros/src/executor/serialization.dart'; void main() async { for (var serializationMode in [ SerializationMode.json, SerializationMode.byteData ]) { await withSerializationMode(serializationMode, () async { await _isolateSpawnBenchmarks(); await _isolateSpawnUriBenchmarks(); await _separateProcessStdioBenchmarks(); await _separateProcessSocketBenchmarks(); }); } } Future _isolateSpawnBenchmarks() async { void Function(SendPort) childIsolateFn(SerializationMode mode) => (SendPort sendPort) => withSerializationMode(mode, () { var isolateReceivePort = ReceivePort(); isolateReceivePort.listen((data) { deserialize(data); var result = serialize(); result = result is Uint8List ? TransferableTypedData.fromList([result]) : result; sendPort.send(result); }); sendPort.send(isolateReceivePort.sendPort); }); Completer? responseCompleter; late SendPort sendPort; var receivePort = ReceivePort(); var isolate = await Isolate.spawn( childIsolateFn(serializationMode), receivePort.sendPort); final sendPortCompleter = Completer(); receivePort.listen((data) { if (!sendPortCompleter.isCompleted) { sendPortCompleter.complete(data as SendPort); } else { responseCompleter!.complete(data); } }); sendPort = await sendPortCompleter.future; // warmup for (var i = 0; i < 100; i++) { responseCompleter = Completer(); var result = serialize(); result = result is Uint8List ? TransferableTypedData.fromList([result]) : result; sendPort.send(result); deserialize(await responseCompleter.future); } // measure var watch = Stopwatch()..start(); for (var i = 0; i < 100; i++) { responseCompleter = Completer(); var result = serialize(); result = result is Uint8List ? TransferableTypedData.fromList([result]) : result; sendPort.send(result); deserialize(await responseCompleter.future); } print('Isolate.spawn + $serializationMode: ${watch.elapsed}'); receivePort.close(); isolate.kill(); } Future _isolateSpawnUriBenchmarks() async { Completer? responseCompleter; late SendPort sendPort; var receivePort = ReceivePort(); var isolate = await Isolate.spawnUri( Uri.dataFromString(childProgram(serializationMode)), [], receivePort.sendPort); final sendPortCompleter = Completer(); receivePort.listen((data) { if (!sendPortCompleter.isCompleted) { sendPortCompleter.complete(data as SendPort); } else { responseCompleter!.complete(data); } }); sendPort = await sendPortCompleter.future; // warmup for (var i = 0; i < 100; i++) { responseCompleter = Completer(); var result = serialize(); result = result is Uint8List ? TransferableTypedData.fromList([result]) : result; sendPort.send(result); deserialize(await responseCompleter.future); } // measure var watch = Stopwatch()..start(); for (var i = 0; i < 100; i++) { responseCompleter = Completer(); var result = serialize(); result = result is Uint8List ? TransferableTypedData.fromList([result]) : result; sendPort.send(result); deserialize(await responseCompleter.future); } print('Isolate.spawnUri + $serializationMode: ${watch.elapsed}'); receivePort.close(); isolate.kill(); } Future _separateProcessStdioBenchmarks() async { Completer? responseCompleter; var tmpDir = Directory.systemTemp.createTempSync('serialize_bench'); try { var file = File(tmpDir.uri.resolve('main.dart').toFilePath()); file.writeAsStringSync(childProgram(serializationMode)); var process = await Process.start(Platform.resolvedExecutable, [ '--packages=${(await Isolate.packageConfig)!.toFilePath()}', file.path, ]); var listeners = [ process.stderr.listen((event) { print('stderr: ${utf8.decode(event)}'); }), (serializationMode == SerializationMode.json ? process.stdout : MessageGrouper(process.stdout).messageStream) .listen((data) { responseCompleter!.complete(data); }), ]; // warmup for (var i = 0; i < 100; i++) { responseCompleter = Completer(); var result = serialize(); if (result is List) { final bytesBuilder = BytesBuilder(copy: false); _writeLength(result, bytesBuilder); bytesBuilder.add(result); process.stdin.add(bytesBuilder.takeBytes()); } else { process.stdin.writeln(jsonEncode(result)); } deserialize(await responseCompleter.future); } // measure var watch = Stopwatch()..start(); for (var i = 0; i < 100; i++) { responseCompleter = Completer(); var result = serialize(); if (result is List) { final bytesBuilder = BytesBuilder(copy: false); _writeLength(result, bytesBuilder); bytesBuilder.add(result); process.stdin.add(bytesBuilder.takeBytes()); } else { process.stdin.writeln(jsonEncode(result)); } deserialize(await responseCompleter.future); } print('Separate process + Stdio + $serializationMode: ${watch.elapsed}'); for (var listener in listeners) { listener.cancel(); } process.kill(); } catch (e, s) { print('Error running benchmark \n$e\n\n$s'); } finally { tmpDir.deleteSync(recursive: true); } } Future _separateProcessSocketBenchmarks() async { Completer? responseCompleter; var tmpDir = Directory.systemTemp.createTempSync('serialize_bench'); try { var file = File(tmpDir.uri.resolve('main.dart').toFilePath()); file.writeAsStringSync(childProgram(serializationMode)); ServerSocket serverSocket; // Try an ipv6 address loopback first, and fall back on ipv4. try { serverSocket = await ServerSocket.bind(InternetAddress.loopbackIPv6, 0); } on SocketException catch (_) { serverSocket = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); } Completer clientCompleter = Completer(); serverSocket.listen((client) { clientCompleter.complete(client); }); var process = await Process.start(Platform.resolvedExecutable, [ '--packages=${(await Isolate.packageConfig)!.toFilePath()}', file.path, serverSocket.address.address, serverSocket.port.toString(), ]); var client = await clientCompleter.future; // Nagle's algorithm slows us down >100x, disable it. client.setOption(SocketOption.tcpNoDelay, true); var listeners = [ (serializationMode == SerializationMode.json ? client : MessageGrouper(client).messageStream) .listen((event) { responseCompleter!.complete(event); }), process.stderr.listen((event) { print('stderr: ${utf8.decode(event)}'); }), process.stdout.listen((event) { print('stdout: ${utf8.decode(event)}'); }), ]; // warmup for (var i = 0; i < 100; i++) { responseCompleter = Completer(); var result = serialize(); if (result is List) { final bytesBuilder = BytesBuilder(copy: false); _writeLength(result, bytesBuilder); bytesBuilder.add(result); client.add(bytesBuilder.takeBytes()); } else { client.write(jsonEncode(result)); } deserialize(await responseCompleter.future); } // measure var watch = Stopwatch()..start(); for (var i = 0; i < 100; i++) { responseCompleter = Completer(); var result = serialize(); if (result is List) { final bytesBuilder = BytesBuilder(copy: false); _writeLength(result, bytesBuilder); bytesBuilder.add(result); client.add(bytesBuilder.takeBytes()); } else { client.write(jsonEncode(result)); } deserialize(await responseCompleter.future); } print('Separate process + Socket + $serializationMode: ${watch.elapsed}'); for (var listener in listeners) { listener.cancel(); } process.kill(); await serverSocket.close(); client.destroy(); } catch (e, s) { print('Error running benchmark \n$e\n\n$s'); } finally { tmpDir.deleteSync(recursive: true); } } void _writeLength(List result, BytesBuilder bytesBuilder) { int length = (result as Uint8List).lengthInBytes; if (length > 0xffffffff) { throw StateError('Message was larger than the allowed size!'); } bytesBuilder.add([ length >> 24 & 0xff, length >> 16 & 0xff, length >> 8 & 0xff, length & 0xff ]); } String childProgram(SerializationMode mode) => ''' import 'dart:convert'; import 'dart:io'; import 'dart:isolate'; import 'dart:typed_data'; import 'package:_fe_analyzer_shared/src/macros/executor/message_grouper.dart'; import 'package:_fe_analyzer_shared/src/macros/executor/serialization.dart'; void main(List args, [SendPort? sendPort]) async { var mode = $mode; await withSerializationMode(mode, () async { if (sendPort != null) { var isolateReceivePort = ReceivePort(); isolateReceivePort.listen((data) { deserialize(data); var result = serialize(); result = result is Uint8List ? TransferableTypedData.fromList([result]) : result; sendPort.send(result); }); sendPort.send(isolateReceivePort.sendPort); } else if (args.isNotEmpty) { var address = args[0]; var port = int.parse(args[1]); var socket = await Socket.connect(address, port); if (mode == SerializationMode.json) { socket.listen((data) { var json = utf8.decode(data).trimRight(); deserialize(jsonDecode(json)); socket.write(jsonEncode(serialize())); }); } else { MessageGrouper(socket).messageStream.listen((data) { deserialize(data); var result = serialize() as Uint8List; final bytesBuilder = BytesBuilder(copy: false); _writeLength(result, bytesBuilder); bytesBuilder.add(result); socket.add(bytesBuilder.takeBytes()); }); } } else { // We allow one empty line to work around some weird data. var allowEmpty = true; if (mode == SerializationMode.json) { stdin.listen((data) { var json = utf8.decode(data).trimRight(); // On exit we tend to get extra empty lines sometimes? if (json.isEmpty && allowEmpty) { allowEmpty = false; return; } deserialize(jsonDecode(json)); stdout.write(jsonEncode(serialize())); }); } else { MessageGrouper(stdin).messageStream.listen((data) { deserialize(data); var result = serialize() as Uint8List; final bytesBuilder = BytesBuilder(copy: false); _writeLength(result, bytesBuilder); bytesBuilder.add(result); stdout.add(bytesBuilder.takeBytes()); }); } } }); } Object? serialize() { var serializer = serializerFactory(); for (var i = 0; i < 100; i++) { serializer.addInt(i * 100); serializer.addString('foo' * i); serializer.addBool(i % 2 == 0); serializer.startList(); for (var j = 0; j < 10; j++) { serializer.addDouble(i * 5); } serializer.endList(); serializer.addNull(); } return serializer.result; } void deserialize(Object? result) { result = result is TransferableTypedData ? result.materialize().asUint8List() : result; var deserializer = deserializerFactory(result); while (deserializer.moveNext()) { deserializer ..expectInt() ..moveNext() ..expectString() ..moveNext() ..expectBool() ..moveNext() ..expectList(); while (deserializer.moveNext()) { deserializer.expectDouble(); } deserializer ..moveNext() ..checkNull(); } } void _writeLength(Uint8List result, BytesBuilder bytesBuilder) { int length = result.lengthInBytes; if (length > 0xffffffff) { throw new StateError('Message was larger than the allowed size!'); } bytesBuilder.add([ length >> 24 & 0xff, length >> 16 & 0xff, length >> 8 & 0xff, length & 0xff ]); }'''; Object? serialize() { var serializer = serializerFactory(); for (var i = -50; i < 50; i++) { serializer.addInt(i % 2 * 100); serializer.addString('foo' * i); serializer.addBool(i < 0); serializer.startList(); for (var j = 0.0; j < 10; j++) { serializer.addDouble(i * j); } serializer.endList(); serializer.addNull(); } return serializer.result; } void deserialize(Object? result) { result = result is TransferableTypedData ? result.materialize().asUint8List() : result; if (serializationMode == SerializationMode.json) { if (result is List) { result = jsonDecode(utf8.decode(result)); } } var deserializer = deserializerFactory(result); while (deserializer.moveNext()) { deserializer ..expectInt() ..moveNext() ..expectString() ..moveNext() ..expectBool() ..moveNext() ..expectList(); while (deserializer.moveNext()) { deserializer.expectDouble(); } deserializer ..moveNext() ..checkNull(); } }