[ VM / Service ] Add support for streaming CPU samples with specific

user tags

TEST=cpu_sample_streaming_test.dart

Change-Id: Ia983217ae2a5da8c3252fafbed8197b4f4a20e2b
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/224040
Reviewed-by: Siva Annamalai <asiva@google.com>
Commit-Queue: Ben Konyi <bkonyi@google.com>

pkg/dds/lib/src/client.dart

@@ -107,6 +107,16 @@ class DartDevelopmentServiceClient {
       return RPCResponses.success;
     });
 
+    _clientPeer.registerMethod('streamCpuSamplesWithUserTag',
+        (parameters) async {
+      final userTags = parameters['userTags'].asList.cast<String>();
+      profilerUserTagFilters.clear();
+      profilerUserTagFilters.addAll(userTags);
+
+      await dds.streamManager.updateUserTagSubscriptions(userTags);
+      return RPCResponses.success;
+    });
+
     _clientPeer.registerMethod('registerService', (parameters) async {
       final serviceId = parameters['service'].asString;
       final alias = parameters['alias'].asString;
@@ -312,6 +322,7 @@ class DartDevelopmentServiceClient {
   final DartDevelopmentServiceImpl dds;
   final StreamChannel connection;
   final Map<String, String> services = {};
+  final Set<String> profilerUserTagFilters = {};
   final json_rpc.Peer _vmServicePeer;
   late json_rpc.Peer _clientPeer;
 }

pkg/dds/lib/src/cpu_samples_manager.dart

@@ -15,34 +15,15 @@ class CpuSamplesManager {
     }
   }
 
-  void handleUserTagEvent(Event event) {
-    assert(event.kind! == EventKind.kUserTagChanged);
-    _currentTag = event.updatedTag!;
-    final previousTag = event.previousTag!;
-    if (cpuSamplesCaches.containsKey(previousTag)) {
-      _lastCachedTag = previousTag;
-    }
-  }
-
   void handleCpuSamplesEvent(Event event) {
-    assert(event.kind! == EventKind.kCpuSamples);
-    // There might be some samples left in the buffer for the previously set
-    // user tag. We'll check for them here and then close out the cache.
-    if (_lastCachedTag != null) {
-      cpuSamplesCaches[_lastCachedTag]!.cacheSamples(
-        event.cpuSamples!,
-      );
-      _lastCachedTag = null;
+    for (final userTag in dds.cachedUserTags) {
+      cpuSamplesCaches[userTag]!.cacheSamples(event.cpuSamples!);
     }
-    cpuSamplesCaches[_currentTag]?.cacheSamples(event.cpuSamples!);
   }
 
   final DartDevelopmentServiceImpl dds;
   final String isolateId;
   final cpuSamplesCaches = <String, CpuSamplesRepository>{};
-
-  String _currentTag = '';
-  String? _lastCachedTag;
 }
 
 class CpuSamplesRepository extends RingBuffer<CpuSample> {

pkg/dds/lib/src/isolate_manager.dart

@@ -123,9 +123,6 @@ class _RunningIsolate {
 
   void handleEvent(Event event) {
     switch (event.kind) {
-      case EventKind.kUserTagChanged:
-        cpuSamplesManager.handleUserTagEvent(event);
-        return;
       case EventKind.kCpuSamples:
         cpuSamplesManager.handleCpuSamplesEvent(event);
         return;
@@ -239,6 +236,14 @@ class IsolateManager {
             isolateStarted(id, name);
           }
         }
+        if (dds.cachedUserTags.isNotEmpty) {
+          await dds.vmServiceClient.sendRequest(
+            'streamCpuSamplesWithUserTag',
+            {
+              'userTags': dds.cachedUserTags,
+            },
+          );
+        }
       },
     );
     _initialized = true;

pkg/dds/lib/src/stream_manager.dart

@@ -39,12 +39,36 @@ class StreamManager {
         if (isBinaryData) {
           listener.connection.sink.add(data);
         } else {
-          listener.sendNotification('streamNotify', data);
+          Map<String, dynamic> processed = data;
+          if (streamId == kProfilerStream) {
+            processed = _processProfilerEvents(listener, data);
+          }
+          listener.sendNotification('streamNotify', processed);
         }
       }
     }
   }
 
+  static Map<String, dynamic> _processProfilerEvents(
+    DartDevelopmentServiceClient client,
+    Map<String, dynamic> data,
+  ) {
+    final event = Event.parse(data['event'])!;
+    if (event.kind != EventKind.kCpuSamples) {
+      return data;
+    }
+    final cpuSamplesEvent = event.cpuSamples!;
+    cpuSamplesEvent.samples = cpuSamplesEvent.samples!
+        .where(
+          (e) => client.profilerUserTagFilters.contains(e.userTag),
+        )
+        .toList();
+    cpuSamplesEvent.sampleCount = cpuSamplesEvent.samples!.length;
+    final updated = Map<String, dynamic>.from(data);
+    updated['event']['cpuSamples'] = cpuSamplesEvent.toJson();
+    return updated;
+  }
+
   static Map<String, dynamic> _buildStreamRegisteredEvent(
           String namespace, String service, String alias) =>
       {
@@ -148,7 +172,7 @@ class StreamManager {
     DartDevelopmentServiceClient? client,
     String stream,
   ) async {
-    await _mutex.runGuarded(
+    await _streamSubscriptionMutex.runGuarded(
       () async {
         assert(stream.isNotEmpty);
         if (!streamListeners.containsKey(stream)) {
@@ -218,7 +242,7 @@ class StreamManager {
     String stream, {
     bool cancelCoreStream = false,
   }) async {
-    await _mutex.runGuarded(
+    await _streamSubscriptionMutex.runGuarded(
       () async {
         assert(stream.isNotEmpty);
         final listeners = streamListeners[stream];
@@ -246,6 +270,28 @@ class StreamManager {
     );
   }
 
+  Future<void> updateUserTagSubscriptions(
+      [List<String> userTags = const []]) async {
+    await _profilerUserTagSubscriptionsMutex.runGuarded(() async {
+      _profilerUserTagSubscriptions.addAll(userTags);
+      for (final subscribedTag in _profilerUserTagSubscriptions.toList()) {
+        bool hasSubscriber = false;
+        for (final c in dds.clientManager.clients) {
+          if (c.profilerUserTagFilters.contains(subscribedTag)) {
+            hasSubscriber = true;
+            break;
+          }
+        }
+        if (!hasSubscriber) {
+          _profilerUserTagSubscriptions.remove(subscribedTag);
+        }
+      }
+      await dds.vmServiceClient.sendRequest('streamCpuSamplesWithUserTag', {
+        'userTags': _profilerUserTagSubscriptions.toList(),
+      });
+    });
+  }
+
   /// Cleanup stream subscriptions for `client` when it has disconnected.
   void clientDisconnect(DartDevelopmentServiceClient client) {
     for (final streamId in streamListeners.keys.toList()) {
@@ -256,6 +302,11 @@ class StreamManager {
         test: (e) => (e is json_rpc.RpcException) || (e is StateError),
       );
     }
+    updateUserTagSubscriptions().catchError(
+      (_) => null,
+      test: (e) => (e is json_rpc.RpcException) || (e is StateError),
+    );
+
     // Notify other service clients of service extensions that are being
     // unregistered.
     _sendServiceUnregisteredEvents(client);
@@ -312,5 +363,7 @@ class StreamManager {
 
   final DartDevelopmentServiceImpl dds;
   final streamListeners = <String, List<DartDevelopmentServiceClient>>{};
-  final _mutex = Mutex();
+  final _profilerUserTagSubscriptions = <String>{};
+  final _streamSubscriptionMutex = Mutex();
+  final _profilerUserTagSubscriptionsMutex = Mutex();
 }

pkg/dds/test/common/test_helper.dart

@@ -13,6 +13,7 @@ late Uri remoteVmServiceUri;
 Future<Process> spawnDartProcess(
   String script, {
   bool pauseOnStart = true,
+  bool disableServiceAuthCodes = false,
 }) async {
   final executable = Platform.executable;
   final tmpDir = await Directory.systemTemp.createTemp('dart_service');
@@ -23,6 +24,7 @@ Future<Process> spawnDartProcess(
     '--disable-dart-dev',
     '--observe=0',
     if (pauseOnStart) '--pause-isolates-on-start',
+    if (disableServiceAuthCodes) '--disable-service-auth-codes',
     '--write-service-info=$serviceInfoUri',
     ...Platform.executableArguments,
     Platform.script.resolve(script).toString(),

pkg/dds/test/cpu_sample_streaming_test.dart

@@ -0,0 +1,135 @@
+// Copyright (c) 2021, 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 'package:dds/dds.dart';
+import 'package:test/test.dart';
+import 'package:vm_service/vm_service.dart';
+import 'package:vm_service/vm_service_io.dart';
+import 'common/test_helper.dart';
+
+void main() {
+  late Process process;
+  DartDevelopmentService? dds;
+
+  setUp(() async {
+    process = await spawnDartProcess(
+      'get_cached_cpu_samples_script.dart',
+      disableServiceAuthCodes: true,
+    );
+  });
+
+  tearDown(() async {
+    await dds?.shutdown();
+    process.kill();
+  });
+
+  void declareTest(bool withDds) {
+    test(
+      'Stream CPU samples for provided UserTag names with${withDds ? "" : "out"} DDS',
+      () async {
+        Uri serviceUri = remoteVmServiceUri;
+        if (withDds) {
+          dds = await DartDevelopmentService.startDartDevelopmentService(
+            remoteVmServiceUri,
+          );
+          serviceUri = dds!.wsUri!;
+          expect(dds!.isRunning, true);
+        } else {
+          serviceUri = serviceUri.replace(scheme: 'ws', path: 'ws');
+        }
+        final service = await vmServiceConnectUri(serviceUri.toString());
+        final otherService = await vmServiceConnectUri(serviceUri.toString());
+
+        IsolateRef isolate;
+        while (true) {
+          final vm = await service.getVM();
+          if (vm.isolates!.isNotEmpty) {
+            isolate = vm.isolates!.first;
+            try {
+              isolate = await service.getIsolate(isolate.id!);
+              if ((isolate as Isolate).runnable!) {
+                break;
+              }
+            } on SentinelException {
+              // ignore
+            }
+          }
+          await Future.delayed(const Duration(seconds: 1));
+        }
+        expect(isolate, isNotNull);
+
+        final expectedUserTags = <String>{};
+
+        Future<void> listenForSamples() {
+          late StreamSubscription sub;
+          final completer = Completer<void>();
+          int i = 0;
+          sub = service.onProfilerEvent.listen(
+            (event) async {
+              if (event.kind == EventKind.kCpuSamples &&
+                  event.isolate!.id! == isolate.id!) {
+                expect(expectedUserTags.isNotEmpty, true);
+                ++i;
+                if (i > 3) {
+                  if (!completer.isCompleted) {
+                    await sub.cancel();
+                    completer.complete();
+                  }
+                  return;
+                }
+                expect(event.cpuSamples, isNotNull);
+                final sampleCount = event.cpuSamples!.samples!
+                    .where((e) => expectedUserTags.contains(e.userTag))
+                    .length;
+                expect(sampleCount, event.cpuSamples!.samples!.length);
+              }
+            },
+          );
+          if (expectedUserTags.isEmpty) {
+            return Future.delayed(const Duration(seconds: 2)).then(
+              (_) async => await sub.cancel(),
+            );
+          }
+          return completer.future;
+        }
+
+        await service.streamListen(EventStreams.kProfiler);
+        Future<void> subscription = listenForSamples();
+        await service.resume(isolate.id!);
+        await subscription;
+        await service.pause(isolate.id!);
+
+        expectedUserTags.add('Testing');
+        await service.streamCpuSamplesWithUserTag(expectedUserTags.toList());
+        subscription = listenForSamples();
+        await service.resume(isolate.id!);
+        await subscription;
+        await service.pause(isolate.id!);
+
+        expectedUserTags.add('Baz');
+        await service.streamCpuSamplesWithUserTag(expectedUserTags.toList());
+        subscription = listenForSamples();
+        await service.resume(isolate.id!);
+        await subscription;
+        await service.pause(isolate.id!);
+
+        expectedUserTags.clear();
+        await service.streamCpuSamplesWithUserTag(expectedUserTags.toList());
+        subscription = listenForSamples();
+        await service.resume(isolate.id!);
+        await subscription;
+
+        await service.dispose();
+        await otherService.dispose();
+      },
+      timeout: Timeout.none,
+    );
+  }
+
+  declareTest(true);
+  declareTest(false);
+}

pkg/dds/test/get_cached_cpu_samples_script.dart

@@ -12,10 +12,14 @@ fib(int n) {
 }
 
 void main() {
-  UserTag('Testing').makeCurrent();
+  final tag = UserTag('Testing')..makeCurrent();
+  final tag2 = UserTag('Baz');
   int i = 5;
   while (true) {
+    tag.makeCurrent();
     ++i;
     fib(i);
+    tag2.makeCurrent();
+    fib(i);
   }
 }

pkg/dds/test/get_cached_cpu_samples_test.dart

@@ -74,6 +74,7 @@ void main() {
       );
       expect(dds.isRunning, true);
       final service = await vmServiceConnectUri(dds.wsUri.toString());
+      final otherService = await vmServiceConnectUri(dds.wsUri.toString());
 
       // Ensure we're caching results for samples under the 'Testing' UserTag.
       final availableCaches = await service.getAvailableCachedCpuSamples();
@@ -85,13 +86,18 @@ void main() {
         final vm = await service.getVM();
         if (vm.isolates!.isNotEmpty) {
           isolate = vm.isolates!.first;
-          isolate = await service.getIsolate(isolate.id!);
-          if ((isolate as Isolate).runnable!) {
-            break;
+          try {
+            isolate = await service.getIsolate(isolate.id!);
+            if ((isolate as Isolate).runnable!) {
+              break;
+            }
+          } on SentinelException {
+            // ignore
           }
         }
         await Future.delayed(const Duration(seconds: 1));
       }
+      expect(isolate, isNotNull);
 
       final completer = Completer<void>();
       int i = 0;
@@ -117,9 +123,11 @@ void main() {
                 // Ensure the number of CPU samples in the CpuSample event is
                 // is consistent with the number of samples in the cache.
                 expect(event.cpuSamples, isNotNull);
-                count += event.cpuSamples!.samples!
+                final sampleCount = event.cpuSamples!.samples!
                     .where((e) => e.userTag == kUserTag)
                     .length;
+                expect(sampleCount, event.cpuSamples!.samples!.length);
+                count += sampleCount;
                 final cache = await service.getCachedCpuSamples(
                   isolate.id!,
                   availableCaches.cacheNames.first,
@@ -134,6 +142,11 @@ void main() {
         },
       );
       await service.streamListen(EventStreams.kProfiler);
+      await service.streamCpuSamplesWithUserTag(['Testing']);
+      // Have another client register for samples from another UserTag. The
+      // main client should not see any samples with the 'Baz' tag.
+      await otherService.streamListen(EventStreams.kProfiler);
+      await otherService.streamCpuSamplesWithUserTag(['Testing', 'Baz']);
       await service.resume(isolate.id!);
       await completer.future;
     },

pkg/vm/testcases/transformations/type_flow/transformer/enum_from_lib_used_as_type.dart.expect

@@ -22,6 +22,6 @@ class Class extends core::Object {
   synthetic constructor •() → self::Class
     : super core::Object::•()
     ;
-[@vm.procedure-attributes.metadata=methodOrSetterCalledDynamically:false,getterCalledDynamically:false,hasThisUses:false,hasTearOffUses:false,methodOrSetterSelectorId:3304,getterSelectorId:3305]  method method([@vm.inferred-type.metadata=dart.core::Null? (value: null)] self::Enum e) → core::int
+[@vm.procedure-attributes.metadata=methodOrSetterCalledDynamically:false,getterCalledDynamically:false,hasThisUses:false,hasTearOffUses:false,methodOrSetterSelectorId:3380,getterSelectorId:3381]  method method([@vm.inferred-type.metadata=dart.core::Null? (value: null)] self::Enum e) → core::int
     return [@vm.inferred-type.metadata=!] e.{core::_Enum::index}{core::int};
 }

pkg/vm/testcases/transformations/type_flow/transformer/tree_shake_enum_from_lib.dart.expect

@@ -51,6 +51,6 @@ class ConstClass extends core::Object {
   synthetic constructor •() → self::ConstClass
     : super core::Object::•()
     ;
-[@vm.procedure-attributes.metadata=methodOrSetterCalledDynamically:false,getterCalledDynamically:false,hasThisUses:false,hasTearOffUses:false,methodOrSetterSelectorId:3308,getterSelectorId:3309]  method method([@vm.inferred-type.metadata=dart.core::Null? (value: null)] self::ConstEnum e) → core::int
+[@vm.procedure-attributes.metadata=methodOrSetterCalledDynamically:false,getterCalledDynamically:false,hasThisUses:false,hasTearOffUses:false,methodOrSetterSelectorId:3384,getterSelectorId:3385]  method method([@vm.inferred-type.metadata=dart.core::Null? (value: null)] self::ConstEnum e) → core::int
     return [@vm.inferred-type.metadata=!] e.{core::_Enum::index}{core::int};
 }

pkg/vm_service/CHANGELOG.md

@@ -1,6 +1,11 @@
 # Changelog
 
+## 8.1.0
+- Update to version `3.55` of the spec.
+- Added `streamCpuSamplesWithUserTag` RPC.
+
 ## 8.0.0
+- Update to version `3.54` of the spec.
 - *breaking* Updated type of `Event.cpuSamples` from `CpuSamples` to
   `CpuSamplesEvent`, which is less expensive to generate and serialize.
 - Added `CpuSamplesEvent` object.

pkg/vm_service/java/version.properties

@@ -1 +1 @@
-version=3.54
+version=3.55

pkg/vm_service/lib/src/vm_service.dart

@@ -26,7 +26,7 @@ export 'snapshot_graph.dart'
         HeapSnapshotObjectNoData,
         HeapSnapshotObjectNullData;
 
-const String vmServiceVersion = '3.54.0';
+const String vmServiceVersion = '3.55.0';
 
 /// @optional
 const String optional = 'optional';
@@ -245,6 +245,7 @@ Map<String, List<String>> _methodReturnTypes = {
   'setVMName': const ['Success'],
   'setVMTimelineFlags': const ['Success'],
   'streamCancel': const ['Success'],
+  'streamCpuSamplesWithUserTag': const ['Success'],
   'streamListen': const ['Success'],
 };
 
@@ -1205,6 +1206,15 @@ abstract class VmServiceInterface {
   /// See [Success].
   Future<Success> streamCancel(String streamId);
 
+  /// The `streamCpuSamplesWithUserTag` RPC allows for clients to specify which
+  /// CPU samples collected by the profiler should be sent over the `Profiler`
+  /// stream. When called, the VM will stream `CpuSamples` events containing
+  /// `CpuSample`'s collected while a user tag contained in `userTags` was
+  /// active.
+  ///
+  /// See [Success].
+  Future<Success> streamCpuSamplesWithUserTag(List<String> userTags);
+
   /// The `streamListen` RPC subscribes to a stream in the VM. Once subscribed,
   /// the client will begin receiving events from the stream.
   ///
@@ -1640,6 +1650,11 @@ class VmServerConnection {
           await existing.cancel();
           response = Success();
           break;
+        case 'streamCpuSamplesWithUserTag':
+          response = await _serviceImplementation.streamCpuSamplesWithUserTag(
+            List<String>.from(params!['userTags'] ?? []),
+          );
+          break;
         case 'streamListen':
           var id = params!['streamId'];
           if (_streamSubscriptions.containsKey(id)) {
@@ -2168,6 +2183,10 @@ class VmService implements VmServiceInterface {
   Future<Success> streamCancel(String streamId) =>
       _call('streamCancel', {'streamId': streamId});
 
+  @override
+  Future<Success> streamCpuSamplesWithUserTag(List<String> userTags) =>
+      _call('streamCpuSamplesWithUserTag', {'userTags': userTags});
+
   @override
   Future<Success> streamListen(String streamId) =>
       _call('streamListen', {'streamId': streamId});

runtime/lib/vmservice.cc

@@ -384,4 +384,72 @@ DEFINE_NATIVE_ENTRY(VMService_DecodeAssets, 0, 1) {
 #endif
 }
 
+#ifndef PRODUCT
+class UserTagIsolatesVisitor : public IsolateVisitor {
+ public:
+  UserTagIsolatesVisitor(Thread* thread,
+                         const GrowableObjectArray* user_tags,
+                         bool set_streamable)
+      : IsolateVisitor(),
+        thread_(thread),
+        user_tags_(user_tags),
+        set_streamable_(set_streamable) {}
+
+  virtual void VisitIsolate(Isolate* isolate) {
+    if (Isolate::IsVMInternalIsolate(isolate)) {
+      return;
+    }
+    Zone* zone = thread_->zone();
+    UserTag& tag = UserTag::Handle(zone);
+    String& label = String::Handle(zone);
+    for (intptr_t i = 0; i < user_tags_->Length(); ++i) {
+      label ^= user_tags_->At(i);
+      tag ^= UserTag::FindTagInIsolate(isolate, thread_, label);
+      if (!tag.IsNull()) {
+        tag.set_streamable(set_streamable_);
+      }
+    }
+  }
+
+ private:
+  Thread* thread_;
+  const GrowableObjectArray* user_tags_;
+  bool set_streamable_;
+
+  DISALLOW_COPY_AND_ASSIGN(UserTagIsolatesVisitor);
+};
+#endif  // !PRODUCT
+
+DEFINE_NATIVE_ENTRY(VMService_AddUserTagsToStreamableSampleList, 0, 1) {
+#ifndef PRODUCT
+  GET_NON_NULL_NATIVE_ARGUMENT(GrowableObjectArray, user_tags,
+                               arguments->NativeArgAt(0));
+
+  Object& obj = Object::Handle();
+  for (intptr_t i = 0; i < user_tags.Length(); ++i) {
+    obj = user_tags.At(i);
+    UserTags::AddStreamableTagName(obj.ToCString());
+  }
+  UserTagIsolatesVisitor visitor(thread, &user_tags, true);
+  Isolate::VisitIsolates(&visitor);
+#endif
+  return Object::null();
+}
+
+DEFINE_NATIVE_ENTRY(VMService_RemoveUserTagsFromStreamableSampleList, 0, 1) {
+#ifndef PRODUCT
+  GET_NON_NULL_NATIVE_ARGUMENT(GrowableObjectArray, user_tags,
+                               arguments->NativeArgAt(0));
+
+  Object& obj = Object::Handle();
+  for (intptr_t i = 0; i < user_tags.Length(); ++i) {
+    obj = user_tags.At(i);
+    UserTags::RemoveStreamableTagName(obj.ToCString());
+  }
+  UserTagIsolatesVisitor visitor(thread, &user_tags, false);
+  Isolate::VisitIsolates(&visitor);
+#endif
+  return Object::null();
+}
+
 }  // namespace dart

runtime/observatory/tests/service/get_version_rpc_test.dart

@@ -12,7 +12,7 @@ var tests = <VMTest>[
     final result = await vm.invokeRpcNoUpgrade('getVersion', {});
     expect(result['type'], 'Version');
     expect(result['major'], 3);
-    expect(result['minor'], 54);
+    expect(result['minor'], 55);
     expect(result['_privateMajor'], 0);
     expect(result['_privateMinor'], 0);
   },

runtime/observatory_2/tests/service_2/get_version_rpc_test.dart

@@ -12,7 +12,7 @@ var tests = <VMTest>[
     final result = await vm.invokeRpcNoUpgrade('getVersion', {});
     expect(result['type'], equals('Version'));
     expect(result['major'], equals(3));
-    expect(result['minor'], equals(54));
+    expect(result['minor'], equals(55));
     expect(result['_privateMajor'], equals(0));
     expect(result['_privateMinor'], equals(0));
   },

runtime/vm/bootstrap_natives.h

@@ -365,6 +365,8 @@ namespace dart {
   V(VMService_CancelStream, 1)                                                 \
   V(VMService_RequestAssets, 0)                                                \
   V(VMService_DecodeAssets, 1)                                                 \
+  V(VMService_AddUserTagsToStreamableSampleList, 1)                            \
+  V(VMService_RemoveUserTagsFromStreamableSampleList, 1)                       \
   V(Ffi_loadInt8, 2)                                                           \
   V(Ffi_loadInt16, 2)                                                          \
   V(Ffi_loadInt32, 2)                                                          \

runtime/vm/compiler/runtime_offsets_extracted.h

@@ -571,7 +571,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     12;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 16;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 12;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 12;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 16;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 16;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 12;
@@ -1143,7 +1143,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     24;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 32;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 24;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 32;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 24;
@@ -1700,7 +1700,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     12;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 16;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 12;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 12;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 16;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 16;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 12;
@@ -2273,7 +2273,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     24;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 32;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 24;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 32;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 24;
@@ -2842,7 +2842,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     16;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 32;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 16;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 24;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 16;
@@ -3412,7 +3412,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     16;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 32;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 16;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 24;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 16;
@@ -3968,7 +3968,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     12;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 16;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 12;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 12;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 16;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 16;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 12;
@@ -4534,7 +4534,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     24;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 32;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 24;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 32;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 24;
@@ -5085,7 +5085,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     12;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 16;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 12;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 12;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 16;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 16;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 12;
@@ -5652,7 +5652,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     24;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 32;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 24;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 32;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 24;
@@ -6215,7 +6215,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     16;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 32;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 16;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 24;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 16;
@@ -6779,7 +6779,7 @@ static constexpr dart::compiler::target::word UnhandledException_InstanceSize =
     16;
 static constexpr dart::compiler::target::word UnlinkedCall_InstanceSize = 32;
 static constexpr dart::compiler::target::word UnwindError_InstanceSize = 16;
-static constexpr dart::compiler::target::word UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word WeakProperty_InstanceSize = 24;
 static constexpr dart::compiler::target::word
     WeakSerializationReference_InstanceSize = 16;
@@ -7411,7 +7411,7 @@ static constexpr dart::compiler::target::word
 static constexpr dart::compiler::target::word AOT_UnlinkedCall_InstanceSize =
     16;
 static constexpr dart::compiler::target::word AOT_UnwindError_InstanceSize = 12;
-static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 12;
+static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 16;
 static constexpr dart::compiler::target::word AOT_WeakProperty_InstanceSize =
     16;
 static constexpr dart::compiler::target::word
@@ -8044,7 +8044,7 @@ static constexpr dart::compiler::target::word
 static constexpr dart::compiler::target::word AOT_UnlinkedCall_InstanceSize =
     32;
 static constexpr dart::compiler::target::word AOT_UnwindError_InstanceSize = 24;
-static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word AOT_WeakProperty_InstanceSize =
     32;
 static constexpr dart::compiler::target::word
@@ -8681,7 +8681,7 @@ static constexpr dart::compiler::target::word
 static constexpr dart::compiler::target::word AOT_UnlinkedCall_InstanceSize =
     32;
 static constexpr dart::compiler::target::word AOT_UnwindError_InstanceSize = 24;
-static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word AOT_WeakProperty_InstanceSize =
     32;
 static constexpr dart::compiler::target::word
@@ -9313,7 +9313,7 @@ static constexpr dart::compiler::target::word
 static constexpr dart::compiler::target::word AOT_UnlinkedCall_InstanceSize =
     32;
 static constexpr dart::compiler::target::word AOT_UnwindError_InstanceSize = 16;
-static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word AOT_WeakProperty_InstanceSize =
     24;
 static constexpr dart::compiler::target::word
@@ -9946,7 +9946,7 @@ static constexpr dart::compiler::target::word
 static constexpr dart::compiler::target::word AOT_UnlinkedCall_InstanceSize =
     32;
 static constexpr dart::compiler::target::word AOT_UnwindError_InstanceSize = 16;
-static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word AOT_WeakProperty_InstanceSize =
     24;
 static constexpr dart::compiler::target::word
@@ -10570,7 +10570,7 @@ static constexpr dart::compiler::target::word
 static constexpr dart::compiler::target::word AOT_UnlinkedCall_InstanceSize =
     16;
 static constexpr dart::compiler::target::word AOT_UnwindError_InstanceSize = 12;
-static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 12;
+static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 16;
 static constexpr dart::compiler::target::word AOT_WeakProperty_InstanceSize =
     16;
 static constexpr dart::compiler::target::word
@@ -11196,7 +11196,7 @@ static constexpr dart::compiler::target::word
 static constexpr dart::compiler::target::word AOT_UnlinkedCall_InstanceSize =
     32;
 static constexpr dart::compiler::target::word AOT_UnwindError_InstanceSize = 24;
-static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word AOT_WeakProperty_InstanceSize =
     32;
 static constexpr dart::compiler::target::word
@@ -11826,7 +11826,7 @@ static constexpr dart::compiler::target::word
 static constexpr dart::compiler::target::word AOT_UnlinkedCall_InstanceSize =
     32;
 static constexpr dart::compiler::target::word AOT_UnwindError_InstanceSize = 24;
-static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word AOT_WeakProperty_InstanceSize =
     32;
 static constexpr dart::compiler::target::word
@@ -12451,7 +12451,7 @@ static constexpr dart::compiler::target::word
 static constexpr dart::compiler::target::word AOT_UnlinkedCall_InstanceSize =
     32;
 static constexpr dart::compiler::target::word AOT_UnwindError_InstanceSize = 16;
-static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word AOT_WeakProperty_InstanceSize =
     24;
 static constexpr dart::compiler::target::word
@@ -13077,7 +13077,7 @@ static constexpr dart::compiler::target::word
 static constexpr dart::compiler::target::word AOT_UnlinkedCall_InstanceSize =
     32;
 static constexpr dart::compiler::target::word AOT_UnwindError_InstanceSize = 16;
-static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 24;
+static constexpr dart::compiler::target::word AOT_UserTag_InstanceSize = 32;
 static constexpr dart::compiler::target::word AOT_WeakProperty_InstanceSize =
     24;
 static constexpr dart::compiler::target::word

runtime/vm/dart.cc

@@ -44,6 +44,7 @@
 #include "vm/stack_frame.h"
 #include "vm/stub_code.h"
 #include "vm/symbols.h"
+#include "vm/tags.h"
 #include "vm/thread_interrupter.h"
 #include "vm/thread_pool.h"
 #include "vm/timeline.h"
@@ -340,6 +341,7 @@ char* Dart::DartInit(const uint8_t* vm_isolate_snapshot,
 #endif
   IsolateGroup::Init();
   Isolate::InitVM();
+  UserTags::Init();
   PortMap::Init();
   FreeListElement::Init();
   ForwardingCorpse::Init();
@@ -801,6 +803,7 @@ char* Dart::Cleanup() {
   vm_isolate_ = NULL;
   ASSERT(Isolate::IsolateListLength() == 0);
   PortMap::Cleanup();
+  UserTags::Cleanup();
   IsolateGroup::Cleanup();
   ICData::Cleanup();
   SubtypeTestCache::Cleanup();

runtime/vm/isolate.cc

@@ -2400,6 +2400,18 @@ void Isolate::FreeSampleBlock(SampleBlock* block) {
                                                    std::memory_order_release));
 }
 
+class StreamableSampleFilter : public SampleFilter {
+ public:
+  explicit StreamableSampleFilter(Dart_Port port)
+      : SampleFilter(port, kNoTaskFilter, -1, -1) {}
+
+  bool FilterSample(Sample* sample) override {
+    const UserTag& tag =
+        UserTag::Handle(UserTag::FindTagById(sample->user_tag()));
+    return tag.streamable();
+  }
+};
+
 void Isolate::ProcessFreeSampleBlocks(Thread* thread) {
   SampleBlock* head = free_block_list_.exchange(nullptr);
   if (head == nullptr) {
@@ -2426,11 +2438,14 @@ void Isolate::ProcessFreeSampleBlocks(Thread* thread) {
     SampleBlockListProcessor buffer(head);
     StackZone zone(thread);
     HandleScope handle_scope(thread);
+    StreamableSampleFilter filter(main_port());
     Profile profile;
-    profile.Build(thread, nullptr, &buffer);
-    ServiceEvent event(this, ServiceEvent::kCpuSamples);
-    event.set_cpu_profile(&profile);
-    Service::HandleEvent(&event);
+    profile.Build(thread, &filter, &buffer);
+    if (profile.sample_count() > 0) {
+      ServiceEvent event(this, ServiceEvent::kCpuSamples);
+      event.set_cpu_profile(&profile);
+      Service::HandleEvent(&event);
+    }
   }
 
   do {

runtime/vm/object.cc

@@ -26184,6 +26184,7 @@ UserTagPtr UserTag::New(const String& label, Heap::Space space) {
     result ^= raw;
   }
   result.set_label(label);
+  result.set_streamable(UserTags::IsTagNameStreamable(label.ToCString()));
   AddTagToIsolate(thread, result);
   return result.ptr();
 }
@@ -26205,10 +26206,13 @@ UserTagPtr UserTag::DefaultTag() {
   return result.ptr();
 }
 
-UserTagPtr UserTag::FindTagInIsolate(Thread* thread, const String& label) {
-  Isolate* isolate = thread->isolate();
+UserTagPtr UserTag::FindTagInIsolate(Isolate* isolate,
+                                     Thread* thread,
+                                     const String& label) {
   Zone* zone = thread->zone();
-  ASSERT(isolate->tag_table() != GrowableObjectArray::null());
+  if (isolate->tag_table() == GrowableObjectArray::null()) {
+    return UserTag::null();
+  }
   const GrowableObjectArray& tag_table =
       GrowableObjectArray::Handle(zone, isolate->tag_table());
   UserTag& other = UserTag::Handle(zone);
@@ -26225,6 +26229,11 @@ UserTagPtr UserTag::FindTagInIsolate(Thread* thread, const String& label) {
   return UserTag::null();
 }
 
+UserTagPtr UserTag::FindTagInIsolate(Thread* thread, const String& label) {
+  Isolate* isolate = thread->isolate();
+  return FindTagInIsolate(isolate, thread, label);
+}
+
 void UserTag::AddTagToIsolate(Thread* thread, const UserTag& tag) {
   Isolate* isolate = thread->isolate();
   Zone* zone = thread->zone();

runtime/vm/object.h

@@ -12030,6 +12030,12 @@ class UserTag : public Instance {
     ASSERT(t < UserTags::kUserTagIdOffset + UserTags::kMaxUserTags);
     StoreNonPointer(&untag()->tag_, t);
   }
+
+  bool streamable() const { return untag()->streamable(); }
+  void set_streamable(bool streamable) {
+    StoreNonPointer(&untag()->streamable_, streamable);
+  }
+
   static intptr_t tag_offset() { return OFFSET_OF(UntaggedUserTag, tag_); }
 
   StringPtr label() const { return untag()->label(); }
@@ -12045,6 +12051,9 @@ class UserTag : public Instance {
 
   static bool TagTableIsFull(Thread* thread);
   static UserTagPtr FindTagById(uword tag_id);
+  static UserTagPtr FindTagInIsolate(Isolate* isolate,
+                                     Thread* thread,
+                                     const String& label);
 
  private:
   static UserTagPtr FindTagInIsolate(Thread* thread, const String& label);

runtime/vm/profiler_service.cc

@@ -955,12 +955,12 @@ class ProfileBuilder : public ValueObject {
     if (!FilterSamples()) {
       return;
     }
-
     Setup();
     BuildCodeTable();
     FinalizeCodeIndexes();
     BuildFunctionTable();
     PopulateFunctionTicks();
+    SanitizeMinMaxTimes();
   }
 
  private:
@@ -1062,7 +1062,6 @@ class ProfileBuilder : public ValueObject {
       TickExitFrame(sample->vm_tag(), sample_index, sample);
       thread_->CheckForSafepoint();
     }
-    SanitizeMinMaxTimes();
   }
 
   void FinalizeCodeIndexes() {

runtime/vm/raw_object.h

@@ -3283,10 +3283,14 @@ class UntaggedUserTag : public UntaggedInstance {
   // Isolate unique tag.
   uword tag_;
 
+  // Should CPU samples with this tag be streamed?
+  bool streamable_;
+
   friend class Object;
 
  public:
   uword tag() const { return tag_; }
+  bool streamable() const { return streamable_; }
 };
 
 class UntaggedFutureOr : public UntaggedInstance {

runtime/vm/service.h

@@ -15,7 +15,7 @@
 namespace dart {
 
 #define SERVICE_PROTOCOL_MAJOR_VERSION 3
-#define SERVICE_PROTOCOL_MINOR_VERSION 54
+#define SERVICE_PROTOCOL_MINOR_VERSION 55
 
 class Array;
 class EmbedderServiceHandler;

runtime/vm/service/service.md

@@ -1,8 +1,8 @@
-# Dart VM Service Protocol 3.54
+# Dart VM Service Protocol 3.55
 
 > Please post feedback to the [observatory-discuss group][discuss-list]
 
-This document describes of _version 3.54_ of the Dart VM Service Protocol. This
+This document describes of _version 3.55_ of the Dart VM Service Protocol. This
 protocol is used to communicate with a running Dart Virtual Machine.
 
 To use the Service Protocol, start the VM with the *--observe* flag.
@@ -79,6 +79,7 @@ The Service Protocol uses [JSON-RPC 2.0][].
   - [setVMName](#setvmname)
   - [setVMTimelineFlags](#setvmtimelineflags)
   - [streamCancel](#streamcancel)
+  - [streamCpuSamplesWithUserTag](#streamcpusampleswithusertag)
   - [streamListen](#streamlisten)
 - [Public Types](#public-types)
   - [AllocationProfile](#allocationprofile)
@@ -1536,6 +1537,19 @@ subscribed) [RPC error](#rpc-error) code is returned.
 
 See [Success](#success).
 
+### streamCpuSamplesWithUserTag
+
+```
+Success streamCpuSamplesWithUserTag(string[] userTags)
+```
+
+The _streamCpuSamplesWithUserTag_ RPC allows for clients to specify which CPU
+samples collected by the profiler should be sent over the `Profiler` stream.
+When called, the VM will stream `CpuSamples` events containing `CpuSample`'s
+collected while a user tag contained in `userTags` was active.
+
+See [Success](#success).
+
 ### streamListen
 
 ```
@@ -4304,5 +4318,6 @@ version | comments
 3.52 | Added `lookupResolvedPackageUris` and `lookupPackageUris` RPCs and `UriList` type.
 3.53 | Added `setIsolatePauseMode` RPC.
 3.54 | Added `CpuSamplesEvent`, updated `cpuSamples` property on `Event` to have type `CpuSamplesEvent`.
+3.55 | Added `streamCpuSamplesWithUserTag` RPC.
 
 [discuss-list]: https://groups.google.com/a/dartlang.org/forum/#!forum/observatory-discuss

runtime/vm/tags.cc

@@ -12,6 +12,9 @@
 
 namespace dart {
 
+MallocGrowableArray<const char*> UserTags::subscribed_tags_(4);
+Mutex* UserTags::subscribed_tags_lock_ = nullptr;
+
 const char* VMTag::TagName(uword tag) {
   if (IsNativeEntryTag(tag)) {
     const uint8_t* native_reverse_lookup = NativeEntry::ResolveSymbol(tag);
@@ -148,4 +151,55 @@ const char* UserTags::TagName(uword tag_id) {
   return label.ToCString();
 }
 
+void UserTags::AddStreamableTagName(const char* tag) {
+  MutexLocker ml(subscribed_tags_lock_);
+  // Check this tag isn't already in the subscription list.
+  for (intptr_t i = 0; i < subscribed_tags_.length(); ++i) {
+    if (strcmp(tag, subscribed_tags_.At(i)) == 0) {
+      return;
+    }
+  }
+  subscribed_tags_.Add(strdup(tag));
+}
+
+void UserTags::RemoveStreamableTagName(const char* tag) {
+  MutexLocker ml(subscribed_tags_lock_);
+  bool found = false;
+  for (intptr_t i = 0; i < subscribed_tags_.length(); ++i) {
+    if (strcmp(tag, subscribed_tags_.At(i)) == 0) {
+      free(const_cast<char*>(subscribed_tags_.At(i)));
+      subscribed_tags_.RemoveAt(i);
+      found = true;
+      break;
+    }
+  }
+  ASSERT(found);
+}
+
+bool UserTags::IsTagNameStreamable(const char* tag) {
+  MutexLocker ml(subscribed_tags_lock_);
+  for (intptr_t i = 0; i < subscribed_tags_.length(); ++i) {
+    if (strcmp(tag, subscribed_tags_.At(i)) == 0) {
+      return true;
+    }
+  }
+  return false;
+}
+
+void UserTags::Init() {
+  subscribed_tags_lock_ = new Mutex();
+}
+
+void UserTags::Cleanup() {
+  {
+    MutexLocker ml(subscribed_tags_lock_);
+    for (intptr_t i = 0; i < subscribed_tags_.length(); ++i) {
+      free(const_cast<char*>(subscribed_tags_.At(i)));
+    }
+    subscribed_tags_.Clear();
+  }
+  delete subscribed_tags_lock_;
+  subscribed_tags_lock_ = nullptr;
+}
+
 }  // namespace dart

runtime/vm/tags.h

@@ -5,7 +5,9 @@
 #ifndef RUNTIME_VM_TAGS_H_
 #define RUNTIME_VM_TAGS_H_
 
+#include "platform/growable_array.h"
 #include "vm/allocation.h"
+#include "vm/os_thread.h"
 #include "vm/thread_stack_resource.h"
 
 namespace dart {
@@ -113,6 +115,15 @@ class UserTags : public AllStatic {
     return (tag_id >= kUserTagIdOffset) &&
            (tag_id < kUserTagIdOffset + kMaxUserTags);
   }
+  static void AddStreamableTagName(const char* tag);
+  static void RemoveStreamableTagName(const char* tag);
+  static bool IsTagNameStreamable(const char* tag);
+  static void Init();
+  static void Cleanup();
+
+ private:
+  static Mutex* subscribed_tags_lock_;
+  static MallocGrowableArray<const char*> subscribed_tags_;
 };
 
 }  // namespace dart

sdk/lib/vmservice/client.dart

@@ -20,9 +20,12 @@ abstract class Client {
   set name(String? n) => _name = (n ?? defaultClientName);
   late String _name;
 
-  /// A set streamIds which describes the streams the client is connected to
+  /// The set of streams the client is subscribed to.
   final streams = <String>{};
 
+  /// The set of user tags that the client wants to receive CPU samples for.
+  final profilerUserTagFilters = <String>{};
+
   /// Services registered and their aliases
   /// key: service
   /// value: alias

sdk/lib/vmservice/vmservice.dart

@@ -228,6 +228,8 @@ class VMService extends MessageRouter {
 
   final devfs = DevFS();
 
+  final _profilerUserTagSubscriptions = <String>{};
+
   Uri? get ddsUri => _ddsUri;
   Uri? _ddsUri;
 
@@ -300,7 +302,7 @@ class VMService extends MessageRouter {
         _vmCancelStream(streamId);
       }
     }
-
+    _cleanupUnusedUserTagSubscriptions();
     for (final service in client.services.keys) {
       _eventMessageHandler(
           'Service',
@@ -336,10 +338,35 @@ class VMService extends MessageRouter {
     }
   }
 
+  void _profilerEventMessageHandler(Client client, Response event) {
+    final eventJson = event.decodeJson() as Map<String, dynamic>;
+    final eventData = eventJson['params']!['event']!;
+    if (eventData['kind']! != 'CpuSamples') {
+      client.post(event);
+      return;
+    }
+    final cpuSamplesEvent = eventData['cpuSamples']!;
+    final updatedSamples = cpuSamplesEvent['samples']!
+        .where(
+          (s) => client.profilerUserTagFilters.contains(s['userTag']),
+        )
+        .toList();
+    if (updatedSamples.isEmpty) {
+      return;
+    }
+    cpuSamplesEvent['samples'] = updatedSamples;
+    cpuSamplesEvent['sampleCount'] = updatedSamples.length;
+    client.post(Response.json(eventJson));
+  }
+
   void _eventMessageHandler(String streamId, Response event) {
     for (final client in clients) {
       if (client.sendEvents && client.streams.contains(streamId)) {
-        client.post(event);
+        if (streamId == 'Profiler') {
+          _profilerEventMessageHandler(client, event);
+        } else {
+          client.post(event);
+        }
       }
     }
   }
@@ -665,6 +692,57 @@ class VMService extends MessageRouter {
     return encodeResult(message, protocols);
   }
 
+  void _cleanupUnusedUserTagSubscriptions() {
+    final unsubscribeableTags = <String>[];
+    for (final subscribedTag in _profilerUserTagSubscriptions) {
+      bool hasSubscriber = false;
+      for (final c in clients) {
+        if (c.profilerUserTagFilters.contains(subscribedTag)) {
+          hasSubscriber = true;
+          break;
+        }
+      }
+      if (!hasSubscriber) {
+        unsubscribeableTags.add(subscribedTag);
+      }
+    }
+    if (unsubscribeableTags.isNotEmpty) {
+      _profilerUserTagSubscriptions.removeAll(unsubscribeableTags);
+      _removeUserTagsFromStreamableSampleList(unsubscribeableTags);
+    }
+  }
+
+  Future<String> _streamCpuSamplesWithUserTag(Message message) async {
+    if (!message.params.containsKey('userTags')) {
+      return encodeRpcError(message, kInvalidParams,
+          details: "Missing required parameter 'userTags'.");
+    }
+
+    // TODO(bkonyi): handle "subscribe all" case.
+    final client = message.client!;
+    final tags = message.params['userTags']!.cast<String>().toSet();
+    final newTags = tags.difference(_profilerUserTagSubscriptions);
+
+    // Clear the previously set user tag subscriptions for the client and
+    // update with the new list of user tags.
+    client.profilerUserTagFilters.clear();
+    _profilerUserTagSubscriptions.addAll(tags);
+    client.profilerUserTagFilters.addAll(tags);
+
+    // If any previously unseen user tag is provided, let the VM know that
+    // samples with that user tag should be streamed on the Profiler stream.
+    if (newTags.isNotEmpty) {
+      _addUserTagsToStreamableSampleList(newTags.toList());
+    }
+
+    // Some user tags may no longer be of any interest to the existing clients.
+    // Check that all user tags have at least one client interested in them,
+    // otherwise notify the VM that we're no longer interested in samples with
+    // those user tags.
+    _cleanupUnusedUserTagSubscriptions();
+    return encodeSuccess(message);
+  }
+
   Future<Response?> routeRequest(VMService _, Message message) async {
     final response = await _routeRequestImpl(message);
     if (response == null) {
@@ -695,6 +773,9 @@ class VMService extends MessageRouter {
       if (message.method == 'getSupportedProtocols') {
         return await _getSupportedProtocols(message);
       }
+      if (message.method == 'streamCpuSamplesWithUserTag') {
+        return await _streamCpuSamplesWithUserTag(message);
+      }
       if (devfs.shouldHandleMessage(message)) {
         return await devfs.handleMessage(message);
       }
@@ -759,3 +840,9 @@ external void _vmCancelStream(String streamId);
 /// Get the bytes to the tar archive.
 @pragma("vm:external-name", "VMService_RequestAssets")
 external Uint8List _requestAssets();
+
+@pragma("vm:external-name", "VMService_AddUserTagsToStreamableSampleList")
+external void _addUserTagsToStreamableSampleList(List<String> userTags);
+
+@pragma("vm:external-name", "VMService_RemoveUserTagsFromStreamableSampleList")
+external void _removeUserTagsFromStreamableSampleList(List<String> userTags);