⚡️ Deallocate operations and requests for cancellation (#2252)

Resolve #2111

Signed-off-by: Alex Li <github@alexv525.com>

.github/workflows/tests.yml

@@ -39,6 +39,8 @@ jobs:
           cache: true
           flutter-version: ${{ matrix.sdk == 'min' && '3.3.0' || '' }}
           channel: ${{ matrix.sdk == 'min' && '' || matrix.channel }}
+      - run: |
+          echo TARGET_DART_SDK=${{ matrix.sdk }} >> $GITHUB_ENV
       - name: Prepare dependencies for the project management
         run: dart pub get
       - uses: bluefireteam/melos-action@v3

dio/CHANGELOG.md

@@ -17,6 +17,7 @@ See the [Migration Guide][] for the complete breaking changes list.**
   throws the exception directly rather than cut actual HTTP requests afterward.
 - Catch `MediaType` parse exception in `Transformer.isJsonMimeType`.
 - Improves warning logs on the Web platform.
+- Improves memory allocating when using `CancelToken`.
 
 ## 5.4.3+1
 

dio/dart_test.yaml

@@ -1,6 +1,14 @@
 file_reporters:
   json: build/reports/test-results.json
 
+presets:
+  # empty placeholders required in CI scripts
+  all:
+  default:
+  min:
+  stable:
+  beta:
+
 tags:
   tls:
     skip: "Skipping TLS test with specific setup requirements by default. Use '-P all' to run all tests."
@@ -9,6 +17,7 @@ tags:
         skip: false
       default:
         skip: true
+  gc: # We have that tag, but we are not skipping in any preset.
 
 override_platforms:
   chrome:

dio/lib/src/adapters/io_adapter.dart

@@ -2,8 +2,6 @@ import 'dart:async';
 import 'dart:io';
 import 'dart:typed_data';
 
-import 'package:async/async.dart';
-
 import '../adapter.dart';
 import '../dio_exception.dart';
 import '../options.dart';
@@ -70,15 +68,7 @@ class IOHttpClientAdapter implements HttpClientAdapter {
         "Can't establish connection after the adapter was closed.",
       );
     }
-    final operation = CancelableOperation.fromFuture(
-      _fetch(
-        options,
-        requestStream,
-        cancelFuture,
-      ),
-    );
-    cancelFuture?.whenComplete(() => operation.cancel());
-    return operation.value;
+    return _fetch(options, requestStream, cancelFuture);
   }
 
   Future<ResponseBody> _fetch(
@@ -88,7 +78,6 @@ class IOHttpClientAdapter implements HttpClientAdapter {
   ) async {
     final httpClient = _configHttpClient(options.connectTimeout);
     final reqFuture = httpClient.openUrl(options.method, options.uri);
-
     late HttpClientRequest request;
     try {
       final connectionTimeout = options.connectTimeout;
@@ -106,7 +95,10 @@ class IOHttpClientAdapter implements HttpClientAdapter {
         request = await reqFuture;
       }
 
-      cancelFuture?.whenComplete(() => request.abort());
+      final requestWR = WeakReference<HttpClientRequest>(request);
+      cancelFuture?.whenComplete(() {
+        requestWR.target?.abort();
+      });
 
       // Set Headers
       options.headers.forEach((key, value) {

dio/lib/src/dio_mixin.dart

@@ -4,6 +4,7 @@ import 'dart:convert';
 import 'dart:math' as math;
 import 'dart:typed_data';
 
+import 'package:async/async.dart';
 import 'package:meta/meta.dart';
 
 import 'adapter.dart';
@@ -524,11 +525,18 @@ abstract class DioMixin implements Dio {
     final cancelToken = reqOpt.cancelToken;
     try {
       final stream = await _transformData(reqOpt);
-      final responseBody = await httpClientAdapter.fetch(
-        reqOpt,
-        stream,
-        cancelToken?.whenCancel,
+      final operation = CancelableOperation.fromFuture(
+        httpClientAdapter.fetch(
+          reqOpt,
+          stream,
+          cancelToken?.whenCancel,
+        ),
       );
+      final operationWeakReference = WeakReference(operation);
+      cancelToken?.whenCancel.whenComplete(() {
+        operationWeakReference.target?.cancel();
+      });
+      final responseBody = await operation.value;
       final headers = Headers.fromMap(
         responseBody.headers,
         preserveHeaderCase: reqOpt.preserveHeaderCase,

dio/test/cancel_token_test.dart

@@ -1,3 +1,5 @@
+import 'dart:typed_data' show Uint8List;
+
 import 'package:dio/dio.dart';
 import 'package:dio/src/adapters/io_adapter.dart';
 import 'package:dio_test/util.dart';
@@ -112,4 +114,72 @@ void main() {
       expect(walkThroughHandlers, isFalse);
     });
   });
+
+  test(
+    'deallocates HttpClientRequest',
+    () async {
+      final client = MockHttpClient();
+      final dio = Dio();
+      dio.httpClientAdapter = IOHttpClientAdapter(
+        createHttpClient: () => client,
+      );
+      final token = CancelToken();
+      final requests = <MockHttpClientRequest>{};
+      final requestsReferences = <WeakReference<MockHttpClientRequest>>{};
+      when(client.openUrl(any, any)).thenAnswer((_) async {
+        final request = MockHttpClientRequest();
+        requests.add(request);
+        requestsReferences.add(WeakReference(request));
+        when(request.close()).thenAnswer((_) async {
+          final response = MockHttpClientResponse();
+          when(response.headers).thenReturn(MockHttpHeaders());
+          when(response.statusCode).thenReturn(200);
+          when(response.reasonPhrase).thenReturn('OK');
+          when(response.isRedirect).thenReturn(false);
+          when(response.redirects).thenReturn([]);
+          when(response.cast())
+              .thenAnswer((_) => const Stream<Uint8List>.empty());
+          await Future.delayed(const Duration(milliseconds: 200));
+          return response;
+        });
+        when(request.abort()).thenAnswer((realInvocation) {
+          requests.remove(request);
+        });
+        return request;
+      });
+
+      final futures = [
+        dio.get('https://does.not.exists', cancelToken: token),
+        dio.get('https://does.not.exists', cancelToken: token),
+      ];
+      for (final future in futures) {
+        expectLater(
+          future,
+          throwsDioException(DioExceptionType.cancel),
+        );
+      }
+
+      // Opening requests.
+      await Future.delayed(const Duration(milliseconds: 100));
+      token.cancel();
+      // Aborting requests.
+      await Future.delayed(const Duration(seconds: 1));
+      expect(requests, isEmpty);
+
+      try {
+        await Future.wait(futures);
+      } catch (_) {
+        // Waiting here until all futures are completed.
+      }
+      expect(requests, isEmpty);
+      expect(requestsReferences, hasLength(2));
+
+      // GC.
+      produceGarbage();
+      await Future.delayed(const Duration(seconds: 1));
+      expect(requestsReferences.every((e) => e.target == null), isTrue);
+    },
+    tags: ['gc'],
+    testOn: 'vm',
+  );
 }

dio_test/lib/src/utils.dart

@@ -3,3 +3,21 @@ const kIsWeb = bool.hasEnvironment('dart.library.js_util')
     : identical(0, 0.0);
 
 const nonRoutableUrl = 'http://10.0.0.0';
+
+/// https://github.com/dart-lang/sdk/blob/59add4f01ef4741e10f64db9c2c8655cfe738ccb/tests/corelib/finalizer_test.dart#L86-L101
+void produceGarbage() {
+  const approximateWordSize = 4;
+
+  List<dynamic> sink = [];
+  for (int i = 0; i < 500; ++i) {
+    final filler = i % 2 == 0 ? 1 : sink;
+    if (i % 250 == 1) {
+      // 2 x 25 MB in old space.
+      sink = List.filled(25 * 1024 * 1024 ~/ approximateWordSize, filler);
+    } else {
+      // 498 x 50 KB in new space
+      sink = List.filled(50 * 1024 ~/ approximateWordSize, filler);
+    }
+  }
+  print(sink.hashCode); // Ensure there's real use of the allocation.
+}

melos.yaml

@@ -57,7 +57,12 @@ scripts:
       melos run test:flutter
   test:vm:
     name: Dart VM tests
-    exec: dart test --preset ${TEST_PRESET:-default} --coverage coverage/vm --chain-stack-traces
+    exec: |
+      if [ "$TARGET_DART_SDK" = "min" ]; then
+        dart test --preset=${TEST_PRESET:-default},${TARGET_DART_SDK:-stable} --chain-stack-traces
+      else
+        dart test --preset=${TEST_PRESET:-default},${TARGET_DART_SDK:-stable} --coverage=coverage/vm --chain-stack-traces
+      fi
     packageFilters:
       flutter: false
       dirExists: test
@@ -81,7 +86,12 @@ scripts:
       TEST_PLATFORM: firefox
   test:web:single:
     name: Dart Web tests in a browser
-    exec: dart test --platform ${TEST_PLATFORM} --coverage coverage/${TEST_PLATFORM} --preset ${TEST_PRESET:-default} --chain-stack-traces
+    exec: |
+      if [ "$TARGET_DART_SDK" = "min" ]; then
+        dart test --platform ${TEST_PLATFORM} --preset=${TEST_PRESET:-default},${TARGET_DART_SDK:-stable} --chain-stack-traces
+      else
+        dart test --platform ${TEST_PLATFORM} --coverage=coverage/${TEST_PLATFORM} --preset=${TEST_PRESET:-default},${TARGET_DART_SDK:-stable} --chain-stack-traces
+      fi
     packageFilters:
       flutter: false
       dirExists: test

plugins/compatibility_layer/dart_test.yaml

@@ -2,6 +2,9 @@ presets:
   # empty placeholders required in CI scripts
   all:
   default:
+  min:
+  stable:
+  beta:
 
 override_platforms:
   chrome:

plugins/cookie_manager/dart_test.yaml

@@ -5,3 +5,6 @@ presets:
   # empty placeholders required in CI scripts
   all:
   default:
+  min:
+  stable:
+  beta:

plugins/http2_adapter/CHANGELOG.md

@@ -2,7 +2,7 @@
 
 ## Unreleased
 
-*None.*
+- Improves memory allocating when using `CancelToken`.
 
 ## 2.5.2
 

plugins/http2_adapter/dart_test.yaml

@@ -1,6 +1,14 @@
 file_reporters:
   json: build/reports/test-results.json
 
+presets:
+  # empty placeholders required in CI scripts
+  all:
+  default:
+  min:
+  stable:
+  beta:
+
 tags:
   tls:
     skip: "Skipping TLS test with specific setup requirements by default. Use '-P all' to run all tests."

plugins/http2_adapter/lib/src/http2_adapter.dart

@@ -45,11 +45,21 @@ class Http2Adapter implements HttpClientAdapter {
     RequestOptions options,
     Stream<Uint8List>? requestStream,
     Future<void>? cancelFuture,
-  ) async {
+  ) {
     // Recursive fetching.
     final redirects = <RedirectRecord>[];
+    return _fetch(options, requestStream, cancelFuture, redirects);
+  }
+
+  Future<ResponseBody> _fetch(
+    RequestOptions options,
+    Stream<Uint8List>? requestStream,
+    Future<void>? cancelFuture,
+    List<RedirectRecord> redirects,
+  ) async {
+    late final ClientTransportConnection transport;
     try {
-      return await _fetch(options, requestStream, cancelFuture, redirects);
+      transport = await connectionManager.getConnection(options);
     } on DioH2NotSupportedException catch (e) {
       // Fallback to use the callback
       // or to another adapter (typically IOHttpClientAdapter)
@@ -79,15 +89,7 @@ class Http2Adapter implements HttpClientAdapter {
         error: e,
       );
     }
-  }
 
-  Future<ResponseBody> _fetch(
-    RequestOptions options,
-    Stream<Uint8List>? requestStream,
-    Future<void>? cancelFuture,
-    List<RedirectRecord> redirects,
-  ) async {
-    final transport = await connectionManager.getConnection(options);
     final uri = options.uri;
     String path = uri.path;
     const excludeMethods = ['PUT', 'POST', 'PATCH'];
@@ -122,11 +124,12 @@ class Http2Adapter implements HttpClientAdapter {
 
     // Creates a new outgoing stream.
     final stream = transport.makeRequest(headers);
+    final streamWR = WeakReference<ClientTransportStream>(stream);
 
     final hasRequestData = requestStream != null;
-    if (hasRequestData) {
-      cancelFuture?.whenComplete(() {
-        stream.outgoingMessages.close();
+    if (hasRequestData && cancelFuture != null) {
+      cancelFuture.whenComplete(() {
+        streamWR.target?.outgoingMessages.close();
       });
     }
 
@@ -277,7 +280,7 @@ class Http2Adapter implements HttpClientAdapter {
       onClose: () {
         responseSubscription.cancel();
         responseSink.close();
-        stream.outgoingMessages.close();
+        streamWR.target?.outgoingMessages.close();
       },
     );
   }

plugins/web_adapter/dart_test.yaml

@@ -2,6 +2,9 @@ presets:
   # empty placeholders required in CI scripts
   all:
   default:
+  min:
+  stable:
+  beta:
 
 override_platforms:
   chrome: