TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-14 14:46:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Implement single-copy optimizations for BLE/L2CAP sockets.

Browse Source 
PiperOrigin-RevId: 897745906
This commit is contained in:
Edwin Wu
2026-04-10 09:58:48 -07:00
committed by Copybara-Service
parent 26efd91ff4
commit bd66643160
2 changed files with 175 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/platform/implementation/apple/ble_l2cap_socket.mm
+81 -28
View File
@@ -14,9 +14,11 @@
#import "internal/platform/implementation/apple/ble_l2cap_socket.h"
#import "internal/platform/implementation/apple/Flags/GNCFeatureFlags.h"
#import "internal/platform/implementation/apple/Log/GNCLogger.h"
#import "internal/platform/implementation/apple/Mediums/BLE/GNCBLEL2CAPConnection.h"
#import "internal/platform/implementation/apple/utils.h"
#include "internal/platform/implementation/ble.h"
namespace nearby {
@@ -52,40 +54,81 @@ BleL2capInputStream::~BleL2capInputStream() {
}
ExceptionOr<ByteArray> BleL2capInputStream::Read(std::int64_t size) {
// Block until either (a) the connection has been closed, (b) we have enough data to return.
NSData *dataToReturn;
[condition_ lock];
while (true) {
// Check if the stream has been closed or severed.
if (!newDataPackets_) break;
if (GNCFeatureFlags.singleCopyEnabled) {
std::string dataToReturn;
bool success = false;
if (newDataPackets_.count > 0) {
// Add the packet data to the accumulated data.
for (NSData *data in newDataPackets_) {
if (data.length > 0) {
[accumulatedData_ appendData:data];
[condition_ lock];
while (true) {
// Check if the stream has been closed or severed.
if (!newDataPackets_) break;
if (newDataPackets_.count > 0) {
// Add the packet data to the accumulated data.
for (NSData *data in newDataPackets_) {
if (data.length > 0) {
[accumulatedData_ appendData:data];
}
}
[newDataPackets_ removeAllObjects];
}
[newDataPackets_ removeAllObjects];
if (accumulatedData_.length > 0) {
std::int64_t sizeToReturn =
(accumulatedData_.length < size) ? accumulatedData_.length : size;
NSRange range = NSMakeRange(0, (NSUInteger)sizeToReturn);
// Copy bytes directly into std::string, avoiding [NSData subdataWithRange:]
dataToReturn.assign((const char *)accumulatedData_.bytes, sizeToReturn);
[accumulatedData_ replaceBytesInRange:range withBytes:nil length:0];
success = true;
break;
}
[condition_ wait];
}
[condition_ unlock];
if (accumulatedData_.length > 0) {
// Return up to |size| bytes of the data.
std::int64_t sizeToReturn = (accumulatedData_.length < size) ? accumulatedData_.length : size;
NSRange range = NSMakeRange(0, (NSUInteger)sizeToReturn);
dataToReturn = [accumulatedData_ subdataWithRange:range];
[accumulatedData_ replaceBytesInRange:range withBytes:nil length:0];
break;
if (success) {
// OPTIMIZATION: Zero-copy transfer from std::string to ByteArray
return ExceptionOr<ByteArray>{ByteArray(std::move(dataToReturn))};
} else {
return ExceptionOr<ByteArray>{Exception::kIo};
}
[condition_ wait];
}
[condition_ unlock];
if (dataToReturn) {
return ExceptionOr<ByteArray>{ByteArray((const char *)dataToReturn.bytes, dataToReturn.length)};
} else {
return ExceptionOr<ByteArray>{Exception::kIo};
// Legacy Path
NSData *dataToReturn;
[condition_ lock];
while (true) {
if (!newDataPackets_) break;
if (newDataPackets_.count > 0) {
for (NSData *data in newDataPackets_) {
if (data.length > 0) {
[accumulatedData_ appendData:data];
}
}
[newDataPackets_ removeAllObjects];
}
if (accumulatedData_.length > 0) {
std::int64_t sizeToReturn =
(accumulatedData_.length < size) ? accumulatedData_.length : size;
NSRange range = NSMakeRange(0, (NSUInteger)sizeToReturn);
dataToReturn = [accumulatedData_ subdataWithRange:range];
[accumulatedData_ replaceBytesInRange:range withBytes:nil length:0];
break;
}
[condition_ wait];
}
[condition_ unlock];
if (dataToReturn) {
return ExceptionOr<ByteArray>{
ByteArray((const char *)dataToReturn.bytes, dataToReturn.length)};
} else {
return ExceptionOr<ByteArray>{Exception::kIo};
}
}
}
@@ -111,7 +154,17 @@ Exception BleL2capOutputStream::Write(absl::string_view data) {
return {Exception::kIo};
}
NSMutableData *packet = [NSMutableData dataWithBytes:data.data() length:data.size()];
NSData *packet;
if (GNCFeatureFlags.singleCopyEnabled) {
// OPTIMIZATION: Use DISPATCH_DATA_DESTRUCTOR_DEFAULT to perform a
// single copy into a GCD-managed buffer. No NSData required.
dispatch_data_t dispatchData =
dispatch_data_create(data.data(), data.size(), nil, DISPATCH_DATA_DESTRUCTOR_DEFAULT);
// dispatch_data_t is toll-free bridged to NSData
packet = (NSData *)dispatchData;
} else {
packet = [NSMutableData dataWithBytes:data.data() length:data.size()];
}
// Send the data, blocking until the completion handler is called.
__block BOOL isComplete = NO;
internal/platform/implementation/apple/ble_socket.mm
+94 -35
View File
@@ -14,13 +14,13 @@
#import "internal/platform/implementation/apple/ble_socket.h"
#include "internal/platform/implementation/ble.h"
#import "internal/platform/implementation/apple/Flags/GNCFeatureFlags.h"
#import "internal/platform/implementation/apple/Mediums/BLE/GNCMBleConnection.h"
#import "internal/platform/implementation/apple/ble_peripheral.h"
#import "internal/platform/implementation/apple/ble_utils.h"
#import "internal/platform/implementation/apple/utils.h"
#include "internal/platform/implementation/ble.h"
// TODO(b/293336684): Remove this file when shared Weave is complete.
namespace nearby {
@@ -55,45 +55,94 @@ BleInputStream::~BleInputStream() {
}
ExceptionOr<ByteArray> BleInputStream::Read(std::int64_t size) {
// Block until either (a) the connection has been closed, (b) we have enough data to return.
NSData *dataToReturn;
[condition_ lock];
while (true) {
// Check if the stream has been closed or severed.
if (!newDataPackets_) break;
if (GNCFeatureFlags.singleCopyEnabled) {
std::string dataToReturn;
bool success = false;
if (newDataPackets_.count > 0) {
// Add the packet data to the accumulated data.
for (NSData *data in newDataPackets_) {
if (data.length > 0) {
[accumulatedData_ appendData:data];
[condition_ lock];
while (true) {
// Check if the stream has been closed or severed.
if (!newDataPackets_) break;
if (newDataPackets_.count > 0) {
// Add the packet data to the accumulated data.
for (NSData *data in newDataPackets_) {
if (data.length > 0) {
[accumulatedData_ appendData:data];
}
}
[newDataPackets_ removeAllObjects];
}
[newDataPackets_ removeAllObjects];
if ((size == -1) && (accumulatedData_.length > 0)) {
// Return all of the data.
dataToReturn.assign((const char *)accumulatedData_.bytes, accumulatedData_.length);
accumulatedData_ = [NSMutableData data];
success = true;
break;
} else if (accumulatedData_.length > 0) {
// Return up to |size| bytes of the data.
std::int64_t sizeToReturn =
(accumulatedData_.length < size) ? accumulatedData_.length : size;
NSRange range = NSMakeRange(0, (NSUInteger)sizeToReturn);
// Copy bytes directly into std::string, avoiding [NSData subdataWithRange:]
dataToReturn.assign((const char *)accumulatedData_.bytes, sizeToReturn);
[accumulatedData_ replaceBytesInRange:range withBytes:nil length:0];
success = true;
break;
}
[condition_ wait];
}
[condition_ unlock];
if ((size == -1) && (accumulatedData_.length > 0)) {
// Return all of the data.
dataToReturn = accumulatedData_;
accumulatedData_ = [NSMutableData data];
break;
} else if (accumulatedData_.length > 0) {
// Return up to |size| bytes of the data.
std::int64_t sizeToReturn = (accumulatedData_.length < size) ? accumulatedData_.length : size;
NSRange range = NSMakeRange(0, (NSUInteger)sizeToReturn);
dataToReturn = [accumulatedData_ subdataWithRange:range];
[accumulatedData_ replaceBytesInRange:range withBytes:nil length:0];
break;
if (success) {
// OPTIMIZATION: Zero-copy transfer from std::string to ByteArray
return ExceptionOr<ByteArray>{ByteArray(std::move(dataToReturn))};
} else {
return ExceptionOr<ByteArray>{Exception::kIo};
}
[condition_ wait];
}
[condition_ unlock];
if (dataToReturn) {
return ExceptionOr<ByteArray>(ByteArrayFromNSData(dataToReturn));
} else {
return ExceptionOr<ByteArray>{Exception::kIo};
// Legacy path
NSData *dataToReturn;
[condition_ lock];
while (true) {
// Check if the stream has been closed or severed.
if (!newDataPackets_) break;
if (newDataPackets_.count > 0) {
for (NSData *data in newDataPackets_) {
if (data.length > 0) {
[accumulatedData_ appendData:data];
}
}
[newDataPackets_ removeAllObjects];
}
if ((size == -1) && (accumulatedData_.length > 0)) {
// Return all of the data.
dataToReturn = accumulatedData_;
accumulatedData_ = [NSMutableData data];
break;
} else if (accumulatedData_.length > 0) {
// Return up to |size| bytes of the data.
std::int64_t sizeToReturn =
(accumulatedData_.length < size) ? accumulatedData_.length : size;
NSRange range = NSMakeRange(0, (NSUInteger)sizeToReturn);
dataToReturn = [accumulatedData_ subdataWithRange:range];
[accumulatedData_ replaceBytesInRange:range withBytes:nil length:0];
break;
}
[condition_ wait];
}
[condition_ unlock];
if (dataToReturn) {
return ExceptionOr<ByteArray>(ByteArrayFromNSData(dataToReturn));
} else {
return ExceptionOr<ByteArray>{Exception::kIo};
}
}
}
@@ -119,7 +168,17 @@ Exception BleOutputStream::Write(absl::string_view data) {
return {Exception::kIo};
}
NSMutableData *packet = [NSMutableData dataWithBytes:data.data() length:data.size()];
NSData *packet;
if (GNCFeatureFlags.singleCopyEnabled) {
// OPTIMIZATION: Use DISPATCH_DATA_DESTRUCTOR_DEFAULT to perform a
// single copy into a GCD-managed buffer. No NSData required.
dispatch_data_t dispatchData =
dispatch_data_create(data.data(), data.size(), nil, DISPATCH_DATA_DESTRUCTOR_DEFAULT);
// dispatch_data_t is toll-free bridged to NSData
packet = (NSData *)dispatchData;
} else {
packet = [NSMutableData dataWithBytes:data.data() length:data.size()];
}
// Send the data, blocking until the completion handler is called.
__block bool isComplete = NO;