[ package:dds ] Add locking when modifying DDS state via client requests

Fixes https://github.com/dart-lang/sdk/issues/46696

Change-Id: I666b59a0661f4df3b1f0a47aba52096133f5fbb7
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/209140
Reviewed-by: Anna Gringauze <annagrin@google.com>

.dart_tool/package_config.json

@@ -11,7 +11,7 @@
     "constraint, update this by running tools/generate_package_config.dart."
   ],
   "configVersion": 2,
-  "generated": "2021-08-04T16:42:24.433381",
+  "generated": "2021-08-05T11:33:04.746536",
   "generator": "tools/generate_package_config.dart",
   "packages": [
     {
@@ -256,7 +256,7 @@
       "name": "dds",
       "rootUri": "../pkg/dds",
       "packageUri": "lib/",
-      "languageVersion": "2.12"
+      "languageVersion": "2.14"
     },
     {
       "name": "dev_compiler",

pkg/dds/lib/src/isolate_manager.dart

@@ -2,6 +2,7 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
+import 'package:dds/src/utils/mutex.dart';
 import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
 
 import 'client.dart';
@@ -139,6 +140,8 @@ class IsolateManager {
   }
 
   void _updateIsolateState(String id, String name, String eventKind) {
+    _mutex.runGuarded(
+      () {
         switch (eventKind) {
           case ServiceEvents.isolateStart:
             isolateStarted(id, name);
@@ -165,6 +168,8 @@ class IsolateManager {
                 break;
             }
         }
+      },
+    );
   }
 
   /// Initializes the set of running isolates.
@@ -172,8 +177,11 @@ class IsolateManager {
     if (_initialized) {
       return;
     }
+    await _mutex.runGuarded(
+      () async {
         final vm = await dds.vmServiceClient.sendRequest('getVM');
-    final List<Map> isolateRefs = vm['isolates'].cast<Map<String, dynamic>>();
+        final List<Map> isolateRefs =
+            vm['isolates'].cast<Map<String, dynamic>>();
         // Check the pause event for each isolate to determine whether or not the
         // isolate is already paused.
         for (final isolateRef in isolateRefs) {
@@ -191,6 +199,8 @@ class IsolateManager {
             isolateStarted(id, name);
           }
         }
+      },
+    );
     _initialized = true;
   }
 
@@ -218,6 +228,8 @@ class IsolateManager {
     DartDevelopmentServiceClient client,
     json_rpc.Parameters parameters,
   ) async {
+    return await _mutex.runGuarded(
+      () async {
         final isolateId = parameters['isolateId'].asString;
         final isolate = isolates[isolateId];
         if (isolate == null) {
@@ -228,6 +240,8 @@ class IsolateManager {
           return await _sendResumeRequest(isolateId, parameters);
         }
         return RPCResponses.success;
+      },
+    );
   }
 
   /// Forwards a `resume` request to the VM service.
@@ -248,5 +262,6 @@ class IsolateManager {
 
   bool _initialized = false;
   final DartDevelopmentServiceImpl dds;
+  final _mutex = Mutex();
   final Map<String, _RunningIsolate> isolates = {};
 }

pkg/dds/lib/src/stream_manager.dart

@@ -10,6 +10,7 @@ import 'client.dart';
 import 'dds_impl.dart';
 import 'logging_repository.dart';
 import 'rpc_error_codes.dart';
+import 'utils/mutex.dart';
 
 class StreamManager {
   StreamManager(this.dds);
@@ -133,6 +134,8 @@ class StreamManager {
     DartDevelopmentServiceClient? client,
     String stream,
   ) async {
+    await _mutex.runGuarded(
+      () async {
         assert(stream.isNotEmpty);
         if (!streamListeners.containsKey(stream)) {
           // Initialize the list of clients for the new stream before we do
@@ -144,7 +147,8 @@ class StreamManager {
               stream != kDebugStream) {
             // This will return an RPC exception if the stream doesn't exist. This
             // will throw and the exception will be forwarded to the client.
-        final result = await dds.vmServiceClient.sendRequest('streamListen', {
+            final result =
+                await dds.vmServiceClient.sendRequest('streamListen', {
               'streamId': stream,
             });
             assert(result['type'] == 'Success');
@@ -178,6 +182,8 @@ class StreamManager {
             }
           }
         }
+      },
+    );
   }
 
   List<Map<String, dynamic>>? getStreamHistory(String stream) {
@@ -198,9 +204,12 @@ class StreamManager {
     String stream, {
     bool cancelCoreStream = false,
   }) async {
+    await _mutex.runGuarded(
+      () async {
         assert(stream.isNotEmpty);
         final listeners = streamListeners[stream];
-    if (listeners == null || client != null && !listeners.contains(client)) {
+        if (listeners == null ||
+            client != null && !listeners.contains(client)) {
           throw kStreamNotSubscribedException;
         }
         listeners.remove(client);
@@ -219,6 +228,8 @@ class StreamManager {
         } else {
           streamListeners[stream] = listeners;
         }
+      },
+    );
   }
 
   /// Cleanup stream subscriptions for `client` when it has disconnected.
@@ -280,4 +291,5 @@ class StreamManager {
 
   final DartDevelopmentServiceImpl dds;
   final streamListeners = <String, List<DartDevelopmentServiceClient>>{};
+  final _mutex = Mutex();
 }

pkg/dds/lib/src/utils/mutex.dart

@@ -0,0 +1,48 @@
+// 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:collection';
+
+typedef _LockRequest = Completer<void>;
+
+/// Used to protect global state accessed in blocks containing calls to
+/// asynchronous methods.
+class Mutex {
+  /// Executes a block of code containing asynchronous calls atomically.
+  ///
+  /// If no other asynchronous context is currently executing within
+  /// [criticalSection], it will immediately be called. Otherwise, the caller
+  /// will be suspended and entered into a queue to be resumed once the lock is
+  /// released.
+  Future<T> runGuarded<T>(FutureOr<T> Function() criticalSection) async {
+    try {
+      await _acquireLock();
+      return await criticalSection();
+    } finally {
+      _releaseLock();
+    }
+  }
+
+  Future<void> _acquireLock() async {
+    if (!_locked) {
+      _locked = true;
+      return;
+    }
+    final request = _LockRequest();
+    _outstandingRequests.add(request);
+    await request.future;
+  }
+
+  void _releaseLock() {
+    _locked = false;
+    if (_outstandingRequests.isNotEmpty) {
+      final request = _outstandingRequests.removeFirst();
+      request.complete();
+    }
+  }
+
+  bool _locked = false;
+  final _outstandingRequests = Queue<_LockRequest>();
+}

pkg/dds/pubspec.yaml

@@ -8,7 +8,7 @@ version: 2.0.2
 homepage: https://github.com/dart-lang/sdk/tree/master/pkg/dds
 
 environment:
-  sdk: '>=2.12.0 <3.0.0'
+  sdk: '>=2.14.0 <3.0.0'
 
 dependencies:
   async: ^2.4.1

pkg/dds/test/regress_45569_test.dart

@@ -3,10 +3,13 @@
 // BSD-style license that can be found in the LICENSE file.
 
 import 'dart:io';
+import 'dart:math';
 
 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() {
@@ -27,8 +30,20 @@ void main() {
     process.kill();
   });
 
+  Future<void> streamSubscribeUnsubscribe(
+    VmService client, {
+    required bool delay,
+  }) async {
+    await client.streamListen('Service');
+    await Future.delayed(
+      Duration(milliseconds: delay ? Random().nextInt(200) : 0),
+    );
+    await client.streamCancel('Service');
+  }
+
   test('Ensure streamListen and streamCancel calls are handled atomically',
       () async {
+    for (int i = 0; i < 100; ++i) {
       dds = await DartDevelopmentService.startDartDevelopmentService(
         remoteVmServiceUri,
       );
@@ -36,18 +51,11 @@ void main() {
       final connection1 = await vmServiceConnectUri(dds.wsUri.toString());
       final connection2 = await vmServiceConnectUri(dds.wsUri.toString());
 
-    for (int i = 0; i < 50; ++i) {
-      final listenFutures = <Future>[
-        connection1.streamListen('Service'),
-        connection2.streamListen('Service'),
-      ];
-      await Future.wait(listenFutures);
-
-      final cancelFutures = <Future>[
-        connection1.streamCancel('Service'),
-        connection2.streamCancel('Service'),
-      ];
-      await Future.wait(cancelFutures);
+      await Future.wait([
+        streamSubscribeUnsubscribe(connection1, delay: true),
+        streamSubscribeUnsubscribe(connection2, delay: false),
+      ]);
+      await dds.shutdown();
     }
   });
 }