Add debounce support to daemon hot reload requests (#55376)

packages/flutter_tools/doc/daemon.md

@@ -106,6 +106,7 @@ The `restart()` restarts the given application. It returns a Map of `{ int code,
 - `fullRestart`: optional; whether to do a full (rather than an incremental) restart of the application
 - `reason`: optional; the reason for the full restart (eg. `save`, `manual`) for reporting purposes
 - `pause`: optional; when doing a hot restart the isolate should enter a paused mode
+- `debounce`: optional; whether to automatically debounce multiple requests sent in quick succession (this may introduce a short delay in processing the request)
 
 #### app.reloadMethod
 
@@ -115,6 +116,7 @@ Performs a limited hot restart which does not sync assets and only marks element
 - `library`: the absolute file URI of the library to be updated; this is required.
 - `class`: the name of the StatelessWidget that was updated, or the StatefulWidget
 corresponding to the updated State class; this is required.
+- `debounce`: optional; whether to automatically debounce multiple requests sent in quick succession (this may introduce a short delay in processing the request)
 
 #### app.callServiceExtension
 
@@ -289,6 +291,7 @@ See the [source](https://github.com/flutter/flutter/blob/master/packages/flutter
 
 ## Changelog
 
+- 0.6.0: Added `debounce` option to `app.restart` command.
 - 0.5.3: Added `emulatorId` field to device.
 - 0.5.2: Added `platformType` and `category` fields to emulator.
 - 0.5.1: Added `platformType`, `ephemeral`, and `category` fields to device.

packages/flutter_tools/lib/src/commands/daemon.dart

@@ -4,6 +4,7 @@
 
 import 'dart:async';
 
+import 'package:async/async.dart';
 import 'package:meta/meta.dart';
 import 'package:uuid/uuid.dart';
 
@@ -26,7 +27,7 @@ import '../run_hot.dart';
 import '../runner/flutter_command.dart';
 import '../web/web_runner.dart';
 
-const String protocolVersion = '0.5.3';
+const String protocolVersion = '0.6.0';
 
 /// A server process command. This command will start up a long-lived server.
 /// It reads JSON-RPC based commands from stdin, executes them, and returns
@@ -431,6 +432,8 @@ class AppDomain extends Domain {
 
   final List<AppInstance> _apps = <AppInstance>[];
 
+  final DebounceOperationQueue<OperationResult, OperationType> operationQueue = DebounceOperationQueue<OperationResult, OperationType>();
+
   Future<AppInstance> startApp(
     Device device,
     String projectDirectory,
@@ -592,51 +595,81 @@ class AppDomain extends Domain {
   bool isRestartSupported(bool enableHotReload, Device device) =>
       enableHotReload && device.supportsHotRestart;
 
-  Future<OperationResult> _inProgressHotReload;
+  final int _hotReloadDebounceDurationMs = 50;
 
   Future<OperationResult> restart(Map<String, dynamic> args) async {
     final String appId = _getStringArg(args, 'appId', required: true);
     final bool fullRestart = _getBoolArg(args, 'fullRestart') ?? false;
     final bool pauseAfterRestart = _getBoolArg(args, 'pause') ?? false;
     final String restartReason = _getStringArg(args, 'reason');
+    final bool debounce = _getBoolArg(args, 'debounce') ?? false;
+    // This is an undocumented parameter used for integration tests.
+    final int debounceDurationOverrideMs = _getIntArg(args, 'debounceDurationOverrideMs');
 
     final AppInstance app = _getApp(appId);
     if (app == null) {
       throw "app '$appId' not found";
     }
 
-    if (_inProgressHotReload != null) {
+    return _queueAndDebounceReloadAction(
-      throw 'hot restart already in progress';
+      app,
-    }
+      fullRestart ? OperationType.restart: OperationType.reload,
-
+      debounce,
-    _inProgressHotReload = app._runInZone<OperationResult>(this, () {
+      debounceDurationOverrideMs,
-      return app.restart(fullRestart: fullRestart, pause: pauseAfterRestart, reason: restartReason);
+      () {
-    });
+        return app.restart(
-    return _inProgressHotReload.whenComplete(() {
+            fullRestart: fullRestart,
-      _inProgressHotReload = null;
+            pause: pauseAfterRestart,
-    });
+            reason: restartReason);
+      },
+    );
   }
 
   Future<OperationResult> reloadMethod(Map<String, dynamic> args) async {
     final String appId = _getStringArg(args, 'appId', required: true);
     final String classId = _getStringArg(args, 'class', required: true);
-    final String libraryId =  _getStringArg(args, 'library', required: true);
+    final String libraryId = _getStringArg(args, 'library', required: true);
+    final bool debounce = _getBoolArg(args, 'debounce') ?? false;
 
     final AppInstance app = _getApp(appId);
     if (app == null) {
       throw "app '$appId' not found";
     }
 
-    if (_inProgressHotReload != null) {
+    return _queueAndDebounceReloadAction(
-      throw 'hot restart already in progress';
+      app,
-    }
+      OperationType.reloadMethod,
+      debounce,
+      null,
+      () {
+        return app.reloadMethod(classId: classId, libraryId: libraryId);
+      },
+    );
+  }
 
-    _inProgressHotReload = app._runInZone<OperationResult>(this, () {
+  /// Debounce and queue reload actions.
-      return app.reloadMethod(classId: classId, libraryId: libraryId);
+  ///
-    });
+  /// Only one reload action will run at a time. Actions requested in quick
-    return _inProgressHotReload.whenComplete(() {
+  /// succession (within [_hotReloadDebounceDuration]) will be merged together
-      _inProgressHotReload = null;
+  /// and all return the same result. If an action is requested after an identical
-    });
+  /// action has already started, it will be queued and run again once the first
+  /// action completes.
+  Future<OperationResult> _queueAndDebounceReloadAction(
+    AppInstance app,
+    OperationType operationType,
+    bool debounce,
+    int debounceDurationOverrideMs,
+    Future<OperationResult> Function() action,
+  ) {
+    final Duration debounceDuration = debounce
+        ? Duration(milliseconds: debounceDurationOverrideMs ?? _hotReloadDebounceDurationMs)
+        : Duration.zero;
+
+    return operationQueue.queueAndDebounce(
+      operationType,
+      debounceDuration,
+      () => app._runInZone<OperationResult>(this, action),
+    );
   }
 
   /// Returns an error, or the service extension result (a map with two fixed
@@ -1272,3 +1305,57 @@ class LaunchMode {
   @override
   String toString() => _value;
 }
+
+enum OperationType {
+  reloadMethod,
+  reload,
+  restart
+}
+
+/// A queue that debounces operations for a period and merges operations of the same type.
+/// Only one action (or any type) will run at a time. Actions of the same type requested
+/// in quick succession will be merged together and all return the same result. If an action
+/// is requested after an identical action has already started, it will be queued
+/// and run again once the first action completes.
+class DebounceOperationQueue<T, K> {
+  final Map<K, RestartableTimer> _debounceTimers = <K, RestartableTimer>{};
+  final Map<K, Future<T>> _operationQueue = <K, Future<T>>{};
+  Future<void> _inProgressAction;
+
+  Future<T> queueAndDebounce(
+    K operationType,
+    Duration debounceDuration,
+    Future<T> Function() action,
+  ) {
+    // If there is already an operation of this type waiting to run, reset its
+    // debounce timer and return its future.
+    if (_operationQueue[operationType] != null) {
+      _debounceTimers[operationType]?.reset();
+      return _operationQueue[operationType];
+    }
+
+    // Otherwise, put one in the queue with a timer.
+    final Completer<T> completer = Completer<T>();
+    _operationQueue[operationType] = completer.future;
+    _debounceTimers[operationType] = RestartableTimer(
+      debounceDuration,
+      () async {
+        // Remove us from the queue so we can't be reset now we've started.
+        unawaited(_operationQueue.remove(operationType));
+        _debounceTimers.remove(operationType);
+
+        // No operations should be allowed to run concurrently even if they're
+        // different types.
+        while (_inProgressAction != null) {
+          await _inProgressAction;
+        }
+
+        _inProgressAction = action()
+            .then(completer.complete, onError: completer.completeError)
+            .whenComplete(() => _inProgressAction = null);
+      },
+    );
+
+    return completer.future;
+  }
+}

packages/flutter_tools/test/commands.shard/hermetic/daemon_test.dart

@@ -13,6 +13,7 @@ import 'package:flutter_tools/src/fuchsia/fuchsia_workflow.dart';
 import 'package:flutter_tools/src/globals.dart' as globals;
 import 'package:flutter_tools/src/ios/ios_workflow.dart';
 import 'package:flutter_tools/src/resident_runner.dart';
+import 'package:quiver/testing/async.dart';
 
 import '../../src/common.dart';
 import '../../src/context.dart';
@@ -345,6 +346,89 @@ void main() {
       );
     });
   });
+
+  group('daemon queue', () {
+    DebounceOperationQueue<int, String> queue;
+    const Duration debounceDuration = Duration(seconds: 1);
+
+    setUp(() {
+      queue = DebounceOperationQueue<int, String>();
+    });
+
+    testWithoutContext(
+        'debounces/merges same operation type and returns same result',
+        () async {
+      await runFakeAsync((FakeAsync time) async {
+        final List<Future<int>> operations = <Future<int>>[
+          queue.queueAndDebounce('OP1', debounceDuration, () async => 1),
+          queue.queueAndDebounce('OP1', debounceDuration, () async => 2),
+        ];
+
+        time.elapse(debounceDuration * 5);
+        final List<int> results = await Future.wait(operations);
+
+        expect(results, orderedEquals(<int>[1, 1]));
+      });
+    });
+
+    testWithoutContext('does not merge results outside of the debounce duration',
+        () async {
+      await runFakeAsync((FakeAsync time) async {
+        final List<Future<int>> operations = <Future<int>>[
+          queue.queueAndDebounce('OP1', debounceDuration, () async => 1),
+          Future<int>.delayed(debounceDuration * 2).then((_) =>
+              queue.queueAndDebounce('OP1', debounceDuration, () async => 2)),
+        ];
+
+        time.elapse(debounceDuration * 5);
+        final List<int> results = await Future.wait(operations);
+
+        expect(results, orderedEquals(<int>[1, 2]));
+      });
+    });
+
+    testWithoutContext('does not merge results of different operations',
+        () async {
+      await runFakeAsync((FakeAsync time) async {
+        final List<Future<int>> operations = <Future<int>>[
+          queue.queueAndDebounce('OP1', debounceDuration, () async => 1),
+          queue.queueAndDebounce('OP2', debounceDuration, () async => 2),
+        ];
+
+        time.elapse(debounceDuration * 5);
+        final List<int> results = await Future.wait(operations);
+
+        expect(results, orderedEquals(<int>[1, 2]));
+      });
+    });
+
+    testWithoutContext('does not run any operations concurrently', () async {
+      // Crete a function thats slow, but throws if another instance of the
+      // function is running.
+      bool isRunning = false;
+      Future<int> f(int ret) async {
+        if (isRunning) {
+          throw 'Functions ran concurrently!';
+        }
+        isRunning = true;
+        await Future<void>.delayed(debounceDuration * 2);
+        isRunning = false;
+        return ret;
+      }
+
+      await runFakeAsync((FakeAsync time) async {
+        final List<Future<int>> operations = <Future<int>>[
+          queue.queueAndDebounce('OP1', debounceDuration, () => f(1)),
+          queue.queueAndDebounce('OP2', debounceDuration, () => f(2)),
+        ];
+
+        time.elapse(debounceDuration * 5);
+        final List<int> results = await Future.wait(operations);
+
+        expect(results, orderedEquals(<int>[1, 2]));
+      });
+    });
+  });
 }
 
 bool _notEvent(Map<String, dynamic> map) => map['event'] == null;

packages/flutter_tools/test/integration.shard/hot_reload_test.dart

@@ -36,6 +36,40 @@ void main() {
     await flutter.hotReload();
   });
 
+  test('multiple overlapping hot reload are debounced and queued', () async {
+    await _flutter.run();
+    // Capture how many *real* hot reloads occur.
+    int numReloads = 0;
+    final StreamSubscription<void> subscription = _flutter.stdout
+        .map(parseFlutterResponse)
+        .where(_isHotReloadCompletionEvent)
+        .listen((_) => numReloads++);
+
+    // To reduce tests flaking, override the debounce timer to something higher than
+    // the default to ensure the hot reloads that are supposed to arrive within the
+    // debounce period will even on slower CI machines.
+    const int hotReloadDebounceOverrideMs = 250;
+    const Duration delay = Duration(milliseconds: hotReloadDebounceOverrideMs * 2);
+
+    Future<void> doReload([void _]) =>
+        _flutter.hotReload(debounce: true, debounceDurationOverrideMs: hotReloadDebounceOverrideMs);
+
+    try {
+      await Future.wait<void>(<Future<void>>[
+        doReload(),
+        doReload(),
+        Future<void>.delayed(delay).then(doReload),
+        Future<void>.delayed(delay).then(doReload),
+      ]);
+
+      // We should only get two reloads, as the first two will have been
+      // merged together by the debounce, and the second two also.
+      expect(numReloads, equals(2));
+    } finally {
+      await subscription.cancel();
+    }
+  });
+
   test('newly added code executes during hot reload', () async {
     final StringBuffer stdout = StringBuffer();
     final StreamSubscription<String> subscription = flutter.stdout.listen(stdout.writeln);
@@ -203,3 +237,11 @@ void main() {
     await subscription.cancel();
   });
 }
+
+bool _isHotReloadCompletionEvent(Map<String, dynamic> event) {
+  return event != null &&
+      event['event'] == 'app.progress' &&
+      event['params'] != null &&
+      event['params']['progressId'] == 'hot.reload' &&
+      event['params']['finished'] == true;
+}

packages/flutter_tools/test/integration.shard/test_driver.dart

@@ -554,8 +554,9 @@ class FlutterRunTestDriver extends FlutterTestDriver {
     return prematureExitGuard.future;
   }
 
-  Future<void> hotRestart({ bool pause = false }) => _restart(fullRestart: true, pause: pause);
+  Future<void> hotRestart({ bool pause = false, bool debounce = false}) => _restart(fullRestart: true, pause: pause);
-  Future<void> hotReload() => _restart(fullRestart: false);
+  Future<void> hotReload({ bool debounce = false, int debounceDurationOverrideMs }) =>
+      _restart(fullRestart: false, debounce: debounce, debounceDurationOverrideMs: debounceDurationOverrideMs);
 
   Future<void> scheduleFrame() async {
     if (_currentRunningAppId == null) {
@@ -580,7 +581,7 @@ class FlutterRunTestDriver extends FlutterTestDriver {
     }
   }
 
-  Future<void> _restart({ bool fullRestart = false, bool pause = false }) async {
+  Future<void> _restart({ bool fullRestart = false, bool pause = false, bool debounce = false, int debounceDurationOverrideMs }) async {
     if (_currentRunningAppId == null) {
       throw Exception('App has not started yet');
     }
@@ -588,7 +589,7 @@ class FlutterRunTestDriver extends FlutterTestDriver {
     _debugPrint('Performing ${ pause ? "paused " : "" }${ fullRestart ? "hot restart" : "hot reload" }...');
     final dynamic hotReloadResponse = await _sendRequest(
       'app.restart',
-      <String, dynamic>{'appId': _currentRunningAppId, 'fullRestart': fullRestart, 'pause': pause},
+      <String, dynamic>{'appId': _currentRunningAppId, 'fullRestart': fullRestart, 'pause': pause, 'debounce': debounce, 'debounceDurationOverrideMs': debounceDurationOverrideMs},
     );
     _debugPrint('${fullRestart ? "Hot restart" : "Hot reload"} complete.');
 

packages/flutter_tools/test/src/common.dart

@@ -19,6 +19,7 @@ import 'package:flutter_tools/src/runner/flutter_command.dart';
 import 'package:flutter_tools/src/runner/flutter_command_runner.dart';
 import 'package:flutter_tools/src/globals.dart' as globals;
 import 'package:meta/meta.dart';
+import 'package:quiver/testing/async.dart';
 import 'package:test_api/test_api.dart' as test_package show TypeMatcher, test; // ignore: deprecated_member_use
 import 'package:test_api/test_api.dart' hide TypeMatcher, isInstanceOf; // ignore: deprecated_member_use
 // ignore: deprecated_member_use
@@ -211,6 +212,21 @@ void testWithoutContext(String description, FutureOr<void> body(), {
   );
 }
 
+/// Runs a callback using FakeAsync.run while continually pumping the
+/// microtask queue. This avoids a deadlock when tests `await` a Future
+/// which queues a microtask that will not be processed unless the queue
+/// is flushed.
+Future<T> runFakeAsync<T>(Future<T> Function(FakeAsync time) f) async {
+  return FakeAsync().run((FakeAsync time) async {
+    bool pump = true;
+    final Future<T> future = f(time).whenComplete(() => pump = false);
+    while (pump) {
+      time.flushMicrotasks();
+    }
+    return future;
+  }) as Future<T>;
+}
+
 /// An implementation of [AppContext] that throws if context.get is called in the test.
 ///
 /// The intention of the class is to ensure we do not accidentally regress when