diff --git a/CHANGELOG.md b/CHANGELOG.md index 85d43942..0ef45114 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ * Add Descriptor Read and Write APIs * 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 * Android: complete only the oldest matching pending write on onCharacteristicWrite ## 2.1.1 diff --git a/README.md b/README.md index 9b35e63c..c8cf0883 100644 --- a/README.md +++ b/README.md @@ -602,6 +602,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 diff --git a/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift b/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift index 1a18e065..8847ca85 100644 --- a/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift +++ b/darwin/universal_ble/Sources/universal_ble/UniversalBleHelper.swift @@ -205,6 +205,13 @@ class CharacteristicWriteFuture { } } +struct PendingWriteWithoutResponse { + let deviceId: String + let characteristic: CBCharacteristic + let data: Data + let result: (Result) -> Void +} + class CharacteristicNotifyFuture { let deviceId: String let characteristicId: String diff --git a/darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift b/darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift index be29438b..f5bb3b8f 100644 --- a/darwin/universal_ble/Sources/universal_ble/UniversalBlePlugin.swift +++ b/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 descriptorReadFutures = [DescriptorReadFuture]() private var descriptorWriteFutures = [DescriptorWriteFuture]() @@ -307,6 +307,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( @@ -445,32 +454,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 + ) + ) + } } } @@ -733,27 +759,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(())) } } diff --git a/lib/src/interfaces/universal_ble_platform_interface.dart b/lib/src/interfaces/universal_ble_platform_interface.dart index c7b41d32..56176ece 100644 --- a/lib/src/interfaces/universal_ble_platform_interface.dart +++ b/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; diff --git a/lib/src/queue.dart b/lib/src/queue.dart index 1203bce7..40ea2763 100644 --- a/lib/src/queue.dart +++ b/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 _activeItems = {}; + final Map _activeItems = {}; int _lastProcessId = 0; bool _isCancelled = false; final List<_QueuedFuture> _nextCycle = []; Function(int)? onRemainingItemsUpdate; - Future add(Future Function() closure, [Duration? timeout]) { + Future add(Future Function() closure, [Duration? timeout]) => + _add(closure, timeout); + + Future addConcurrent( + Future Function() closure, [ + Duration? timeout, + ]) => _add(closure, timeout, canRunConcurrently: true); + + Future _add( + Future Function() closure, + Duration? timeout, { + bool canRunConcurrently = false, + }) { if (_isCancelled) throw Exception('Queue Cancelled'); final completer = Completer(); - _nextCycle.add(_QueuedFuture(closure, completer, timeout)); + _nextCycle.add( + _QueuedFuture( + 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 { final Future 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 execute() async { try { diff --git a/lib/src/universal_ble.dart b/lib/src/universal_ble.dart index ac419410..8fc8c0ed 100644 --- a/lib/src/universal_ble.dart +++ b/lib/src/universal_ble.dart @@ -367,7 +367,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), @@ -380,6 +383,7 @@ class UniversalBle { timeout: timeout, deviceId: deviceId, queueId: queueId, + pipelined: _platform.supportsWritePipelining, ); } diff --git a/lib/src/universal_ble_pigeon/universal_ble_pigeon_channel.dart b/lib/src/universal_ble_pigeon/universal_ble_pigeon_channel.dart index b46938f1..38f745f1 100644 --- a/lib/src/universal_ble_pigeon/universal_ble_pigeon_channel.dart +++ b/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 getBluetoothAvailabilityState() => _executeWithErrorHandling(() => _channel.getBluetoothAvailabilityState()); diff --git a/lib/src/universal_ble_web/universal_ble_web.dart b/lib/src/universal_ble_web/universal_ble_web.dart index c859014f..370d396a 100644 --- a/lib/src/universal_ble_web/universal_ble_web.dart +++ b/lib/src/universal_ble_web/universal_ble_web.dart @@ -21,6 +21,11 @@ class UniversalBleWeb extends UniversalBlePlatform { final Map> _serviceCache = {}; bool _isScanning = false; + @override + bool get supportsWritePipelining => + defaultTargetPlatform == TargetPlatform.iOS || + defaultTargetPlatform == TargetPlatform.macOS; + @override Future getConnectionState(String deviceId) async { BluetoothDevice? device = _getDeviceById(deviceId); diff --git a/lib/src/utils/ble_command_queue.dart b/lib/src/utils/ble_command_queue.dart index 299d620f..15fa51ac 100644 --- a/lib/src/utils/ble_command_queue.dart +++ b/lib/src/utils/ble_command_queue.dart @@ -16,20 +16,56 @@ class BleCommandQueue { String? deviceId, Duration? timeout, String? queueId, + }) => _queueCommand( + command, + deviceId: deviceId, + timeout: timeout, + queueId: queueId, + ); + + Future queueWrite( + Future Function() command, { + String? deviceId, + Duration? timeout, + String? queueId, + bool pipelined = false, + }) => _queueCommand( + command, + deviceId: deviceId, + timeout: timeout, + queueId: queueId, + canRunConcurrently: pipelined, + ); + + Future _queueCommand( + Future Function() command, { + String? deviceId, + Duration? timeout, + String? queueId, + bool canRunConcurrently = false, }) { Duration? timeoutDuration = timeout ?? this.timeout; if (timeoutDuration == null) { - return queueCommandWithoutTimeout( + return _queueCommandWithoutTimeout( command, deviceId: deviceId, queueId: queueId, + canRunConcurrently: canRunConcurrently, ); } return switch (queueType) { - QueueType.global => _queue(queueId).add(command, timeoutDuration), - QueueType.perDevice => _queue( - queueId ?? deviceId, - ).add(command, timeoutDuration), + QueueType.global => _add( + _queue(queueId), + command, + timeoutDuration, + canRunConcurrently, + ), + QueueType.perDevice => _add( + _queue(queueId ?? deviceId), + command, + timeoutDuration, + canRunConcurrently, + ), QueueType.none => command().timeout(timeoutDuration), }; } @@ -38,14 +74,44 @@ class BleCommandQueue { Future Function() command, { String? deviceId, String? queueId, + }) => _queueCommandWithoutTimeout( + command, + deviceId: deviceId, + queueId: queueId, + ); + + Future _queueCommandWithoutTimeout( + Future Function() command, { + String? deviceId, + String? queueId, + bool canRunConcurrently = false, }) { return switch (queueType) { - QueueType.global => _queue(queueId).add(command), - QueueType.perDevice => _queue(queueId ?? deviceId).add(command), + QueueType.global => _add( + _queue(queueId), + command, + null, + canRunConcurrently, + ), + QueueType.perDevice => _add( + _queue(queueId ?? deviceId), + command, + null, + canRunConcurrently, + ), QueueType.none => command(), }; } + Future _add( + Queue queue, + Future Function() command, + Duration? timeout, + bool canRunConcurrently, + ) => canRunConcurrently + ? queue.addConcurrent(command, timeout) + : queue.add(command, timeout); + Queue _queue(String? id) { final queueKey = id ?? globalQueueId; return _queueMap[queueKey] ?? _newQueue(queueKey); diff --git a/test/universal_ble_test_mock.dart b/test/universal_ble_test_mock.dart index 1e5cb56c..43424cc7 100644 --- a/test/universal_ble_test_mock.dart +++ b/test/universal_ble_test_mock.dart @@ -75,6 +75,11 @@ abstract class UniversalBlePlatformMock extends UniversalBlePlatform { throw UnimplementedError(); } + @override + Future readRssi(String deviceId) { + throw UnimplementedError(); + } + @override Future requestConnectionPriority( String deviceId, diff --git a/test/universal_ble_write_queue_test.dart b/test/universal_ble_write_queue_test.dart new file mode 100644 index 00000000..f91774f4 --- /dev/null +++ b/test/universal_ble_write_queue_test.dart @@ -0,0 +1,110 @@ +import 'dart:async'; +import 'dart:typed_data'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:universal_ble/universal_ble.dart'; + +import 'universal_ble_test_mock.dart'; + +void main() { + setUp(() { + UniversalBle.clearQueue(); + UniversalBle.queueType = QueueType.global; + }); + + tearDown(() { + UniversalBle.clearQueue(); + }); + + test('pipelines writes when supported by the platform', () async { + final platform = _PendingWritePlatform(supportsWritePipelining: true); + UniversalBle.setInstance(platform); + + final first = _write(1); + final second = _write(2); + + expect(platform.started, [1, 2]); + + platform.pending[1].complete(); + platform.pending[0].complete(); + await first; + await second; + }); + + test('serializes writes when pipelining is unsupported', () async { + final platform = _PendingWritePlatform(); + UniversalBle.setInstance(platform); + + final first = _write(1); + final second = _write(2); + + expect(platform.started, [1]); + + platform.pending.single.complete(); + await first; + await pumpEventQueue(); + expect(platform.started, [1, 2]); + + platform.pending[1].complete(); + await second; + }); + + test('does not pipeline a write across a queue barrier', () async { + final platform = _PendingWritePlatform(supportsWritePipelining: true); + UniversalBle.setInstance(platform); + + final read = UniversalBle.read('device', '180a', '202a'); + final write = _write(1); + + expect(platform.reads, 1); + expect(platform.started, isEmpty); + + platform.readPending.complete(Uint8List(0)); + await read; + await pumpEventQueue(); + expect(platform.started, [1]); + + platform.pending.single.complete(); + await write; + }); +} + +Future _write(int value) => + UniversalBle.write('device', '180a', '202a', Uint8List.fromList([value])); + +class _PendingWritePlatform extends UniversalBlePlatformMock { + @override + final bool supportsWritePipelining; + + _PendingWritePlatform({this.supportsWritePipelining = false}); + + final started = []; + final pending = >[]; + final readPending = Completer(); + var reads = 0; + + @override + Future readValue( + String deviceId, + String service, + String characteristic, { + Duration? timeout, + }) { + reads++; + return readPending.future; + } + + @override + Future writeValue( + String deviceId, + String service, + String characteristic, + Uint8List value, + BleOutputProperty bleOutputProperty, + ) { + started.add(value.single); + final completion = Completer(); + pending.add(completion); + return completion.future; + } +}