Fix Apple queued write throughput (#271)

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## Unreleased
 * Windows: Harden BLE connection lifetime, asynchronous callbacks, and notification subscription handling
 * Android: instantly close the GATT client on disconnect
+* Apple: Prevent queued writes from degrading throughput by pipelining them through the native GATT queue
 
 ## 2.1.1
 * Android: Fix BluetoothDevice null-safety compile error under Kotlin 2.x

README.md

@@ -592,6 +592,8 @@ UniversalBle.queueType = QueueType.none;
 
 Keep in mind that some platforms (e.g. Android) may not handle well devices that fail to process consecutive commands without a minimum interval. Therefore, it is not advised to set `queueType` to `none`.
 
+On iOS, macOS, and Chrome running on Apple platforms, consecutive writes are submitted to the native GATT queue ahead of completion so the platform can maintain throughput. Later queued non-write commands still wait for all preceding writes.
+
 You can get queue updates by setting:
 
 ```dart

darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift

@@ -197,6 +197,13 @@ class CharacteristicWriteFuture {
     }
 }
 
+struct PendingWriteWithoutResponse {
+    let deviceId: String
+    let characteristic: CBCharacteristic
+    let data: Data
+    let result: (Result<Void, Error>) -> Void
+}
+
 class CharacteristicNotifyFuture {
     let deviceId: String
     let characteristicId: String

darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift

@@ -95,7 +95,7 @@ private class BleCentralDarwin: NSObject, UniversalBlePlatformChannel, CBCentral
   private var activeServiceDiscoveries: [String: UniversalBleAsyncServiceDiscovery] = [:]
   private var characteristicReadFutures = [CharacteristicReadFuture]()
   private var characteristicWriteFutures = [CharacteristicWriteFuture]()
-  private var characteristicWriteWithoutResponseFutures = [CharacteristicWriteFuture]()
+  private var pendingWriteWithoutResponse = [PendingWriteWithoutResponse]()
   private var characteristicNotifyFutures = [CharacteristicNotifyFuture]()
   private var discoverServicesFutures = [DiscoverServicesFuture]()
   private var rssiReadFutures = [RssiReadFuture]()
@@ -305,6 +305,15 @@ private class BleCentralDarwin: NSObject, UniversalBlePlatformChannel, CBCentral
       }
       return false
     }
+    pendingWriteWithoutResponse.removeAll { pending in
+      if pending.deviceId == deviceId {
+        pending.result(
+          Result.failure(createFlutterError(code: .deviceDisconnected, message: "Device Disconnected"))
+        )
+        return true
+      }
+      return false
+    }
     characteristicNotifyFutures.removeAll { future in
       if future.deviceId == deviceId {
         future.result(
@@ -425,32 +434,49 @@ private class BleCentralDarwin: NSObject, UniversalBlePlatformChannel, CBCentral
       completion(Result.failure(createFlutterError(code: .deviceNotFound, message: "Unknown deviceId:\(self)")))
       return
     }
+    guard peripheral.state == .connected else {
+      completion(Result.failure(createFlutterError(code: .deviceDisconnected, message: "Device Disconnected")))
+      return
+    }
     guard let gattCharacteristic = peripheral.getCharacteristic(characteristic, of: service) else {
       completion(Result.failure(createFlutterError(code: .characteristicNotFound, message: "Unknown characteristic:\(characteristic)")))
       return
     }
 
-    let type = bleOutputProperty == .withoutResponse ? CBCharacteristicWriteType.withoutResponse : CBCharacteristicWriteType.withResponse
-
-    if type == CBCharacteristicWriteType.withResponse {
-      if !gattCharacteristic.properties.contains(.write) {
+    switch bleOutputProperty {
+    case .withResponse:
+      guard gattCharacteristic.properties.contains(.write) else {
         completion(Result.failure(createFlutterError(code: .characteristicDoesNotSupportWrite, message: "Characteristic does not support write withResponse")))
         return
       }
-    } else if type == CBCharacteristicWriteType.withoutResponse {
-      if !gattCharacteristic.properties.contains(.writeWithoutResponse) {
+      characteristicWriteFutures.append(
+        CharacteristicWriteFuture(
+          deviceId: deviceId,
+          characteristicId: gattCharacteristic.uuid.uuidStr,
+          serviceId: gattCharacteristic.service?.uuid.uuidStr,
+          result: completion
+        )
+      )
+      peripheral.writeValue(value.data, for: gattCharacteristic, type: .withResponse)
+    case .withoutResponse:
+      guard gattCharacteristic.properties.contains(.writeWithoutResponse) else {
         completion(Result.failure(createFlutterError(code: .characteristicDoesNotSupportWriteWithoutResponse, message: "Characteristic does not support write withoutResponse")))
         return
       }
-    }
-    peripheral.writeValue(value.data, for: gattCharacteristic, type: type)
 
-    // Wait for future response
-    let future = CharacteristicWriteFuture(deviceId: deviceId, characteristicId: gattCharacteristic.uuid.uuidStr, serviceId: gattCharacteristic.service?.uuid.uuidStr, result: completion)
-    if type == CBCharacteristicWriteType.withResponse {
-      characteristicWriteFutures.append(future)
-    } else {
-      characteristicWriteWithoutResponseFutures.append(future)
+      if peripheral.canSendWriteWithoutResponse {
+        peripheral.writeValue(value.data, for: gattCharacteristic, type: .withoutResponse)
+        completion(Result.success(()))
+      } else {
+        pendingWriteWithoutResponse.append(
+          PendingWriteWithoutResponse(
+            deviceId: deviceId,
+            characteristic: gattCharacteristic,
+            data: value.data,
+            result: completion
+          )
+        )
+      }
     }
   }
 
@@ -652,27 +678,32 @@ private class BleCentralDarwin: NSObject, UniversalBlePlatformChannel, CBCentral
   }
 
   public func peripheralIsReady(toSendWriteWithoutResponse peripheral: CBPeripheral) {
-    characteristicWriteWithoutResponseFutures.removeAll { future in
-      if future.deviceId == peripheral.uuid.uuidString {
-        future.result(Result.success({}()))
-        return true
+    let deviceId = peripheral.uuid.uuidString
+    while peripheral.state == .connected && peripheral.canSendWriteWithoutResponse {
+      guard let index = pendingWriteWithoutResponse.firstIndex(where: { $0.deviceId == deviceId }) else {
+        return
       }
-      return false
+      let pending = pendingWriteWithoutResponse.remove(at: index)
+      peripheral.writeValue(pending.data, for: pending.characteristic, type: .withoutResponse)
+      pending.result(Result.success(()))
     }
   }
 
   public func peripheral(_ peripheral: CBPeripheral, didWriteValueFor characteristic: CBCharacteristic, error: Error?) {
-    characteristicWriteFutures.removeAll { future in
-      if future.deviceId == peripheral.uuid.uuidString && future.characteristicId == characteristic.uuid.uuidStr && future.serviceId == characteristic.service?.uuid.uuidStr {
-        if let flutterError = error?.toFlutterError() {
-          UniversalBleLogger.shared.logError("WRITE_FAILED <- \(peripheral.uuid.uuidString) \(characteristic.uuid.uuidStr): \(flutterError.message ?? "")")
-          future.result(Result.failure(flutterError))
-        } else {
-          future.result(Result.success({}()))
-        }
-        return true
-      }
-      return false
+    guard let index = characteristicWriteFutures.firstIndex(where: {
+      $0.deviceId == peripheral.uuid.uuidString &&
+        $0.characteristicId == characteristic.uuid.uuidStr &&
+        $0.serviceId == characteristic.service?.uuid.uuidStr
+    }) else {
+      return
+    }
+
+    let future = characteristicWriteFutures.remove(at: index)
+    if let flutterError = error?.toFlutterError() {
+      UniversalBleLogger.shared.logError("WRITE_FAILED <- \(peripheral.uuid.uuidString) \(characteristic.uuid.uuidStr): \(flutterError.message ?? "")")
+      future.result(Result.failure(flutterError))
+    } else {
+      future.result(Result.success(()))
     }
   }
 

lib/src/interfaces/universal_ble_platform_interface.dart

@@ -6,6 +6,9 @@ import 'package:universal_ble/src/utils/universal_logger.dart';
 import 'package:universal_ble/universal_ble.dart';
 
 abstract class UniversalBlePlatform {
+  /// Whether consecutive writes may be submitted before earlier writes finish.
+  bool get supportsWritePipelining => false;
+
   // Do not use these directly to push updates
   OnScanResult? onScanResultUpdate;
   OnConnectionChange? onConnectionChange;

lib/src/queue.dart

@@ -3,20 +3,40 @@ import 'dart:async';
 /// Original Author: Ryan Knell (https://github.com/rknell/dart_queue)
 
 /// Queue to execute Futures in order.
-/// It awaits each future before executing the next one.
+/// It awaits each future before executing the next one unless consecutive
+/// commands opt into [addConcurrent].
 class Queue {
-  final Set<int> _activeItems = {};
+  final Map<int, bool> _activeItems = {};
   int _lastProcessId = 0;
   bool _isCancelled = false;
   final List<_QueuedFuture> _nextCycle = [];
   Function(int)? onRemainingItemsUpdate;
 
-  Future<T> add<T>(Future<T> Function() closure, [Duration? timeout]) {
+  Future<T> add<T>(Future<T> Function() closure, [Duration? timeout]) =>
+      _add(closure, timeout);
+
+  Future<T> addConcurrent<T>(
+    Future<T> Function() closure, [
+    Duration? timeout,
+  ]) => _add(closure, timeout, canRunConcurrently: true);
+
+  Future<T> _add<T>(
+    Future<T> Function() closure,
+    Duration? timeout, {
+    bool canRunConcurrently = false,
+  }) {
     if (_isCancelled) throw Exception('Queue Cancelled');
     final completer = Completer<T>();
-    _nextCycle.add(_QueuedFuture<T>(closure, completer, timeout));
+    _nextCycle.add(
+      _QueuedFuture<T>(
+        closure,
+        completer,
+        timeout,
+        canRunConcurrently: canRunConcurrently,
+      ),
+    );
     _updateRemainingItems();
-    if (_activeItems.isEmpty) _queueUpNext();
+    _queueUpNext();
     return completer.future;
   }
 
@@ -29,12 +49,11 @@ class Queue {
   }
 
   void _queueUpNext() {
-    if (_nextCycle.isNotEmpty && !_isCancelled && _activeItems.length <= 1) {
+    while (_nextCycle.isNotEmpty && !_isCancelled && _canRunNext()) {
       final processId = _lastProcessId;
-      _activeItems.add(processId);
-      final item = _nextCycle.first;
+      final item = _nextCycle.removeAt(0);
+      _activeItems[processId] = item.canRunConcurrently;
       _lastProcessId++;
-      _nextCycle.remove(item);
       item.onComplete = () async {
         _activeItems.remove(processId);
         _updateRemainingItems();
@@ -44,6 +63,12 @@ class Queue {
     }
   }
 
+  bool _canRunNext() {
+    if (_activeItems.isEmpty) return true;
+    return _nextCycle.first.canRunConcurrently &&
+        _activeItems.values.every((canRunConcurrently) => canRunConcurrently);
+  }
+
   void _updateRemainingItems() {
     int remainingQueueItems = _nextCycle.length + _activeItems.length;
     onRemainingItemsUpdate?.call(remainingQueueItems);
@@ -55,8 +80,15 @@ class _QueuedFuture<T> {
   final Future<T> Function() closure;
   Function? onComplete;
   final Duration? timeout;
+  final bool canRunConcurrently;
 
-  _QueuedFuture(this.closure, this.completer, this.timeout, {this.onComplete});
+  _QueuedFuture(
+    this.closure,
+    this.completer,
+    this.timeout, {
+    this.onComplete,
+    this.canRunConcurrently = false,
+  });
 
   Future<void> execute() async {
     try {

lib/src/universal_ble.dart

@@ -344,7 +344,10 @@ class UniversalBle {
     Duration? timeout,
     String? queueId,
   }) async {
-    await _bleCommandQueue.queueCommand(
+    // CoreBluetooth (and Chromium on Apple platforms) owns a native GATT
+    // operation queue. Let consecutive writes fill it without letting either
+    // writes or other operations cross a queue barrier.
+    await _bleCommandQueue.queueWrite(
       () => _platform.writeValue(
         deviceId,
         BleUuidParser.string(service),
@@ -357,6 +360,7 @@ class UniversalBle {
       timeout: timeout,
       deviceId: deviceId,
       queueId: queueId,
+      pipelined: _platform.supportsWritePipelining,
     );
   }
 

lib/src/universal_ble_pigeon/universal_ble_pigeon_channel.dart

@@ -16,6 +16,11 @@ class UniversalBlePigeonChannel extends UniversalBlePlatform
 
   final _channel = UniversalBlePlatformChannel();
 
+  @override
+  bool get supportsWritePipelining =>
+      defaultTargetPlatform == TargetPlatform.iOS ||
+      defaultTargetPlatform == TargetPlatform.macOS;
+
   @override
   Future<AvailabilityState> getBluetoothAvailabilityState() =>
       _executeWithErrorHandling(() => _channel.getBluetoothAvailabilityState());

lib/src/universal_ble_web/universal_ble_web.dart

@@ -21,6 +21,11 @@ class UniversalBleWeb extends UniversalBlePlatform {
   final Map<String, List<_UniversalWebBluetoothService>> _serviceCache = {};
   bool _isScanning = false;
 
+  @override
+  bool get supportsWritePipelining =>
+      defaultTargetPlatform == TargetPlatform.iOS ||
+      defaultTargetPlatform == TargetPlatform.macOS;
+
   @override
   Future<BleConnectionState> getConnectionState(String deviceId) async {
     BluetoothDevice? device = _getDeviceById(deviceId);