dart-lang/sdk
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

[vm] Create perf_witness package

Browse Source 
This package provides universal observability functionality for Dart
SDK CLI tooling. Any tool can opt-in into observability by starting
PerfWitnessServer. This will create a control socket in a fixed
location which can be then discovered by perf_witness recorder.

TEST=pkg/perf_witness/test

Change-Id: I698617a66fed42c6629c348d53964dae3f2148df
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/413683
Reviewed-by: Martin Kustermann <kustermann@google.com>
Commit-Queue: Slava Egorov <vegorov@google.com>
This commit is contained in:
Vyacheslav Egorov
2025-12-08 05:02:35 -08:00
committed by Commit Queue
parent 95df731257
commit cbdf85ca97
14 changed files with 1636 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/perf_witness/analysis_options.yaml
+11
View File
@@ -0,0 +1,11 @@
include: package:lints/recommended.yaml
analyzer:
exclude:
- lib/src/assets/**
linter:
rules:
- directives_ordering
- prefer_final_locals
- sort_pub_dependencies
pkg/perf_witness/bin/recorder.dart
+9
View File
@@ -0,0 +1,9 @@
// Copyright (c) 2025, 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 'package:perf_witness/recorder.dart' as recorder;
Future<void> main(List<String> args) async {
await recorder.record(recorder.PerfWitnessRecorderConfig.fromArgs(args));
}
pkg/perf_witness/example/example.dart
+42
View File
@@ -0,0 +1,42 @@
// Copyright (c) 2025, 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.
// Simple binary which continuously does some busy work and generates
// timeline events.
import 'dart:async';
import 'dart:developer';
import 'package:perf_witness/server.dart';
import 'package:perf_witness/src/async_span.dart';
int fib(int i) {
if (i < 2) return 1;
return fib(i - 1) + fib(i - 2);
}
Future<void> task(int id) async {
await AsyncSpan.run('task#$id', () async {
for (var i = 0; i < 10; i++) {
Timeline.timeSync('fib', () {
final sw = Stopwatch()..start();
while (sw.elapsedMilliseconds < 100) {
fib(10);
}
});
await Future.delayed(Duration(milliseconds: 100));
}
});
}
void main() async {
await PerfWitnessServer.start();
var id = 0;
while (true) {
await AsyncSpan.run('task-group-${id ~/ 2}', () async {
await Future.wait([task(id++), task(id++)]);
await Future.delayed(Duration(milliseconds: 100));
});
}
}
pkg/perf_witness/lib/recorder.dart
+240
View File
@@ -0,0 +1,240 @@
// Copyright (c) 2025, 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:developer';
import 'dart:io' as io;
import 'package:args/args.dart';
import 'package:path/path.dart' as p;
import 'src/common.dart';
import 'src/json_rpc.dart';
import 'src/process_info.dart';
class PerfWitnessRecorderConfig {
final String? outputDir;
final String? tag;
final bool recordNewProcesses;
final bool enableAsyncSpans;
final bool enableProfiler;
final List<String> streams;
PerfWitnessRecorderConfig({
this.outputDir,
this.tag,
this.recordNewProcesses = false,
this.enableAsyncSpans = false,
this.enableProfiler = true,
this.streams = const [],
});
factory PerfWitnessRecorderConfig.fromParsedArgs(ArgResults args) {
var streams = args['streams'] as List<String>;
if (streams.contains('all')) {
streams = TimelineStream.values.map((s) => s.name).toList();
}
return PerfWitnessRecorderConfig(
outputDir: args['output-dir'] as String?,
tag: args['tag'] as String?,
recordNewProcesses: args['record-new-processes'] as bool,
enableAsyncSpans: args['enable-async-spans'] as bool,
enableProfiler: args['enable-profiler'] as bool,
streams: streams,
);
}
factory PerfWitnessRecorderConfig.fromArgs(List<String> args) {
final parsedArgs = configureArgParser().parse(args);
return PerfWitnessRecorderConfig.fromParsedArgs(parsedArgs);
}
static ArgParser configureArgParser([ArgParser? parser]) {
return (parser ?? ArgParser())
..addOption('output-dir', abbr: 'o')
..addOption('tag', help: 'Tag to filter processes by.')
..addFlag(
'record-new-processes',
help: 'Record processes that start after the recorder.',
negatable: false,
)
..addFlag(
'enable-async-spans',
help: 'Enable async spans.',
negatable: false,
)
..addFlag(
'enable-profiler',
help: 'Enable profiler.',
negatable: true,
defaultsTo: true,
)
..addMultiOption(
'streams',
help: 'Streams to record.',
allowed: [...TimelineStream.values.map((s) => s.name), 'all'],
defaultsTo: [TimelineStream.gc.name, TimelineStream.dart.name],
);
}
}
Future<void> record(PerfWitnessRecorderConfig config) async {
final io.Directory outputDir;
if (config.outputDir case final String outputDirPath) {
outputDir = io.Directory(outputDirPath);
} else {
outputDir = io.Directory.systemTemp.createTempSync('recording');
}
final sockets = getAllControlSockets();
final connections = (await Future.wait([
for (var s in sockets) Connection._tryConnectTo(s.socketPath),
])).nonNulls.toList(growable: false);
print('Found ${connections.length} processes:');
for (final c in connections) {
print(' ${c.info}');
}
final matchedConnections = _closeNotMatching(connections, config.tag);
if (config.tag != null) {
print('Tag ${config.tag} matched ${matchedConnections.length} processes.');
}
print('... data will be written to $outputDir');
final sw = Stopwatch()..start();
await Future.wait([
for (var conn in matchedConnections)
conn.startRecording(outputDir.path, config: config),
]);
bool recording = true;
JsonRpcServer? newProcessServer;
if (config.recordNewProcesses) {
if (recorderSocketPath case final path?) {
if (io.FileSystemEntity.typeSync(path) ==
io.FileSystemEntityType.unixDomainSock) {
print(
'Warning: Control socket $path already exists '
'(another recorder might be running).',
);
} else {
newProcessServer = JsonRpcServer(await UnixDomainSocket.bind(path), {
'process.announce': (requestor, params) async {
if (!recording) {
return null;
}
final info = ProcessInfo.fromJson(params as Map<String, Object?>);
print('New process announced: $info');
if (config.tag == null || info.tag == config.tag) {
try {
final conn = Connection._(info, requestor);
matchedConnections.add(conn);
await conn.startRecording(outputDir.path, config: config);
} catch (e) {
print('Failed to start recording: $e');
}
}
return null;
},
});
print('Listening for new processes on $path');
}
} else {
print(
'Warning: Unable to listen for new processes '
'(path to the control socket is null).',
);
}
}
if (matchedConnections.isNotEmpty || config.recordNewProcesses) {
await io.ProcessSignal.sigint.watch().first;
recording = false;
await Future.wait([
for (var conn in matchedConnections)
conn.stopRecording().catchError((e) {
print('Failed to stop recording of process ${conn.info.pid}: $e');
}),
]);
print('Recorded for ${sw.elapsed}');
}
for (final conn in matchedConnections) {
conn.disconnect();
}
await newProcessServer?.close();
}
class Connection {
final ProcessInfo info;
final JsonRpcPeer _endpoint;
Connection._(this.info, this._endpoint);
Future<void> startRecording(
String outputDir, {
required PerfWitnessRecorderConfig config,
}) async {
await _endpoint.sendRequest('timeline.streamTo', {
'recorder': 'perfetto',
'path': p.join(outputDir, '${info.pid}.timeline'),
'enableProfiler': config.enableProfiler,
'enableAsyncSpans': config.enableAsyncSpans,
'streams': config.streams,
});
}
Future<void> stopRecording() async {
await _endpoint.sendRequest('timeline.stopStreaming');
}
void disconnect() async {
try {
await _endpoint.close();
} catch (_) {
// Ignore exceptions
}
}
static Future<Connection> connectTo(String controlSocketPath) async {
final client = jsonRpcPeerFromSocket(
await UnixDomainSocket.connect(controlSocketPath),
);
final info = ProcessInfo.fromJson(
await client.sendRequest('process.getInfo') as Map<String, Object?>,
);
return Connection._(info, client);
}
static Future<Connection?> _tryConnectTo(io.File controlSocket) async {
try {
return await Connection.connectTo(controlSocket.path);
} catch (_) {
try {
controlSocket.deleteSync(); // Likely stale file. Purge it.
} catch (_) {}
return null;
}
}
}
List<Connection> _closeNotMatching(List<Connection> v, String? tag) {
if (tag == null) {
return v.toList(growable: true);
}
final open = <Connection>[];
for (final c in v) {
if (c.info.tag == tag) {
open.add(c);
continue;
}
c.disconnect();
}
return open;
}
pkg/perf_witness/lib/server.dart
+201
View File
@@ -0,0 +1,201 @@
// Copyright (c) 2025, 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:developer' as developer;
import 'dart:ffi' as ffi;
import 'dart:io' as io;
import 'package:ffi/ffi.dart' show calloc;
import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
import 'src/common.dart';
import 'src/json_rpc.dart';
import 'src/process_info.dart';
class PerfWitnessServer {
final String? _tag;
final String _controlSocketPath;
final String _recorderSocketPath;
final ffi.Pointer<ffi.Bool> _isRecordingTimelineWithAsyncSpans;
bool _isRecordingTimeline = false;
JsonRpcServer? _server;
json_rpc.Peer? _recorderConnection;
static ffi.Pointer<ffi.Bool>? _sharedIsRecordingTimelineWithAsyncSpans;
late final Map<String, JsonRpcMethod> _methods = {
'process.getInfo': _getProcessInfo,
'timeline.streamTo': _timelineStreamTo,
'timeline.stopStreaming': _timelineStopStreaming,
'process._isRecordingTimelineWithAsyncSpansAddr':
_isRecordingTimelineWithAsyncSpansAddr,
};
static PerfWitnessServer? _instance;
PerfWitnessServer._(
this._tag,
this._controlSocketPath,
this._recorderSocketPath,
) : _isRecordingTimelineWithAsyncSpans = calloc(ffi.sizeOf<ffi.Bool>());
static Future<void> start({String? tag}) async {
if (_instance != null) {
return;
}
if (controlSocketPath case final socketPath?) {
if (io.FileSystemEntity.typeSync(socketPath) == .unixDomainSock) {
// Another isolate is already serving the process. We assume that
// server will remain open as long as the process is running.
// However we want to make sure that setting global settings (e.g.
// whether async spans are enabled or not) will affect all isolates
// not just the one that created the server.
final client = jsonRpcPeerFromSocket(
await UnixDomainSocket.connect(socketPath),
);
final {'address': int addr, 'pid': int pid} =
await client.sendRequest(
'process._isRecordingTimelineWithAsyncSpansAddr',
)
as Map<String, dynamic>;
// Just double check that we are the very same process.
if (pid != io.pid) {
return;
}
_sharedIsRecordingTimelineWithAsyncSpans = .fromAddress(addr);
return;
}
_instance = PerfWitnessServer._(tag, socketPath, recorderSocketPath!);
await _instance!._start();
}
}
static Future<void> shutdown() async {
await _instance?._shutdown();
_instance = null;
}
static bool get isRecordingTimelineWithAsyncSpans {
return _sharedIsRecordingTimelineWithAsyncSpans?.value ?? false;
}
Future<void> _timelineStreamTo(
json_rpc.Peer requestor,
Map<String, Object?>? params,
) async {
if (_isRecordingTimeline) {
throw StateError('Timeline is already being recorded');
}
final paramsObj = StreamTimelineToRequest(params ?? {});
final streams = paramsObj.streams
?.map((s) => developer.TimelineStream.values.byName(s))
.toList();
final samplingIntervalUs = paramsObj.samplingInterval;
final samplingInterval = samplingIntervalUs != null
? Duration(microseconds: samplingIntervalUs)
: const Duration(microseconds: 1000);
final enableAsyncSpans = paramsObj.enableAsyncSpans ?? false;
developer.NativeRuntime.streamTimelineTo(
developer.TimelineRecorder.values.byName(paramsObj.recorder),
path: paramsObj.path,
streams:
streams ??
const [developer.TimelineStream.dart, developer.TimelineStream.gc],
enableProfiler: paramsObj.enableProfiler ?? false,
samplingInterval: samplingInterval,
);
_isRecordingTimeline = true;
_isRecordingTimelineWithAsyncSpans.value = enableAsyncSpans;
}
Future<void> _timelineStopStreaming(
json_rpc.Peer requestor,
Map<String, Object?>? params,
) async {
if (!_isRecordingTimeline) {
throw StateError('Timeline is not being recorded');
}
developer.NativeRuntime.stopStreamingTimeline();
_isRecordingTimeline = false;
_isRecordingTimelineWithAsyncSpans.value = false;
}
Future<Map<String, Object?>> _getProcessInfo(
json_rpc.Peer requestor,
Map<String, Object?>? params,
) async {
return ProcessInfo.current(tag: _tag).toJson();
}
Future<Map<String, Object?>> _isRecordingTimelineWithAsyncSpansAddr(
json_rpc.Peer requestor,
Map<String, Object?>? params,
) async {
return {
'address': _isRecordingTimelineWithAsyncSpans.address,
'pid': io.pid,
};
}
Future<void> _start() async {
_sharedIsRecordingTimelineWithAsyncSpans =
_isRecordingTimelineWithAsyncSpans;
_server = JsonRpcServer(
await UnixDomainSocket.bind(_controlSocketPath),
_methods,
);
await _announceProcessTo(
_recorderSocketPath,
_tag,
).timeout(Duration(milliseconds: 100), onTimeout: () => Future.value());
}
Future<void> _announceProcessTo(String recorderPath, String? tag) async {
try {
_recorderConnection = jsonRpcPeerFromSocket(
await UnixDomainSocket.connect(recorderPath),
_methods,
);
await _recorderConnection!.sendRequest(
'process.announce',
ProcessInfo.current(tag: tag).toJson(),
);
} catch (e) {
// ignore, recorder might not be running.
}
}
Future<void> _shutdown() async {
_recorderConnection?.close();
await _server?.close();
if (io.FileSystemEntity.typeSync(_controlSocketPath) != .notFound) {
io.File(_controlSocketPath).deleteSync();
}
calloc.free(_isRecordingTimelineWithAsyncSpans);
if (_isRecordingTimeline) {
developer.NativeRuntime.stopStreamingTimeline();
_isRecordingTimeline = false;
}
}
}
extension type StreamTimelineToRequest(Map<String, Object?> json) {
String get recorder => json['recorder'] as String;
String? get path => json['path'] as String?;
List<String>? get streams => (json['streams'] as List?)?.cast<String>();
bool? get enableProfiler => json['enableProfiler'] as bool?;
int? get samplingInterval => json['samplingInterval'] as int?;
bool? get enableAsyncSpans => json['enableAsyncSpans'] as bool?;
}
pkg/perf_witness/lib/src/async_span.dart
+183
View File
@@ -0,0 +1,183 @@
// Copyright (c) 2025, 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:developer';
import '../../server.dart';
/// With synchronous execution the nesting between spans is naturally induced
/// by the callstack. Consider:
///
/// ```dart
/// Timeline.timeSync('a', () {
/// work();
/// Timeline.timeSync('b', () {
/// work();
/// });
/// work();
/// Timeline.timeSync('c', () {
/// work();
/// });
/// work();
/// })
/// ```
//
/// This will created three spans `a`, `b` and `c` all properly nested. The time
/// outside of `b` and `c` will be correctly attributed to `a`.
//
/// However the same is not easy to achieve for async computations. Compare:
///
/// ```
/// void a() async {
/// work();
/// await b();
/// work();
/// await c();
/// work();
/// }
/// ```
//
/// There is no functionality available in `dart:developer` which would allow
/// to create proper span structure to automatically accurately capture the
/// work done in `a`, `b` and `c`. The best you can do is to manually wrap
/// synchronous parts of work into `timeSync`.
///
/// This class tries to help with this by creating a `Zone` which automatically
/// does this - though result still might be confusing: completion of async task
/// causes resumption of async task that awaits on the current task which creates
/// inversely nested spans (e.g. if `b` is suspended and completes
/// asynchronously you get span `a` nested inside span `b` - even though you
/// would like an opposite picture or worst case you want these spans to be
/// siblings).
///
/// Wrapping execution in a `Zone` is expensive so we only enable it if
/// recorder requests it explicitly. When disabled we only emit timeline
/// spans for the first synchronous portion of the computation and then
/// a instantaneous span for the completion. This allows developer to
/// estimate how long asynchronous action took - but it will not actually
/// reveal when it was actively running on the stack.
class AsyncSpan {
final String name;
final Map<String, Object?>? parameters;
final Flow _flow = Flow.begin();
bool issuedBegin = false;
int running = 0;
AsyncSpan._(this.name, {this.parameters});
static AsyncSpan of(Zone zone) => zone[AsyncSpan] as AsyncSpan;
static final _zoneSpecification = ZoneSpecification(
run: <R>(self, parent, zone, R Function() f) {
final span = AsyncSpan.of(self);
span.startSync();
try {
return parent.run(zone, f);
} finally {
span.finishSync();
}
},
runUnary: <R, T1>(self, parent, zone, R Function(T1) f, T1 a1) {
final span = AsyncSpan.of(self);
span.startSync();
try {
return parent.runUnary(zone, f, a1);
} finally {
span.finishSync();
}
},
runBinary:
<R, T1, T2>(self, parent, zone, R Function(T1, T2) f, T1 a1, T2 a2) {
final span = AsyncSpan.of(self);
span.startSync();
try {
return parent.runBinary(zone, f, a1, a2);
} finally {
span.finishSync();
}
},
);
static Future<R> run<R>(
String name,
Future<R> Function() action, {
Map<String, Object?>? parameters,
}) async {
if (PerfWitnessServer.isRecordingTimelineWithAsyncSpans) {
return AsyncSpan._create(name, parameters: parameters).run(action);
} else {
final Future<R> result;
final flow = Flow.begin();
try {
Timeline.startSync(name, flow: flow);
result = action();
} finally {
Timeline.finishSync();
}
try {
return await result;
} finally {
Timeline.startSync(name, flow: Flow.end(flow.id));
Timeline.finishSync();
}
}
}
static Future<R> runUnary<R, T>(
String name,
Future<R> Function(T) action,
T arg, {
Map<String, Object?>? parameters,
}) async {
if (PerfWitnessServer.isRecordingTimelineWithAsyncSpans) {
return AsyncSpan._create(
name,
parameters: parameters,
).runUnary(action, arg);
} else {
final Future<R> result;
final flow = Flow.begin();
try {
Timeline.startSync(name, flow: flow);
result = action(arg);
} finally {
Timeline.finishSync();
}
try {
return await result;
} finally {
Timeline.startSync(name, flow: Flow.end(flow.id));
Timeline.finishSync();
}
}
}
static Zone _create(String name, {Map<String, Object?>? parameters}) =>
Zone.current.fork(
specification: _zoneSpecification,
zoneValues: {AsyncSpan: AsyncSpan._(name, parameters: parameters)},
);
void startSync() {
if (running == 0) {
Timeline.startSync(
name,
flow: issuedBegin ? Flow.step(_flow.id) : _flow,
arguments: issuedBegin ? null : parameters,
);
issuedBegin = true;
}
running++;
}
void finishSync() {
if (--running == 0) {
Timeline.finishSync();
}
}
}
pkg/perf_witness/lib/src/common.dart
+75
View File
@@ -0,0 +1,75 @@
// Copyright (c) 2025, 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:io' as io;
import 'package:dart_data_home/dart_data_home.dart';
import 'package:path/path.dart' as p;
final String? _controlSocketsDirectory = () {
final dir = getDartDataHome('perf');
try {
// Ensure that directory exists.
io.Directory(dir).createSync(recursive: true);
return dir;
} catch (_) {
// Ignore any sort of exceptions.
return null;
}
}();
List<({int pid, io.File socketPath})> getAllControlSockets() {
if (_controlSocketsDirectory == null) {
return const [];
}
try {
final allPidFiles = io.Directory(
_controlSocketsDirectory!,
).listSync().whereType<io.File>();
return [
for (var file in allPidFiles)
if (int.tryParse(p.basenameWithoutExtension(file.path)) case final pid?)
(pid: pid, socketPath: file),
];
} catch (_) {
// Ignore
return [];
}
}
final String? controlSocketPath = () {
final dirPath = _controlSocketsDirectory;
if (dirPath == null) {
return null;
}
return p.join(dirPath, '${io.pid}');
}();
final String? recorderSocketPath = () {
final dirPath = _controlSocketsDirectory;
if (dirPath == null) {
return null;
}
return p.join(dirPath, 'rec');
}();
abstract class UnixDomainSocket {
static Future<io.Socket> connect(String path) => io.Socket.connect(
io.InternetAddress(path, type: io.InternetAddressType.unix),
0,
);
static Future<io.ServerSocket> bind(String path) {
if (io.FileSystemEntity.typeSync(path) !=
io.FileSystemEntityType.notFound) {
io.File(path).deleteSync();
}
return io.ServerSocket.bind(
io.InternetAddress(path, type: io.InternetAddressType.unix),
0,
);
}
}
pkg/perf_witness/lib/src/json_rpc.dart
+71
View File
@@ -0,0 +1,71 @@
// Copyright (c) 2025, 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' as io;
import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
import 'package:stream_channel/stream_channel.dart';
typedef JsonRpcPeer = json_rpc.Peer;
typedef JsonRpcException = json_rpc.RpcException;
typedef JsonRpcMethod =
FutureOr<Object?> Function(
JsonRpcPeer requestor,
Map<String, Object?>? params,
);
JsonRpcPeer jsonRpcPeerFromSocket(
io.Socket socket, [
Map<String, JsonRpcMethod>? methods,
]) {
final lineChannel = StreamChannel<String>(
const LineSplitter().bind(utf8.decoder.bind(socket)),
StreamController<String>(sync: true, onCancel: socket.close)
..stream.listen((line) {
socket.write(line);
socket.write('\n');
}),
);
final peer = json_rpc.Peer(lineChannel);
if (methods != null) {
for (final MapEntry(:key, :value) in methods.entries) {
peer.registerMethod(key, (json_rpc.Parameters params) {
return value(
peer,
params.value == null ? null : params.asMap.cast<String, Object?>(),
);
});
}
}
peer.listen().ignore();
return peer;
}
class JsonRpcServer {
final io.ServerSocket _serverSocket;
final _endpoints = <JsonRpcPeer>{};
JsonRpcServer(this._serverSocket, [Map<String, JsonRpcMethod>? methods]) {
_serverSocket.listen((client) {
final endpoint = jsonRpcPeerFromSocket(client, methods);
_endpoints.add(endpoint);
endpoint.done.whenComplete(() {
_endpoints.remove(endpoint);
});
});
}
/// Returns a list of currently connected endpoints.
List<JsonRpcPeer> get endpoints => _endpoints.toList();
Future<void> close() async {
await Future.wait(_endpoints.toList().map((e) => e.close()));
await _serverSocket.close();
}
}
pkg/perf_witness/lib/src/process_info.dart
+57
View File
@@ -0,0 +1,57 @@
// Copyright (c) 2025, 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:io' as io;
class ProcessInfo {
final int pid;
final String command;
final String script;
final String dartBinary;
final int rss;
final String? tag;
ProcessInfo({
required this.pid,
required this.command,
required this.script,
required this.dartBinary,
required this.rss,
this.tag,
});
ProcessInfo.current({String? tag})
: this(
pid: io.pid,
command: io.Platform.executableArguments.join(' '),
script: io.Platform.script.toFilePath(),
dartBinary: io.Platform.executable,
rss: io.ProcessInfo.currentRss,
tag: tag,
);
factory ProcessInfo.fromJson(Map<String, dynamic> json) => ProcessInfo(
pid: json['pid'] as int,
command: json['command'] as String,
script: json['script'] as String,
dartBinary: json['dartBinary'] as String,
rss: json['rss'] as int,
tag: json['tag'] as String?,
);
Map<String, dynamic> toJson() => {
'pid': pid,
'command': command,
'script': script,
'dartBinary': dartBinary,
'rss': rss,
if (tag != null) 'tag': tag,
};
@override
String toString() {
final tagString = tag != null ? ' (tag: $tag)' : '';
return '[PID: $pid, script: $script, RSS: $rss]$tagString';
}
}
pkg/perf_witness/pubspec.yaml
+26
View File
@@ -0,0 +1,26 @@
name: perf_witness
version: 0.0.1
description: Shared performance observability infrastructure for Dart CLI tools
repository: https://github.com/dart-lang/sdk/tree/main/pkg/perf_witness
publish_to: none
resolution: workspace
environment:
sdk: ^3.11.0-edge
dependencies:
args: any
dart_data_home: any
ffi: any
json_rpc_2: any
path: any
stream_channel: any
# We use 'any' version constraints here as we get our package versions from
# the dart-lang/sdk repo's DEPS file. Note that this is a special case; the
# best practice for packages is to specify their compatible version ranges.
# See also https://dart.dev/tools/pub/dependencies.
dev_dependencies:
lints: any
test: any
vm_service_protos: any
pkg/perf_witness/test/common/busy_loop.dart
+59
View File
@@ -0,0 +1,59 @@
// Copyright (c) 2025, 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:io';
import 'dart:isolate';
import 'package:args/args.dart';
import 'package:perf_witness/server.dart';
import 'package:perf_witness/src/async_span.dart';
final parser = ArgParser()
..addOption('tag', abbr: 't', help: 'Tag for the process')
..addFlag(
'start-isolate',
abbr: 'i',
help: 'Start test isolate',
defaultsTo: false,
);
bool shouldStop = false;
Future<void> busyLoop({required String name}) async {
print('[$name] BUSY LOOP READY');
while (!shouldStop) {
await AsyncSpan.run('sleep', () async {
print(
'[$name] AsyncSpan.create is nop: ${identical(Zone.current, Zone.root)}',
);
await Future.delayed(const Duration(milliseconds: 500));
});
}
print('done');
}
void main(List<String> args) async {
ProcessSignal.sigint.watch().listen((_) {
print('SIGINT received');
shouldStop = true;
});
final parsedArgs = parser.parse(args);
final tag = parsedArgs['tag'] as String?;
await PerfWitnessServer.start(tag: tag);
if (parsedArgs.flag('start-isolate')) {
Isolate.run(() async {
await PerfWitnessServer.start(tag: tag);
await busyLoop(name: 'child-isolate');
}).onError((e, s) {
print('Isolate error: $e');
print(s);
exit(1);
});
}
await busyLoop(name: 'main');
await PerfWitnessServer.shutdown();
exit(0);
}
pkg/perf_witness/test/json_rpc_test.dart
+94
View File
@@ -0,0 +1,94 @@
// Copyright (c) 2025, 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:io' as io;
import 'package:path/path.dart' as p;
import 'package:perf_witness/src/common.dart';
import 'package:perf_witness/src/json_rpc.dart';
import 'package:test/test.dart';
void main() {
group('JsonRpc', () {
late JsonRpcServer server;
late JsonRpcPeer client;
late String socketPath;
setUp(() async {
socketPath = p.join(
io.Directory.systemTemp.createTempSync().path,
'test.sock',
);
server = JsonRpcServer(await UnixDomainSocket.bind(socketPath), {
'testMethod': (requestor, params) => 'Hello, ${params!['name']}',
'errorMethod': (requestor, params) => throw 'Something went wrong',
'ping': (requestor, params) => 'pong',
'checkEndpoint': (requestor, params) {
expect(server.endpoints, contains(requestor));
return 'ok';
},
});
client =
jsonRpcPeerFromSocket(await UnixDomainSocket.connect(socketPath), {
'reverse': (requestor, params) =>
(params!['text'] as String).split('').reversed.join(),
});
});
tearDown(() async {
await client.close();
await server.close();
final file = io.File(socketPath);
if (file.existsSync()) {
file.deleteSync();
}
});
test('can make a successful request', () async {
final result = await client.sendRequest('testMethod', {'name': 'World'});
expect(result, 'Hello, World');
});
test('handles method not found', () async {
try {
await client.sendRequest('nonExistentMethod');
fail('Expected an error');
} catch (e) {
expect(e, isA<JsonRpcException>());
expect(
(e as JsonRpcException).message,
'Unknown method "nonExistentMethod".',
);
}
});
test('handles internal server error', () async {
try {
await client.sendRequest('errorMethod');
fail('Expected an error');
} catch (e) {
expect(e, isA<JsonRpcException>());
expect((e as JsonRpcException).message, 'Something went wrong');
}
});
test('can make bidirectional requests', () async {
// Client calls server
expect(await client.sendRequest('ping'), 'pong');
// Server calls client
// Wait for the server to accept the connection.
while (server.endpoints.isEmpty) {
await Future.delayed(const Duration(milliseconds: 10));
}
final endpoint = server.endpoints.first;
expect(await endpoint.sendRequest('reverse', {'text': 'hello'}), 'olleh');
});
test('method receives correct endpoint', () async {
expect(await client.sendRequest('checkEndpoint'), equals('ok'));
});
});
}
pkg/perf_witness/test/recorder_server_test.dart
+567
View File
@@ -0,0 +1,567 @@
// Copyright (c) 2025, 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' as io;
import 'dart:isolate';
import 'package:path/path.dart' as p;
import 'package:test/test.dart';
import 'package:vm_service_protos/vm_service_protos.dart';
final packageRoot = p.dirname(
p.dirname(
Isolate.resolvePackageUriSync(
Uri.parse('package:perf_witness/server.dart'),
)!.toFilePath(),
),
);
final testsDir = p.join(packageRoot, 'test');
final binDir = p.join(packageRoot, 'bin');
Future<io.Process> runProcess(
String executable,
List<String> arguments, {
String tag = '',
String? waitFor,
Map<String, String>? environment,
List<String>? stdout,
}) async {
final ready = Completer();
final process = await io.Process.start(
executable,
arguments,
environment: environment,
);
process.stdout.transform(Utf8Decoder()).transform(LineSplitter()).listen((
line,
) {
if (waitFor != null && !ready.isCompleted && line.contains(waitFor)) {
ready.complete();
}
print('[$tag]stdout> $line');
stdout?.add(line);
});
process.stderr.transform(Utf8Decoder()).transform(LineSplitter()).listen((
line,
) {
print('[$tag]stderr> $line');
});
if (waitFor != null) {
await ready.future;
}
return process;
}
class BusyLoopProcess {
final io.Process process;
final String tag;
final List<String> stdout;
BusyLoopProcess._(this.process, this.tag, this.stdout);
static Future<BusyLoopProcess> start(
String tag,
io.Directory tempDir, {
bool startIsolate = false,
}) async {
final stdout = <String>[];
return BusyLoopProcess._(
await runProcess(
io.Platform.executable,
[
'run',
p.join(testsDir, 'common', 'busy_loop.dart'),
'--tag',
tag,
if (startIsolate) '--start-isolate',
],
tag: 'busy-loop($tag)',
waitFor: 'BUSY LOOP READY',
environment: {'DART_DATA_HOME': tempDir.path},
stdout: stdout,
),
tag,
stdout,
);
}
int get pid => process.pid;
void kill() {
process.kill();
}
}
class RecorderProcess {
final io.Process process;
RecorderProcess._(this.process);
static Future<RecorderProcess> start(
io.Directory tempDir,
io.Directory outputDir, {
String? tag,
bool recordNewProcesses = false,
bool enableAsyncSpans = false,
bool enableProfiler = true,
List<String> streams = const ['dart', 'gc'],
String? waitFor,
}) async {
return RecorderProcess._(
await runProcess(
io.Platform.executable,
[
'run',
p.join(binDir, 'recorder.dart'),
'-o',
outputDir.path,
if (tag != null) ...['--tag', tag],
if (recordNewProcesses) '--record-new-processes',
if (enableAsyncSpans) '--enable-async-spans',
if (!enableProfiler) '--no-enable-profiler',
if (streams != const ['dart', 'gc']) ...[
'--streams',
streams.join(','),
],
],
tag: 'recorder',
environment: {'DART_DATA_HOME': tempDir.path},
waitFor: waitFor,
),
);
}
Future<void> stop() async {
process.kill(io.ProcessSignal.sigint);
if (await process.exitCode case final int exitCode when exitCode != 0) {
throw Exception('Recorder process failed with exit code $exitCode');
}
}
}
void main() {
group('Recorder and Server', () {
late BusyLoopProcess busyLoopProcess;
late io.Directory tempDir;
setUp(() async {
tempDir = io.Directory.systemTemp.createTempSync();
busyLoopProcess = await BusyLoopProcess.start('busy-loop-tag', tempDir);
});
tearDown(() {
busyLoopProcess.kill();
tempDir.deleteSync(recursive: true);
});
test('end-to-end test with recorder script', () async {
final outputDir = io.Directory('${tempDir.path}/output')..createSync();
// Run the recorder in a separate process.
final recorder = await RecorderProcess.start(tempDir, outputDir);
await Future.delayed(const Duration(seconds: 2));
await recorder.stop();
final timelineFiles = outputDir
.listSync()
.whereType<io.File>()
.where((file) => file.path.endsWith('.timeline'))
.toList();
final timelines = timelineFiles.map((e) => p.basename(e.path)).toList();
expect(
timelines,
equals(['${busyLoopProcess.pid}.timeline']),
reason: 'Expected timeline file to be created',
);
final trace = Trace()
..mergeFromBuffer(timelineFiles.first.readAsBytesSync());
expect(trace.packet, isNotEmpty);
expect(trace.packet.any((p) => p.hasPerfSample()), isTrue);
// Dart track should be enabled by default.
expect(extractSeenEvents(trace), containsAll(['sleep']));
});
test('end-to-end test with recorder script - early exit', () async {
final outputDir = io.Directory('${tempDir.path}/output')..createSync();
// Run the recorder in a separate process.
final recorder = await RecorderProcess.start(tempDir, outputDir);
await Future.delayed(const Duration(seconds: 2));
busyLoopProcess.process.kill(io.ProcessSignal.sigint);
await busyLoopProcess.process.exitCode;
await recorder.stop();
final timelineFiles = outputDir
.listSync()
.whereType<io.File>()
.where((file) => file.path.endsWith('.timeline'))
.toList();
final timelines = timelineFiles.map((e) => p.basename(e.path)).toList();
expect(
timelines,
equals(['${busyLoopProcess.pid}.timeline']),
reason: 'Expected timeline file to be created',
);
final trace = Trace()
..mergeFromBuffer(timelineFiles.first.readAsBytesSync());
expect(trace.packet, isNotEmpty);
expect(trace.packet.any((p) => p.hasPerfSample()), isTrue);
// Dart track should be enabled by default.
expect(extractSeenEvents(trace), containsAll(['sleep']));
});
test('profiler can be disabled', () async {
final outputDir = io.Directory('${tempDir.path}/output')..createSync();
// Run the recorder in a separate process.
final recorder = await RecorderProcess.start(
tempDir,
outputDir,
enableProfiler: false,
);
await Future.delayed(const Duration(seconds: 2));
await recorder.stop();
final timelineFiles = outputDir
.listSync()
.whereType<io.File>()
.where((file) => file.path.endsWith('.timeline'))
.toList();
final timelines = timelineFiles.map((e) => p.basename(e.path)).toList();
expect(
timelines,
equals(['${busyLoopProcess.pid}.timeline']),
reason: 'Expected timeline file to be created',
);
final trace = Trace()
..mergeFromBuffer(timelineFiles.first.readAsBytesSync());
expect(trace.packet, isNotEmpty);
expect(trace.packet.any((p) => p.hasPerfSample()), isFalse);
});
test('streams can be configured', () async {
final outputDir = io.Directory('${tempDir.path}/output')..createSync();
// Run the recorder in a separate process.
final recorder = await RecorderProcess.start(
tempDir,
outputDir,
enableProfiler: false,
streams: ['isolate', 'compiler'],
);
await Future.delayed(const Duration(seconds: 2));
await recorder.stop();
final timelineFiles = outputDir
.listSync()
.whereType<io.File>()
.where((file) => file.path.endsWith('.timeline'))
.toList();
final timelines = timelineFiles.map((e) => p.basename(e.path)).toList();
expect(
timelines,
equals(['${busyLoopProcess.pid}.timeline']),
reason: 'Expected timeline file to be created',
);
final trace = Trace()
..mergeFromBuffer(timelineFiles.first.readAsBytesSync());
expect(trace.packet, isNotEmpty);
expect(trace.packet.any((p) => p.hasPerfSample()), isFalse);
final seenEvents = extractSeenEvents(trace);
expect(seenEvents, containsAll(['HandleMessage', 'CompileFunction']));
// Dart trace is disabled.
expect(seenEvents, isNot(contains('sleep')));
});
test('tag filtering positive test', () async {
final outputDir = io.Directory('${tempDir.path}/output')..createSync();
// Run the recorder in a separate process.
final recorder = await RecorderProcess.start(
tempDir,
outputDir,
tag: 'busy-loop-tag',
);
await Future.delayed(const Duration(seconds: 2));
await recorder.stop();
final timelines = outputDir
.listSync()
.map((e) => p.basename(e.path))
.toList();
expect(
timelines,
equals(['${busyLoopProcess.pid}.timeline']),
reason: 'Expected timeline file to be created',
);
});
test('tag filtering negative test', () async {
final outputDir = io.Directory('${tempDir.path}/output')..createSync();
// Run the recorder in a separate process.
final recorder = await RecorderProcess.start(
tempDir,
outputDir,
tag: 'unmatched-tag',
);
await Future.delayed(const Duration(seconds: 2));
await recorder.stop();
final timelines = outputDir
.listSync()
.map((e) => p.basename(e.path))
.toList();
expect(
timelines,
isEmpty,
reason: 'Expected no timeline file to be created',
);
});
test('async spans are not activated by default', () async {
final outputDir = io.Directory('${tempDir.path}/output')..createSync();
final busyLoopWithIsolate = await BusyLoopProcess.start(
'busy-loop-with-isolate-tag',
tempDir,
startIsolate: true,
);
// Run the recorder in a separate process.
final recorder = await RecorderProcess.start(tempDir, outputDir);
await Future.delayed(const Duration(seconds: 2));
await recorder.stop();
busyLoopWithIsolate.kill();
expect(
busyLoopProcess.stdout,
contains('[main] AsyncSpan.create is nop: true'),
);
expect(
busyLoopProcess.stdout,
isNot(contains('[main] AsyncSpan.create is nop: false')),
);
expect(
busyLoopProcess.stdout,
isNot(contains('[child-isolate] AsyncSpan.create is nop: true')),
);
expect(
busyLoopProcess.stdout,
isNot(contains('[child-isolate] AsyncSpan.create is nop: false')),
);
expect(
busyLoopWithIsolate.stdout,
contains('[main] AsyncSpan.create is nop: true'),
);
expect(
busyLoopWithIsolate.stdout,
isNot(contains('[main] AsyncSpan.create is nop: false')),
);
expect(
busyLoopWithIsolate.stdout,
contains('[child-isolate] AsyncSpan.create is nop: true'),
);
expect(
busyLoopWithIsolate.stdout,
isNot(contains('[child-isolate] AsyncSpan.create is nop: false')),
);
final timelines = outputDir
.listSync()
.map((e) => p.basename(e.path))
.toList();
expect(
timelines,
unorderedEquals([
'${busyLoopProcess.pid}.timeline',
'${busyLoopWithIsolate.pid}.timeline',
]),
reason: 'Expected timeline file to be created',
);
});
test('async spans are activated when requested', () async {
final outputDir = io.Directory('${tempDir.path}/output')..createSync();
final busyLoopWithIsolate = await BusyLoopProcess.start(
'busy-loop-with-isolate-tag',
tempDir,
startIsolate: true,
);
// Run the recorder in a separate process.
final recorder = await RecorderProcess.start(
tempDir,
outputDir,
enableAsyncSpans: true,
);
await Future.delayed(const Duration(seconds: 2));
await recorder.stop();
busyLoopWithIsolate.kill();
expect(
busyLoopProcess.stdout,
contains('[main] AsyncSpan.create is nop: false'),
);
expect(
busyLoopProcess.stdout,
isNot(contains('[child-isolate] AsyncSpan.create is nop: true')),
);
expect(
busyLoopProcess.stdout,
isNot(contains('[child-isolate] AsyncSpan.create is nop: false')),
);
expect(
busyLoopWithIsolate.stdout,
contains('[main] AsyncSpan.create is nop: false'),
);
expect(
busyLoopWithIsolate.stdout,
contains('[child-isolate] AsyncSpan.create is nop: false'),
);
final timelines = outputDir
.listSync()
.map((e) => p.basename(e.path))
.toList();
expect(
timelines,
unorderedEquals([
'${busyLoopProcess.pid}.timeline',
'${busyLoopWithIsolate.pid}.timeline',
]),
reason: 'Expected timeline file to be created',
);
});
test('record new processes - all', () async {
final outputDir = io.Directory('${tempDir.path}/output')..createSync();
// Run the recorder in a separate process.
final recorder = await RecorderProcess.start(
tempDir,
outputDir,
recordNewProcesses: true,
waitFor: 'Listening for new processes',
);
// Start a new process that should be recorded.
final newProcess = await BusyLoopProcess.start(
'new-process-tag',
tempDir,
);
await Future.delayed(const Duration(seconds: 2));
await recorder.stop();
newProcess.kill();
final timelines = outputDir
.listSync()
.map((e) => p.basename(e.path))
.toList();
expect(
timelines,
unorderedEquals([
'${newProcess.pid}.timeline',
'${busyLoopProcess.pid}.timeline',
]),
);
}, timeout: Timeout(Duration(seconds: 15)));
test(
'record new processes - specific tag',
() async {
final outputDir = io.Directory('${tempDir.path}/output')..createSync();
// Run the recorder in a separate process.
final recorder = await RecorderProcess.start(
tempDir,
outputDir,
tag: 'new-process-tag',
recordNewProcesses: true,
);
// Start a new process that should be recorded.
final newProcess = await BusyLoopProcess.start(
'new-process-tag',
tempDir,
);
// Start a new process that should NOT be recorded.
final ignoredProcess = await BusyLoopProcess.start(
'ignored-tag',
tempDir,
);
await Future.delayed(const Duration(seconds: 2));
await recorder.stop();
newProcess.kill();
ignoredProcess.kill();
final timelines = outputDir
.listSync()
.map((e) => p.basename(e.path))
.toList();
expect(timelines, unorderedEquals(['${newProcess.pid}.timeline']));
},
timeout: Timeout(Duration(seconds: 15)),
);
});
}
class IncrementalState {
final eventNames = <int, String>{};
void update(InternedData internedData) {
for (var eventName in internedData.eventNames) {
eventNames[eventName.iid.toInt()] = eventName.name;
}
}
}
Set<String> extractSeenEvents(Trace trace) {
var state = IncrementalState();
final seenEvents = <String>{};
for (var packet in trace.packet) {
if ((packet.sequenceFlags &
TracePacket_SequenceFlags.SEQ_INCREMENTAL_STATE_CLEARED.value) !=
0) {
state = IncrementalState();
}
if (packet.hasInternedData()) {
state.update(packet.internedData);
}
if (packet.hasTrackEvent()) {
final trackEvent = packet.trackEvent;
if (trackEvent.type == TrackEvent_Type.TYPE_SLICE_BEGIN) {
final name = state.eventNames[packet.trackEvent.nameIid.toInt()]!;
seenEvents.add(name);
}
}
}
return seenEvents;
}
pubspec.yaml
+1
View File
@@ -63,6 +63,7 @@ workspace:
- pkg/native_compiler
- pkg/native_stack_traces
- pkg/node_preamble
- pkg/perf_witness
- pkg/reload_test
- pkg/scrape
- pkg/server_plugin