[ Service ] Add support for isolate-based RPCs to dart_runtime_service_vm

This change adds initial support for working with isolates in the Dart
Runtime Service and its backends.

The new `IsolateManager` base class tracks the set of active isolates
and their lifecycle events. The `VmIsolateManager` extends this class,
adding support specific to interacting with isolates within the Dart VM.

TEST=vm/cc/DartAPI_InvokeVMServiceMethod*_Exp

Change-Id: I3dfa298722c40dbdfdd58105cc78f31d058dd7a2
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/486560
Reviewed-by: Ryan Macnak <rmacnak@google.com>
Commit-Queue: Ben Konyi <bkonyi@google.com>
Reviewed-by: Jessy Yameogo <yjessy@google.com>
This commit is contained in:
Ben Konyi
2026-03-12 09:08:14 -07:00
committed by Commit Queue
parent b54f4b718e
commit a8aba7fab8
22 changed files with 902 additions and 75 deletions
+1
View File
@@ -61,6 +61,7 @@ group("runtime") {
deps += [
"runtime/bin:dartaotruntime",
"runtime/bin:dartaotruntime_product",
"utils/dart_runtime_service_vm",
"utils/dart_runtime_service_vm:dart_runtime_service_vm_aot_snapshot",
"utils/dartdev:dartdev_aot_snapshot",
"utils/dds:dds_aot",
+12
View File
@@ -360,6 +360,9 @@ trace to find the place to insert the appropriate support.
elif arg == 'gen/utils/bazel/kernel_worker.dart.dill':
self.extra_paths.add(self.rebase(arg))
return self.parse_kernel_worker()
elif arg == 'gen/utils/dart_runtime_service_vm/generate_dart_runtime_service_vm_snapshot.dart.dill':
self.extra_paths.add(self.rebase(arg))
return self.parse_generate_dart_runtime_service_vm_snapshot()
elif arg == 'gen/utils/dartdev/generate_dartdev_snapshot.dart.dill':
self.extra_paths.add(self.rebase(arg))
return self.parse_generate_dartdev_snapshot()
@@ -677,6 +680,15 @@ trace to find the place to insert the appropriate support.
else:
self.unsupported('kernel_worker', arg)
def parse_generate_dart_runtime_service_vm_snapshot(self):
while self.has_next_arg:
arg = self.next_arg()
if arg in ['--help']:
pass
else:
self.unsupported('generate_dart_runtime_service_vm_snapshot',
arg)
def parse_generate_dartdev_snapshot(self):
while self.has_next_arg:
arg = self.next_arg()
@@ -5,5 +5,7 @@
export 'src/dart_runtime_service.dart';
export 'src/dart_runtime_service_backend.dart';
export 'src/dart_runtime_service_options.dart';
export 'src/event_streams.dart';
export 'src/exceptions.dart';
export 'src/isolate_manager.dart';
export 'src/rpc_exceptions.dart';
+15 -3
View File
@@ -25,7 +25,9 @@ base class Client {
required UnmodifiableNamedLookup<Client> clients,
required EventStreamMethods eventStreamMethods,
required this.backend,
String? name,
}) {
_name = name ?? defaultClientName;
_clientPeer = json_rpc.Peer(connection, strictProtocolChecks: false);
_internalRpcs = DartRuntimeServiceRpcs(
clients: clients,
@@ -47,12 +49,15 @@ base class Client {
late final Future<void> done;
Future<void> initialize({required String namespace}) {
logger.info('Initializing...');
this.namespace = namespace;
registerRpcHandlers();
done = _listen().then((_) {
logger.info('Client connection closed.');
// Cleanup stream subscription state when the client disconnects.
_internalRpcs.eventStreamMethods.onClientDisconnect(this);
});
logger.info('Initialization complete.');
return done;
}
@@ -64,6 +69,7 @@ base class Client {
/// Called if the connection to the client should be closed.
@mustCallSuper
Future<void> close() async {
logger.info('Cleaning up.');
await _clientPeer.close();
}
@@ -157,8 +163,13 @@ base class Client {
/// Sets the name associated with this client.
///
/// If [n] is null, the client name is reset to [defaultClientName].
void setName(String? n) => _name = n ?? defaultClientName;
late String _name = defaultClientName;
void setName(String? n) {
final updated = n ?? defaultClientName;
logger.info('Changing client name to $updated.');
_name = updated;
}
late String _name;
}
/// Used for keeping track and managing clients that are connected to a given
@@ -185,12 +196,13 @@ base class ClientManager {
///
/// This should be called when a client connects to the service.
@mustCallSuper
Client addClient(StreamChannel<String> connection) {
Client addClient({required StreamChannel<String> connection, String? name}) {
final client = Client(
connection: connection,
clients: UnmodifiableNamedLookup(clients),
eventStreamMethods: eventStreamMethods,
backend: backend,
name: name,
);
final namespace = clients.add(client);
client.initialize(namespace: namespace).then((_) {
@@ -3,12 +3,14 @@
// BSD-style license that can be found in the LICENSE file.
import 'dart:async';
import 'dart:developer';
import 'dart:io';
import 'package:logging/logging.dart';
import 'package:meta/meta.dart';
import 'package:shelf/shelf.dart' as shelf;
import 'package:shelf/shelf_io.dart' as io;
import 'package:stream_channel/stream_channel.dart';
import 'clients.dart';
import 'dart_runtime_service_backend.dart';
@@ -19,32 +21,43 @@ import 'handlers.dart';
import 'utils.dart';
typedef RpcResponse = Map<String, Object?>;
typedef DartRuntimeServiceBackendBuilder =
DartRuntimeServiceBackend Function(DartRuntimeService);
class DartRuntimeService {
DartRuntimeService._({required this.config, required this.backend})
: authCode = config.disableAuthCodes ? null : generateSecret() {
DartRuntimeService._({
required this.config,
required DartRuntimeServiceBackendBuilder backendBuilder,
}) : authCode = config.disableAuthCodes ? null : generateSecret() {
if (config.enableLogging) {
_logger.onRecord.listen(stdout.writeln);
}
backend = backendBuilder(this);
}
static Future<DartRuntimeService> initialize({
required DartRuntimeServiceOptions config,
required DartRuntimeServiceBackend backend,
required DartRuntimeServiceBackendBuilder backendBuilder,
}) async {
final service = DartRuntimeService._(config: config, backend: backend);
final service = DartRuntimeService._(
config: config,
backendBuilder: backendBuilder,
);
await service._initialize();
return service;
}
final DartRuntimeServiceOptions config;
final DartRuntimeServiceBackend backend;
late final DartRuntimeServiceBackend backend;
/// The ws:// URI pointing to this [DartRuntimeService]'s server.
///
/// Throws [DartRuntimeServiceServerNotRunning] if the HTTP server is not
/// active.
///
/// It's possible that the returned [Uri] is no longer valid if the server
/// was recently shut down.
Uri get uri {
if (_server == null) {
throw const DartRuntimeServiceServerNotRunning();
@@ -58,6 +71,9 @@ class DartRuntimeService {
///
/// Throws [DartRuntimeServiceServerNotRunning] if the HTTP server is not
/// active.
///
/// It's possible that the returned [Uri] is no longer valid if the server
/// was recently shut down.
Uri get httpUri => uri.replace(scheme: 'http');
/// The sse:// URI pointing to this [DartRuntimeService]'s server.
@@ -67,6 +83,9 @@ class DartRuntimeService {
///
/// Throws [DartRuntimeServiceServerNotRunning] if the HTTP server is not
/// active.
///
/// It's possible that the returned [Uri] is no longer valid if the server
/// was recently shut down.
Uri get sseUri {
if (config.sseHandlerPath == null) {
throw StateError('SSE handler path not configured.');
@@ -92,11 +111,16 @@ class DartRuntimeService {
@visibleForTesting
late final eventStreamManager = EventStreamManager(
backend: backend,
clientsGetter: () => UnmodifiableNamedLookup(clientManager.clients),
);
// TODO(bkonyi): this should be protected by a mutex.
HttpServer? _server;
/// Returns true if the HTTP server is active.
bool get isServerRunning => _server != null;
/// Initializes the service's state without starting the web server.
Future<void> _initialize() async {
await backend.initialize();
@@ -117,15 +141,46 @@ class DartRuntimeService {
Logger.root.clearListeners();
}
/// Enables or disables the server based on the value of [enable].
///
/// [silenceOutput] is used to determine if the service will output
/// non-logging information to the terminal.
///
/// This is called when `dart:developer`'s [Service.controlWebServer] is
/// invoked.
// TODO(bkonyi): respect silenceOutput
Future<void> serverControl({
required bool enable,
bool? silenceOutput,
}) async {
// TODO(bkonyi): verify there's no race conditions
if (!enable && isServerRunning) {
await _shutdownServer();
} else if (enable && !isServerRunning) {
await _startServer();
}
}
/// Toggles the state of the HTTP server, enabling it if it's not running and
/// disabling it if it is running.
Future<void> toggleServer() async {
// TODO(bkonyi): verify there's no race conditions
if (_server != null) {
if (isServerRunning) {
await _shutdownServer();
} else {
await _startServer();
}
}
/// Creates an artificial client to process JSON-RPC requests from
/// non-standard sources (e.g., from native code).
void addArtificialClient({
required StreamChannel<String> connection,
required String name,
}) {
clientManager.addClient(connection: connection, name: name);
}
Future<void> _startServer() async {
if (_server != null) {
_logger.warning(
@@ -193,6 +248,11 @@ class DartRuntimeService {
await server.close();
}
/// Send a [StreamEvent] to subscribed clients.
void sendEvent({required StreamEvent event}) {
event.send(eventStreamMethods: eventStreamManager);
}
shelf.Handler _handlers() {
_logger.info('Building Shelf handlers.');
var pipeline = const shelf.Pipeline();
@@ -3,16 +3,28 @@
// BSD-style license that can be found in the LICENSE file.
import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
import 'package:meta/meta.dart';
import 'dart_runtime_service.dart';
import 'event_streams.dart';
import 'isolate_manager.dart';
/// A backend implementation of a service used to inject non-common
/// functionality into a [DartRuntimeService].
abstract class DartRuntimeServiceBackend {
abstract class DartRuntimeServiceBackend<IM extends IsolateManager> {
DartRuntimeServiceBackend({required this.frontend});
/// The active service frontend hosting this [DartRuntimeServiceBackend].
final DartRuntimeService frontend;
/// Manages and tracks the lifecycle of isolates for the backend.
IM get isolateManager;
/// Invoked by the [DartRuntimeService] when the service is initializing,
/// before the service's HTTP server is started.
///
/// The backend should not expect for this to be invoked more than once.
@mustCallSuper
Future<void> initialize();
/// Invoked by the [DartRuntimeService] once it has completely finished
@@ -38,6 +50,24 @@ abstract class DartRuntimeServiceBackend {
/// started.
Future<void> onServerStarted({required Uri httpUri, required Uri wsUri});
/// Invoked when [EventStreamManager.streamListen] is called and the first
/// client has subscribed to [streamId].
///
/// [params] contains all of the parameters sent as part of the
/// `streamListen` request.
///
/// Returns true when the stream was successfully listened to.
bool onStreamListen({
required String streamId,
required Map<String, Object?> params,
}) {
return true;
}
/// Invoked when [EventStreamManager.streamCancel] is called and there are no
/// more clients listening to [streamId].
void onStreamCancel({required String streamId}) {}
/// Invoked by the [DartRuntimeService] to register handlers for the RPCs
/// provided by the backend.
void registerRpcs(json_rpc.Peer clientPeer);
@@ -201,9 +201,13 @@ final class DartRuntimeServiceRpcs {
///
/// If the stream ID does not correspond with a known stream, an error
/// response may be returned.
RpcResponse streamListen(json_rpc.Parameters parameters) {
Future<RpcResponse> streamListen(json_rpc.Parameters parameters) async {
final stream = parameters[_kStreamId].asString;
eventStreamMethods.streamListen(client: client, streamId: stream);
eventStreamMethods.streamListen(
client: client,
streamId: stream,
params: parameters.asMap.cast<String, Object?>(),
);
return Success().toJson();
}
@@ -214,7 +218,7 @@ final class DartRuntimeServiceRpcs {
///
/// If the stream ID does not correspond with a known stream, an error
/// response may be returned.
RpcResponse streamCancel(json_rpc.Parameters parameters) {
Future<RpcResponse> streamCancel(json_rpc.Parameters parameters) async {
final streamId = parameters[_kStreamId].asString;
eventStreamMethods.streamCancel(client: client, streamId: streamId);
return Success().toJson();
@@ -10,6 +10,7 @@ import 'package:meta/meta.dart';
import 'package:vm_service/vm_service.dart';
import 'clients.dart';
import 'dart_runtime_service_backend.dart';
import 'rpc_exceptions.dart';
import 'utils.dart';
@@ -109,8 +110,16 @@ abstract interface class EventStreamMethods {
ServiceAlias alias,
);
/// Subscribes `client` to a stream.
void streamListen({required Client client, required String streamId});
/// Subscribes [client] to a stream.
///
/// [params] is the unaltered set of parameters included when `streamListen`
/// is invoked. Backends may use these additional parameters for special
/// behavior (e.g., changing the verbosity of responses).
void streamListen({
required Client client,
required String streamId,
required Map<String, Object?> params,
});
/// Unsubscribes `client` from a stream.
void streamCancel({required Client client, required String streamId});
@@ -122,13 +131,12 @@ abstract interface class EventStreamMethods {
/// Used for keeping track of stream subscription state and sending events to
/// clients subscribed to individual streams.
class EventStreamManager implements EventStreamMethods {
EventStreamManager({
required UnmodifiableNamedLookup<Client> Function() clientsGetter,
}) : _clientsGetter = clientsGetter;
EventStreamManager({required this._clientsGetter, required this._backend});
static const kStreamNotify = 'streamNotify';
final UnmodifiableNamedLookup<Client> Function() _clientsGetter;
final DartRuntimeServiceBackend _backend;
late final clients = _clientsGetter();
@visibleForTesting
@@ -225,14 +233,23 @@ class EventStreamManager implements EventStreamMethods {
/// Subscribes `client` to a stream.
@override
void streamListen({required Client client, required String streamId}) {
void streamListen({
required Client client,
required String streamId,
required Map<String, Object?> params,
}) {
assert(streamId.isNotEmpty);
// TODO(bkonyi): invoke backend stream handling logic.
final listeners = streamListeners.putIfAbsent(streamId, () => []);
if (listeners.contains(client)) {
RpcException.streamAlreadySubscribed.throwException();
}
listeners.add(client);
// Tell the backend to start sending events for this stream if this is the
// first listener.
if (listeners.length == 1) {
_backend.onStreamListen(streamId: streamId, params: params);
}
if (streamId == EventStreams.kService) {
// Send all previously registered service extensions when a client
// subscribes to the Service stream.
@@ -265,7 +282,11 @@ class EventStreamManager implements EventStreamMethods {
}
listeners.remove(client);
// TODO(bkonyi): invoke backend stream handling logic.
// Tell the backend to stop sending events for this stream if there's no
// more listeners.
if (listeners.isEmpty) {
_backend.onStreamCancel(streamId: streamId);
}
}
/// Cleanup stream subscriptions for `client` when it has disconnected.
@@ -61,7 +61,7 @@ Handler webSocketClientHandler({required ClientManager clientManager}) =>
// package:shelf_web_socket v2.
final logger = Logger('WebSocketHandler');
logger.info('New web socket connection. Creating $Client.');
clientManager.addClient(ws.cast<String>());
clientManager.addClient(connection: ws.cast<String>());
});
/// Creates a [Handler] for incoming SSE connections.
@@ -82,7 +82,7 @@ Handler sseClientHandler({
handler.connections.rest.listen((sseConnection) {
logger.info('New SSE connection. Creating $Client.');
clientManager.addClient(sseConnection);
clientManager.addClient(connection: sseConnection);
});
return handler.handler;
@@ -0,0 +1,178 @@
// Copyright (c) 2026, 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 'package:json_rpc_2/error_code.dart' as json_rpc_error;
import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
import 'package:logging/logging.dart';
import 'package:meta/meta.dart';
import 'dart_runtime_service.dart';
enum IsolateState {
start,
running,
pauseStart,
pauseExit,
pausePostRequest,
unknown,
}
/// Base class for representing the state of a running isolate.
base class RunningIsolate {
RunningIsolate({required this.id, required this.name})
: _state = IsolateState.unknown;
late final _logger = Logger('Isolate ($name)');
final String name;
final int id;
// ignore: unused_field, will be used for resume permission logic.
IsolateState _state;
/// Invoked when the isolate has shutdown.
///
/// Override this to clean up any state associated with this isolate.
@mustCallSuper
void shutdown() {
_logger.info('Shutting down.');
}
// State setters.
void pausedOnExit() => _stateChange(IsolateState.pauseExit);
void pausedOnStart() => _stateChange(IsolateState.pauseStart);
void pausedPostRequest() => _stateChange(IsolateState.pausePostRequest);
void resumed() => running();
void running() => _stateChange(IsolateState.running);
void started() => _stateChange(IsolateState.start);
void _stateChange(IsolateState updated) {
_logger.info('${_state.name} => ${updated.name}');
_state = updated;
}
@override
String toString() => 'Isolate(name: $name id: $id)';
}
/// This file contains functionality used to track the running state of
/// all isolates in a given Dart process.
///
/// [RunningIsolate] is a representation of a single live isolate and contains
/// running state information for that isolate. In addition, approvals from
/// clients used to synchronize isolate resuming across multiple clients are
/// tracked in this class.
///
/// The [IsolateManager] keeps track of all the isolates in the
/// target process and handles isolate lifecycle events including:
/// - Startup
/// - Shutdown
/// - Pauses
///
/// The [IsolateManager] also handles the `resume` RPC, which checks the
/// resume approvals in the target [RunningIsolate] to determine if the
/// isolate should be resumed or wait for additional approvals to be granted.
abstract base class IsolateManager {
final _logger = Logger('$IsolateManager');
final isolates = <int, RunningIsolate>{};
/// The ID of the root isolate.
///
/// Used to support the `isolates/root` isolate ID.
int? _rootIsolateId;
@mustCallSuper
Future<void> shutdown() async {
_logger.info('Shutting down.');
}
/// Forwards the RPC request for [method] to be handled in the context of an
/// isolate.
///
/// [params] is the set of parameters for the RPC and must include a valid
/// `isolateId`.
Future<RpcResponse> sendToIsolate({
required String method,
required Map<String, Object?> params,
});
/// Initializes state for a newly started isolate.
void isolateStarted({required RunningIsolate isolate}) {
_logger.info('Starting isolate: $isolate');
if (_rootIsolateId == null) {
// TODO(bkonyi): ensure this is a non-system isolate
_logger.info('$isolate is the root isolate.');
_rootIsolateId = isolate.id;
}
isolate.running();
isolates[isolate.id] = isolate;
}
/// Cleans up state for an isolate that has exited.
void isolateExited({required int id}) {
final isolate = isolates.remove(id);
if (isolate == null) {
_logger.warning(
'isolateExited called with id: $id, but the isolate is not registered. '
'Ignoring.',
);
return;
}
_logger.info('Isolate exited: $isolate');
isolate.shutdown();
}
/// Looks up a [RunningIsolate] based on the `isolateId` entry in [params].
///
/// If the isolate ID is malformed, a [json_rpc.RpcException] is thrown with
/// an invalid parameters error.
///
/// If the isolate ID is not associated with a running isolate, null is
/// returned.
RunningIsolate? lookupIsolateFromParams({
required String method,
required Map<String, Object?> params,
}) {
const kIsolateId = 'isolateId';
const kIsolateIdPrefix = 'isolates/';
assert(params.containsKey(kIsolateId));
final isolateIdParam = params[kIsolateId] as String;
Never throwInvalidIsolateId() => throw json_rpc.RpcException(
json_rpc_error.INVALID_PARAMS,
'Invalid params',
data: {
'details':
"$method: invalid '$kIsolateId' parameter: "
'$isolateIdParam',
},
);
if (!isolateIdParam.startsWith(kIsolateIdPrefix)) {
_logger.warning('Malformed $kIsolateId: $isolateIdParam');
throwInvalidIsolateId();
}
final isolateId = isolateIdParam.substring(kIsolateIdPrefix.length);
int id;
if (isolateId == 'root') {
if (_rootIsolateId == null) {
throwInvalidIsolateId();
}
id = _rootIsolateId!;
} else {
try {
id = int.parse(isolateId);
} on FormatException {
throwInvalidIsolateId();
}
}
return isolates[id];
}
}
@@ -21,7 +21,11 @@ enum RpcException {
const RpcException({required this.code, required this.message});
/// Throws a [json_rpc.RpcException] with [code] and [message].
Never throwException() => throw json_rpc.RpcException(code, message);
Never throwException() => throw toException();
/// Builds a [json_rpc.RpcException] with [code] and [message] without
/// throwing.
json_rpc.RpcException toException() => json_rpc.RpcException(code, message);
/// The JSON-RPC error code.
final int code;
+1 -1
View File
@@ -3,7 +3,7 @@ name: dart_runtime_service
publish_to: none
environment:
sdk: ^3.8.0
sdk: ^3.12.0-edge
resolution: workspace
@@ -36,4 +36,21 @@ base class FakeDartRuntimeServiceBackend extends Fake
@override
void registerFallbacks(json_rpc.Peer clientPeer) {}
@override
DartRuntimeService get frontend => throw UnimplementedError();
@override
IsolateManager get isolateManager => throw UnimplementedError();
@override
void onStreamCancel({required String streamId}) {}
@override
bool onStreamListen({
required String streamId,
required Map<String, Object?> params,
}) {
return true;
}
}
@@ -24,7 +24,7 @@ Future<DartRuntimeService> createDartRuntimeServiceForTest({
service = await DartRuntimeService.initialize(
config: config,
backend: FakeDartRuntimeServiceBackend(),
backendBuilder: (_) => FakeDartRuntimeServiceBackend(),
);
return service;
}
@@ -4,9 +4,11 @@
import 'dart:async';
import 'dart:io';
import 'dart:isolate';
import 'package:dart_runtime_service/dart_runtime_service.dart';
import 'package:dart_runtime_service_vm/dart_runtime_service_vm.dart';
import 'package:dart_runtime_service_vm/src/vm_isolate_manager.dart';
// ignore: unreachable_from_main
const entrypoint = pragma(
@@ -63,6 +65,19 @@ bool _isFuchsia = false;
Stream<ProcessSignal> Function(ProcessSignal signal)? _signalWatch;
@entrypoint
RawReceivePort boot() => DartRuntimeServiceVMBackend.isolateControlPort;
final _isolateRegistrationStreamController = StreamController<VmRunningIsolate>(
sync: true,
);
@entrypoint
// ignore: unused_element
void _registerIsolate(int portId, SendPort sendPort, String name) =>
_isolateRegistrationStreamController.sink.add(
VmRunningIsolate(id: portId, name: name, sendPort: sendPort),
);
// ignore: unused_element
StreamSubscription<ProcessSignal>? _signalSubscription;
@@ -108,6 +123,10 @@ Future<void> main([List<String> args = const []]) async {
disableAuthCodes: _authCodesDisabled,
autoStart: _autoStart,
),
backend: DartRuntimeServiceVMBackend(signalWatch: _signalWatch!),
backendBuilder: (frontend) => DartRuntimeServiceVMBackend(
frontend: frontend,
signalWatch: _signalWatch!,
runningIsolatesStream: _isolateRegistrationStreamController.stream,
),
);
}
@@ -3,36 +3,99 @@
// BSD-style license that can be found in the LICENSE file.
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'dart:isolate';
import 'package:dart_runtime_service/dart_runtime_service.dart';
import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
import 'package:logging/logging.dart';
import 'package:stream_channel/stream_channel.dart';
import 'src/native_bindings.dart';
import 'src/vm_isolate_manager.dart';
class DartRuntimeServiceVMBackend extends DartRuntimeServiceBackend {
class DartRuntimeServiceVMBackend
extends DartRuntimeServiceBackend<VmIsolateManager> {
/// The backend implementation for the Dart VM Service.
///
/// [signalWatch] is the internal implementation of [ProcessSignal.watch],
/// which bypasses checks that prevent [ProcessSignal.sigquit] from being
/// watched.
DartRuntimeServiceVMBackend({required this.signalWatch});
DartRuntimeServiceVMBackend({
required super.frontend,
required this.signalWatch,
required Stream<VmRunningIsolate> runningIsolatesStream,
}) : isolateManager = VmIsolateManager(
runningIsolatesStream: runningIsolatesStream,
);
static const int _kServiceExitMessageId = 0;
static const int _kIsolateStartupMessageId = 1;
static const int _kIsolateShutdownMessageId = 2;
static const int _kWebServerControlMessageId = 3;
static const int _kServerInfoMessageId = 4;
/// Signals an RPC coming from native code (instead of from a websocket
/// connection). These calls are limited to simple request-response and do
/// not allow arbitrary json-rpc messages.
///
/// The messages are an array of length 3:
/// (kMethodCallFromNative, String jsonRequest, PortId replyPort).
static const int _kMethodCallFromNativeId = 5;
/// The internal implementation of [ProcessSignal.watch].
final Stream<ProcessSignal> Function(ProcessSignal signal) signalWatch;
StreamSubscription<ProcessSignal>? _sigquitSubscription;
/// The port the VM uses to send messages to the VM service.
static final isolateControlPort = RawReceivePort();
/// Used to create an artificial client from within the VM service, allowing
/// for VM service requests to be made from within the VM or through the Dart
/// embedding API.
final _nativeRpcClientStreamChannelController =
StreamChannelController<String>(sync: true);
/// Iterator for RPC responses sent to the artificial client.
late final _nativeRpcClientResponseStream = StreamIterator(
_nativeRpcClientStreamChannelController.local.stream,
);
final _nativeBindings = NativeBindings();
final _logger = Logger('VM Backend');
final _logger = Logger('$DartRuntimeServiceVMBackend');
StreamSubscription<ProcessSignal>? _sigquitSubscription;
@override
final VmIsolateManager isolateManager;
@override
Future<void> initialize() async {
_logger.info('Initializing...');
isolateControlPort.handler = _vmMessageHandler;
frontend.addArtificialClient(
name: 'native-rpc-client',
connection: _nativeRpcClientStreamChannelController.foreign,
);
_nativeBindings.onStart();
_logger.info('Initialized!');
_logger.info('Initialized.');
}
@override
Future<void> clearState() async {
// Do nothing for now.
}
@override
Future<void> shutdown() async {
_logger.info('Shutting down...');
await Future.wait([
_sigquitSubscription?.cancel() ?? Future<void>.value(),
_nativeRpcClientStreamChannelController.local.sink.close(),
]);
isolateControlPort.close();
_nativeBindings.onExit();
_logger.info('Shutdown.');
}
@override
@@ -53,19 +116,28 @@ class DartRuntimeServiceVMBackend extends DartRuntimeServiceBackend {
required Uri wsUri,
}) async {
// TODO(bkonyi): handle DDS connection case.
stdout.writeln('The Dart VM service is listening on $httpUri');
stdout.writeln('The Dart VM service is listening on $httpUri/');
_nativeBindings.onServerAddressChange(httpUri.toString());
}
@override
Future<void> clearState() async {
// Do nothing for now.
bool onStreamListen({
required String streamId,
required Map<String, Object?> params,
}) {
var includePrivates = false;
if (params case {'includePrivates': final bool value}) {
includePrivates = value;
}
return _nativeBindings.streamListen(
streamId: streamId,
includePrivates: includePrivates,
);
}
@override
Future<void> shutdown() async {
await _sigquitSubscription?.cancel();
_nativeBindings.onExit();
void onStreamCancel({required String streamId}) {
_nativeBindings.streamCancel(streamId: streamId);
}
@override
@@ -85,9 +157,168 @@ class DartRuntimeServiceVMBackend extends DartRuntimeServiceBackend {
final method = request.method;
final params = request.asMap.cast<String, Object?>();
if (params case {'isolateId': final String _}) {
// TODO(bkonyi): handle isolate requests
RpcException.serverError.throwException();
_logger.info(
'Sending request to isolate. Method: $method Params: $params',
);
return await isolateManager.sendToIsolate(method: method, params: params);
}
_logger.info('Sending request to VM. Method: $method Params: $params');
return await _nativeBindings.sendToVM(method: method, params: params);
}
/// Handles messages sent directly from the VM via the isolate control port.
///
/// Messages sent over the isolate control port include:
/// - Stream events
/// - Request to shutdown the service
/// - RPC invocations from the VM or VM's embedder
/// - dart:developer API invocations (e.g. Service.getInfo() and
/// Service.toggleWebServer())
/// - Isolate startup and shutdown notifications
void _vmMessageHandler(List<Object?> message) {
_logger.info('VM message: $message');
switch (message) {
case [final String streamId, final String eventJsonString]:
// This is an event.
_eventMessageHandler(
streamId,
json.decode(eventJsonString) as Map<String, Object?>,
);
case [final int opcode]:
// This is a control message directing the vm service to exit.
assert(opcode == _kServiceExitMessageId);
frontend.shutdown();
case [
final int opcode,
final List<int> messageBytes,
final SendPort replyPort,
]
when opcode == _kMethodCallFromNativeId:
_handleNativeRpcCall(messageBytes, replyPort);
case [
final int opcode,
final SendPort sendPort,
final bool enable,
final bool? silenceOutput,
]
when opcode == _kWebServerControlMessageId ||
opcode == _kServerInfoMessageId:
// This is a message interacting with the web server.
_serverMessageHandler(opcode, sendPort, enable, silenceOutput);
case [
final int opcode,
final int portId,
final SendPort sendPort,
final String name,
]
when opcode == _kIsolateStartupMessageId ||
opcode == _kIsolateShutdownMessageId:
// This is a message informing us of the birth or death of an
// isolate.
_isolateControlMessageHandler(opcode, portId, sendPort, name);
default:
print('Internal vm-service error: ignoring illegal message: $message');
}
}
/// Forward VM service events sent from the VM.
void _eventMessageHandler(String streamId, Map<String, Object?> event) {
frontend.sendEvent(
event: ForwardingStreamEvent(streamId: streamId, event: event),
);
}
/// Handle notifications from the VM related to isolate startup and shutdown.
void _isolateControlMessageHandler(
int code,
int portId,
SendPort sp,
String name,
) {
switch (code) {
case _kIsolateStartupMessageId:
isolateManager.onIsolateStartupMessage(
id: portId,
sendPort: sp,
name: name,
);
case _kIsolateShutdownMessageId:
isolateManager.onIsolateShutdownMessage(id: portId);
}
}
/// Handle requests from the VM related to the state of the service's HTTP
/// server.
Future<void> _serverMessageHandler(
int code,
SendPort sp,
bool enable,
bool? silenceOutput,
) async {
void sendServerInfo() {
try {
sp.send(frontend.httpUri.toString());
} on DartRuntimeServiceServerNotRunning {
sp.send(null);
}
}
switch (code) {
case _kWebServerControlMessageId:
await frontend.serverControl(
enable: enable,
silenceOutput: silenceOutput,
);
sendServerInfo();
case _kServerInfoMessageId:
sendServerInfo();
}
}
/// Handle VM service requests from native code (see
/// `Dart_InvokeVMServiceMethod` in `dart_tools_api.h`).
Future<void> _handleNativeRpcCall(
List<int> message,
SendPort replyPort,
) async {
final messageStr = utf8.decode(message);
_logger.info('Native RPC request: $messageStr');
_nativeRpcClientStreamChannelController.local.sink.add(messageStr);
// TODO(bkonyi): handle non-string results
/*
late List<int> bytes;
switch (response.kind) {
case ResponsePayloadKind.String:
bytes = utf8.encode(response.payload as String);
bytes = bytes is Uint8List ? bytes : Uint8List.fromList(bytes);
case ResponsePayloadKind.Binary:
case ResponsePayloadKind.Utf8String:
bytes = response.payload as Uint8List;
}
*/
if (!await _nativeRpcClientResponseStream.moveNext()) {
_logger.warning('Native RPC client stream has closed.');
return;
}
_logger.info(
'Response received: ${_nativeRpcClientResponseStream.current}',
);
replyPort.send(utf8.encode(_nativeRpcClientResponseStream.current));
}
}
final class ForwardingStreamEvent extends StreamEvent {
ForwardingStreamEvent({required super.streamId, required this.event})
: super(kind: '<ignored>');
static const kParams = 'params';
Map<String, Object?> event;
@override
Map<String, Object?> toJson() {
return event[kParams] as Map<String, Object?>;
}
}
@@ -16,11 +16,20 @@ import 'dart:typed_data';
import 'package:dart_runtime_service/dart_runtime_service.dart';
import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
import 'package:logging/logging.dart';
import 'vm_isolate_manager.dart';
/// Allows for sending messages to the native VM service implementation.
class NativeBindings {
factory NativeBindings() => _instance;
NativeBindings._();
static final NativeBindings _instance = NativeBindings._();
static final jsonUtf8Decoder = json.fuse(utf8);
final _logger = Logger('$NativeBindings');
/// Sends a general RPC to the VM for processing.
///
/// The RPC is not executed in the scope of any particular isolate.
@@ -44,6 +53,66 @@ class NativeBindings {
return completer.future;
}
/// Sends an RPC to a specific isolate for processing.
///
/// The RPC is not executed in the scope of any particular isolate.
Future<RpcResponse> sendToIsolate({
required VmRunningIsolate isolate,
required String method,
required Map<String, Object?> params,
}) {
final receivePort = RawReceivePort(
null,
'Isolate Message (${isolate.name})',
);
// Keep track of receive port associated with the request so we can close
// it if isolate exits before sending a response.
isolate.outstandingRequestPorts.add(receivePort);
final completer = Completer<RpcResponse>();
receivePort.handler = (Object value) {
receivePort.close();
isolate.outstandingRequestPorts.remove(receivePort);
try {
completer.complete(_toResponse(value: value));
} on json_rpc.RpcException catch (e) {
completer.completeError(e);
}
};
if (!vm_service_natives.sendIsolateServiceMessage(
isolate.sendPort,
_toRequest(responsePort: receivePort, method: method, params: params),
)) {
receivePort.close();
isolate.outstandingRequestPorts.remove(receivePort);
_logger.warning('Could not send message to $isolate.');
completer.completeError(RpcException.internalError.toException());
}
return completer.future;
}
/// Notifies the VM to start sending events for [streamId].
///
/// This only needs to be called once the first client has subscribed to the
/// stream. Subsequent subscriptions to the stream are handled by the
/// [EventStreamManager].
///
/// If [includePrivates] is true, private event properties starting with '_'
/// will be included in events. This is false by default to reduce the size of
/// events for clients that don't rely on private properties.
bool streamListen({required String streamId, bool includePrivates = false}) =>
// TODO(bkonyi): handle case where some clients want privates included
// and others don't.
vm_service_natives.vmListenStream(streamId, includePrivates);
/// Notifies the VM to stop sending events for [streamId].
///
/// This only needs to be called once the last client subscribed to the
/// stream has cancelled its subscription or disconnected. While clients have
/// active subscriptions to this stream, stream cancellation requests are
/// handled by the [EventStreamManager].
void streamCancel({required String streamId}) =>
vm_service_natives.vmCancelStream(streamId);
/// Notifies the VM that the VM service server has finished initializing.
void onStart() => vm_service_natives.onStart();
@@ -0,0 +1,102 @@
// Copyright (c) 2026, 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:isolate';
import 'package:dart_runtime_service/dart_runtime_service.dart';
import 'package:logging/logging.dart';
import 'package:vm_service/vm_service.dart';
import 'native_bindings.dart';
/// A running isolate for the Dart VM.
final class VmRunningIsolate extends RunningIsolate {
VmRunningIsolate({
required super.id,
required super.name,
required this.sendPort,
});
/// The port used to send service requests to the isolate within the VM.
final SendPort sendPort;
/// The set of ports for outstanding requests that are used by the VM to send
/// responses.
final outstandingRequestPorts = <RawReceivePort>{};
@override
void shutdown() {
for (final requestPort in outstandingRequestPorts) {
requestPort.close();
}
outstandingRequestPorts.clear();
super.shutdown();
}
}
/// Manages and tracks running isolates in the Dart VM.
final class VmIsolateManager extends IsolateManager {
/// Initializes the [VmIsolateManager].
///
/// [runningIsolatesStream] should be a stream of [VmRunningIsolate]s reported
/// by the Dart VM as started once the VM service has finished initializing.
VmIsolateManager({required Stream<VmRunningIsolate> runningIsolatesStream}) {
_runningIsolatesStreamSub = runningIsolatesStream.listen(
(isolate) => isolateStarted(isolate: isolate),
);
}
final _logger = Logger('$VmIsolateManager');
final _nativeBindings = NativeBindings();
late final StreamSubscription<VmRunningIsolate> _runningIsolatesStreamSub;
@override
Future<void> shutdown() async {
await _runningIsolatesStreamSub.cancel();
await super.shutdown();
}
/// Registers a newly started isolate reported via a message over the
/// service's control port.
void onIsolateStartupMessage({
required int id,
required SendPort sendPort,
required String name,
}) {
final isolate = VmRunningIsolate(id: id, name: name, sendPort: sendPort);
_logger.info('Isolate startup message received for $isolate');
isolateStarted(isolate: isolate);
}
/// Reports that an isolate is shutting down based on a message over the
/// service's control port.
void onIsolateShutdownMessage({required int id}) {
_logger.info('Isolate startup message received for isolate ID $id');
isolateExited(id: id);
}
@override
Future<RpcResponse> sendToIsolate({
required String method,
required Map<String, Object?> params,
}) async {
final isolate =
lookupIsolateFromParams(method: method, params: params)
as VmRunningIsolate?;
if (isolate == null) {
// There is some chance that this isolate may have lived before,
// so return a sentinel rather than an error.
return Sentinel(
kind: SentinelKind.kCollected,
valueAsString: '<collected>',
).toJson();
}
return _nativeBindings.sendToIsolate(
isolate: isolate,
method: method,
params: params,
);
}
}
+3 -1
View File
@@ -3,7 +3,7 @@ name: dart_runtime_service_vm
publish_to: none
environment:
sdk: ^3.8.0
sdk: ^3.12.0-edge
resolution: workspace
@@ -12,3 +12,5 @@ dependencies:
dart_runtime_service: any
json_rpc_2: any
logging: any
stream_channel: any
vm_service: any
+50 -15
View File
@@ -110,8 +110,14 @@ static Dart_Isolate CreateAndSetupServiceIsolate(const char* script_uri,
// vm/cc tests to randomly time out due to inability to shut service-isolate
// down.
// Issue(https://dartbug.com/37741):
if ((strcmp(run_filter, "DartAPI_InvokeVMServiceMethod") != 0) &&
(strcmp(run_filter, "DartAPI_InvokeVMServiceMethod_Loop") != 0)) {
const bool is_service_test =
(strcmp(run_filter, "DartAPI_InvokeVMServiceMethod") == 0) ||
(strcmp(run_filter, "DartAPI_InvokeVMServiceMethod_Loop") == 0);
const bool is_exp_service_test =
(strcmp(run_filter, "DartAPI_InvokeVMServiceMethod_Exp") == 0) ||
(strcmp(run_filter, "DartAPI_InvokeVMServiceMethod_Loop_Exp") == 0);
if (!is_service_test && !is_exp_service_test) {
return nullptr;
}
@@ -122,21 +128,50 @@ static Dart_Isolate CreateAndSetupServiceIsolate(const char* script_uri,
packages_config, /*app_snapshot=*/nullptr,
/*isolate_run_app_snapshot=*/false);
const uint8_t* kernel_buffer = nullptr;
intptr_t kernel_buffer_size = 0;
bin::dfe.Init();
bin::dfe.LoadPlatform(&kernel_buffer, &kernel_buffer_size);
RELEASE_ASSERT(kernel_buffer != nullptr);
flags->load_vmservice_library = true;
flags->is_service_isolate = true;
isolate_group_data->SetKernelBufferUnowned(
const_cast<uint8_t*>(kernel_buffer), kernel_buffer_size);
isolate = Dart_CreateIsolateGroupFromKernel(
script_uri, DART_VM_SERVICE_ISOLATE_NAME, kernel_buffer,
kernel_buffer_size, flags, isolate_group_data, /*isolate_data=*/nullptr,
error);
#if defined(EXPERIMENTAL_VM_SERVICE)
if (is_exp_service_test) {
ASSERT(!is_service_test);
const uint8_t* isolate_snapshot_data = nullptr;
const uint8_t* isolate_snapshot_instructions = nullptr;
bin::VmService::enable_experimental_vm_service = true;
auto [app_snapshot, script_name] = bin::Snapshot::TryReadSDKSnapshot(
"dart_runtime_service_vm.dart.snapshot");
if (app_snapshot == nullptr) {
return nullptr;
}
const uint8_t* ignore_vm_snapshot_data;
const uint8_t* ignore_vm_snapshot_instructions;
app_snapshot->SetBuffers(
&ignore_vm_snapshot_data, &ignore_vm_snapshot_instructions,
&isolate_snapshot_data, &isolate_snapshot_instructions);
isolate = Dart_CreateIsolateGroup(
script_uri, DART_VM_SERVICE_ISOLATE_NAME, isolate_snapshot_data,
isolate_snapshot_instructions, flags, isolate_group_data,
/*isolate_data=*/nullptr, error);
}
#endif // defined(EXPERIMENTAL_VM_SERVICE)
if (is_service_test) {
ASSERT(!is_exp_service_test);
const uint8_t* kernel_buffer = nullptr;
intptr_t kernel_buffer_size = 0;
bin::dfe.Init();
bin::dfe.LoadPlatform(&kernel_buffer, &kernel_buffer_size);
RELEASE_ASSERT(kernel_buffer != nullptr);
isolate_group_data->SetKernelBufferUnowned(
const_cast<uint8_t*>(kernel_buffer), kernel_buffer_size);
isolate = Dart_CreateIsolateGroupFromKernel(
script_uri, DART_VM_SERVICE_ISOLATE_NAME, kernel_buffer,
kernel_buffer_size, flags, isolate_group_data, /*isolate_data=*/nullptr,
error);
}
if (isolate == nullptr) {
delete isolate_group_data;
return nullptr;
+24 -4
View File
@@ -10559,11 +10559,11 @@ TEST_CASE(DartAPI_InvokeImportedFunction) {
"NoSuchMethodError: No top-level method 'getCurrentTag' declared.");
}
TEST_CASE(DartAPI_InvokeVMServiceMethod) {
static void InvokeVMServiceMethodCommon() {
char buffer[1024];
Utils::SNPrint(buffer, sizeof(buffer),
R"({
"jsonrpc": 2.0,
"jsonrpc": "2.0",
"id": "foo",
"method": "getVM",
"params": { }
@@ -10618,6 +10618,16 @@ TEST_CASE(DartAPI_InvokeVMServiceMethod) {
EXPECT(result == Dart_True());
}
TEST_CASE(DartAPI_InvokeVMServiceMethod) {
InvokeVMServiceMethodCommon();
}
#if defined(EXPERIMENTAL_VM_SERVICE)
TEST_CASE(DartAPI_InvokeVMServiceMethod_Exp) {
InvokeVMServiceMethodCommon();
}
#endif // defined(EXPERIMENTAL_VM_SERVICE)
static Monitor* loop_test_lock = new Monitor();
static bool loop_test_exit = false;
static bool loop_reset_count = false;
@@ -10630,7 +10640,7 @@ static void InvokeServiceMessages(uword param) {
char buffer[1024];
Utils::SNPrint(buffer, sizeof(buffer),
R"({
"jsonrpc": 2.0,
"jsonrpc": "2.0",
"id": "foo",
"method": "getVM",
"params": { }
@@ -10661,7 +10671,7 @@ static void InvokeServiceMessages(uword param) {
} while (count < 100);
}
TEST_CASE(DartAPI_InvokeVMServiceMethod_Loop) {
static void InvokeVMServiceMethodLoopCommon() {
{
MonitorLocker ml(loop_test_lock);
loop_test_exit = false;
@@ -10673,6 +10683,16 @@ TEST_CASE(DartAPI_InvokeVMServiceMethod_Loop) {
}
OSThread::Join(loop_test_join_id);
}
TEST_CASE(DartAPI_InvokeVMServiceMethod_Loop) {
InvokeVMServiceMethodLoopCommon();
}
#if defined(EXPERIMENTAL_VM_SERVICE)
TEST_CASE(DartAPI_InvokeVMServiceMethod_Loop_Exp) {
InvokeVMServiceMethodLoopCommon();
}
#endif // defined(EXPERIMENTAL_VM_SERVICE)
#endif // !defined(PRODUCT)
static void HandleResponse(Dart_Port dest_port_id, Dart_CObject* message) {
+23 -15
View File
@@ -578,12 +578,18 @@ void ServiceIsolate::Shutdown() {
void ServiceIsolate::BootVmServiceLibrary() {
Thread* thread = Thread::Current();
const Library& vmservice_library =
Library::Handle(Library::LookupLibrary(thread, Symbols::DartVMService()));
ASSERT(!vmservice_library.IsNull());
Library& lib =
Library::Handle(thread->isolate_group()->object_store()->root_library());
const String& boot_function_name = String::Handle(String::New("boot"));
const Function& boot_function = Function::Handle(
vmservice_library.LookupFunctionAllowPrivate(boot_function_name));
Function& boot_function =
Function::Handle(lib.LookupFunctionAllowPrivate(boot_function_name));
if (boot_function.IsNull()) {
lib ^= Library::LookupLibrary(thread, Symbols::DartVMService());
ASSERT(!lib.IsNull());
boot_function ^= lib.LookupFunctionAllowPrivate(boot_function_name);
}
ASSERT(!boot_function.IsNull());
const Object& result = Object::Handle(
DartEntry::InvokeFunction(boot_function, Object::empty_array()));
@@ -608,16 +614,18 @@ void ServiceIsolate::RegisterRunningIsolates(
ASSERT(thread->isolate()->is_service_isolate());
// Obtain "_registerIsolate" function to call.
const String& library_url = Symbols::DartVMService();
ASSERT(!library_url.IsNull());
const Library& library =
Library::Handle(zone, Library::LookupLibrary(thread, library_url));
ASSERT(!library.IsNull());
const String& function_name =
String::Handle(zone, String::New("_registerIsolate"));
ASSERT(!function_name.IsNull());
const Function& register_function_ =
Function::Handle(zone, library.LookupFunctionAllowPrivate(function_name));
Library& lib =
Library::Handle(thread->isolate_group()->object_store()->root_library());
const String& function_name = String::Handle(String::New("_registerIsolate"));
Function& register_function_ =
Function::Handle(lib.LookupFunctionAllowPrivate(function_name));
if (register_function_.IsNull()) {
lib ^= Library::LookupLibrary(thread, Symbols::DartVMService());
ASSERT(!lib.IsNull());
register_function_ ^= lib.LookupFunctionAllowPrivate(function_name);
}
ASSERT(!register_function_.IsNull());
Integer& port_int = Integer::Handle(zone);