Skip avoidable copies on the data packet and data stream paths - #1092
Open
tarsyang wants to merge 8 commits into
Open
Skip avoidable copies on the data packet and data stream paths#1092tarsyang wants to merge 8 commits into
tarsyang wants to merge 8 commits into
Conversation
…RTCDataBuffer.data LKRTCDataBuffer.data builds a fresh NSData from the native buffer on every read (webrtc-sdk RTCDataChannel.mm: `[NSData dataWithBytes:length:]`), and DataChannelPair read it just for `.count` three times per reliable packet: after `sendData`, when the request entered the retry buffer, and when it left it. The request now carries the encoded size captured when the buffer is built, so the payload is not copied to measure it.
`NanopbMsg.init(serializedBytes:)` copied its input into an Array before decoding. nanopb copies every string and bytes field out of the input while decoding (PB_ENABLE_MALLOC: `pb_dec_bytes`/`pb_dec_string` allocate and `pb_read` into the allocation), so the input only has to outlive the call: decode now runs on the collection's contiguous storage (`Data`, arrays, slices all provide it; `Data.withContiguousStorageIfAvailable` is always-emit-into-client) and copies only for a collection without any. `serializedData()` encoded into `[UInt8]` and then copied that into `Data`; it now encodes into `Data` directly.
`collect()`, behind `ByteStreamReader.readAll()` and `TextStreamReader.readAll()`, folded with `$0 + $1`, which copies the accumulated prefix for every chunk: the work grows with the square of the payload divided by the chunk size (15 KiB chunks: 1 MiB reassembles in under 4 ms, 16 MiB in over 100 ms, allocating a thousand times the payload). `reduce(into:)` with `append(contentsOf:)` grows one buffer.
The nanopb facades' getters copy out of the message storage on every read (`lkData` builds a new `Data`, `lkString` a new `String`). `IncomingStreamManager.handle(chunk:)` read `chunk.content` three times (empty check, length accounting, yield) and `chunk.streamID` up to three times, so every stream chunk was copied three times to deliver one; `handle(trailer:)` read `streamID` and `reason` twice. `Room.engine(_: didReceiveUserPacket:)` read `packet.payload` and `packet.topic` inside the delegate notification closures, once per room delegate and once per participant delegate. Each of these now reads the field once and reuses the value.
`String.chunks(of:)` re-sliced the remainder with `subdata(in:)` on every step, so each chunk copied everything not yet chunked (4 MiB of text took 7 ms and 138 payload-sizes of allocation, 16 MiB over 100 ms) and the result held a second copy of the text. It now returns slices of one UTF-8 encoding, as `Data.chunks(of:)` already does; the scalar-boundary rule is unchanged.
`String.byteLength` encoded the whole string into a `Data` to read its count, one full copy per RPC payload size check. `truncate(maxBytes:)` binary-searched the character count and built a prefix string plus that copy at every step. `byteLength` now reads `utf8.count`, and `truncate` walks characters until the byte budget is spent, which touches at most `maxBytes` bytes and allocates only the result. The result is unchanged: the longest prefix of characters whose UTF-8 fits.
`TranscriptionStreamReceiver.processIncoming` mapped `reader.info.attributes` to `TranscriptionAttributes` (a JSONEncoder/JSONDecoder round trip) on every chunk, although the attributes are fixed by the stream header, and an unreadable header was logged once per chunk. The handler now parses them when the stream opens and hands the result to each chunk.
tarsyang
requested review from
hiroshihorie,
pblazej and
xianshijing-lk
as code owners
August 17, 2026 18:34
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
pblazej
added a commit
that referenced
this pull request
Aug 20, 2026
Four efficiency findings, all on the send path the DC-throughput work says is Swift-overhead-bound: - Writes now carry Data until dispatch; the LKRTCDataBuffer — whose init memcpys the payload into a CopyOnWriteBuffer, and whose .data getter copies it back out on every read — is built inside the channel seam at send time. byteCount becomes payload.count (O(1)), so the meter, the retry buffer and the trim loop stop re-copying whole payloads to read a length (~3–5 MB/s of hidden copies for a busy reliable publisher), and an evicted drop-oldest write costs nothing, as the deleted DataTrackFrameSender's queue already ensured. Supersedes the DataChannelPair hunk of #1092, which fixes the same finding on main. - dispatch() and enqueue() no longer take the _state lock per write: the send target and max-message-size are mirrored into the loop's state via .attached/.configured events, FIFO-ordered with the writes they govern. The locked copy remains for isOpen and the delegate identity guard, which read from arbitrary threads. Liveness is still re-checked per write (channel.isOpen — a cheap bypass-proxy read), so a channel closing mid-drain still parks instead of failing. - makeWrites fills a reused scratch instead of returning a fresh array, so the park (reliable) hot path allocates no group array per submit. - RetryBuffer.removeAll clears by reassignment instead of dequeuing element-by-element — a teardown at the 2.5 MB retry floor no longer does per-element bookkeeping just to discard everything. The file_length lint disable is carried over from the predecessor DataChannelPair.swift, which had the same at nearly twice the length. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
pblazej
added a commit
that referenced
this pull request
Aug 26, 2026
Extracts the flow control that `DataChannelPair` and
`DataTrackFrameSender` each grew independently into one per-channel
`DataChannelDrain<Stage>`, then fixes what the extraction and a
multi-angle review surfaced.
## Architecture
Each drain owns one channel's queue, buffered-amount mirror, overflow
policy and per-kind `SendStage` — and is that channel's
`LKRTCDataChannelDelegate` (identity-guarded against replaced channels),
so nothing dispatches on channel labels anymore.
```mermaid
flowchart TB
subgraph ROOM["Room"]
NOTIFY["Room.notify(bufferStatus:of:) — single funnel"]
RD["RoomDelegate.room(_:didUpdateBufferStatus:of:)"]
subgraph PAIR["publisherDataChannel: DataChannelPair — E2EE, receive dedup, open latch"]
LOSSY["DataChannelDrain<LossyStage><br/>.dropOldest · 2 MB mark<br/>serialize only"]
REL["DataChannelDrain<ReliableStage><br/>.park · 2 MB mark<br/>sequence stamping + retry buffer"]
end
SUB["subscriberDataChannel: DataChannelPair<br/>receive-only — constructed without buffer reporting"]
subgraph DT["DataTracks"]
TRACK["DataChannelDrain<DataTrackStage><br/>.dropOldest · 8 KiB mark<br/>frames pre-packetized by Rust"]
end
end
LOSSY ---|"is delegate of"| CH1[("_lossy")]
REL ---|"is delegate of"| CH2[("_reliable")]
TRACK ---|"is delegate of"| CH3[("_data_track")]
SUB --- CH4[("_lossy / _reliable (sub)")]
LOSSY -. "isLow transitions" .-> NOTIFY
REL -. "isLow transitions" .-> NOTIFY
TRACK -. "isLow transitions" .-> NOTIFY
NOTIFY --> RD
```
Inside a drain, every mutation is serialized through one FIFO event loop
— no locks on the per-write path:
```mermaid
flowchart LR
SUBMIT["submit(input) /<br/>send(input) async"] --> LOOP
CB["delegate callbacks<br/>drained bytes (delta) · state"] -->|"isCurrent guard"| LOOP
subgraph LOOP["AsyncStream FIFO event loop — single consumer"]
PREP["SendStage.prepare<br/>reliable: stamp sequence in-loop<br/>(wire order == FIFO order)"]
GUARD["max-message-size guard"]
Q["WriteQueue<br/>.park: unbounded FIFO<br/>.dropOldest: newest group wins,<br/>evicted waiter resolved"]
METER{"BufferedAmountMeter<br/>pending ≤ low-water mark?"}
PREP --> GUARD --> Q --> METER
end
METER -->|"yes"| SEND["channel.send(Data)<br/>LKRTCDataBuffer built here —<br/>evicted writes never pay the copy"]
METER -->|"no"| PARK["wait for the next<br/>drained report"]
SEND --> RETAIN["stage.didDispatch<br/>reliable: retain for resume replay"]
```
### vs. the other SDKs
| | swift (this PR) | client-sdk-js | rust-sdks | android |
|---|---|---|---|---|
| reliable | park unbounded, FIFO, seq + replay | park (await per send)
+ replay | park + replay | direct send under lock + replay |
| lossy overflow | drop **oldest** queued, resolve sender | drop
**incoming**, return normally | n/a (no lossy drain) | park |
| data-track frames | drop oldest whole frame | drop incoming | drop
oldest whole frame | — |
| lossy threshold | fixed 2 MB | adaptive 8–256 KiB (~100 ms) | tunable
| fixed 2 MB |
| buffer status | delegate, transitions only | `DCBufferStatusChanged`
event | — | observable `bufferedAmount` |
Drop-**oldest** over js's drop-**incoming** is deliberate, not a porting
gap: js's behaviour is an artifact of having no app-level queue at all,
and it keeps *stale* data flowing while discarding the *fresh* update.
The lossy channel's typical cargo — cursor positions, presence,
game-state deltas — is supersede-style, where the newest payload makes
its predecessor worthless, so freshest-wins is the policy that matches
(and what rust-sdks' frame sender already does). Either way loss only
starts past the threshold; *which* packet dies is the only difference.
## Fixes (each with a regression test)
- Reliable sends stalled permanently after a full reconnect with >2 MB
buffered (stale mirror), and a resume replay could emit stale sequences
after the counter reset — also pinned end-to-end by a new
full-reconnect-under-load E2E.
- A replaced channel's late callbacks could re-arm the data-track
publish gate (wedging every publish) or corrupt the new channel's meter.
- Received E2EE data messages reported `encryptionType` as `.none`,
because decryption clears the field the type was read from (found by
Devin on #1075; predates it — the E2EE suite now asserts the type).
- Every drop/eviction/teardown path settles its waiter — and settlement
is now a one-shot `SendToken` (first outcome wins, an unsettled drop
fails its waiter from `deinit`), so the double-resume crash Devin caught
on the rejected-send path is unreachable, not just avoided.
## Behaviour changes
- Each drain gates on its own channel's readiness; the pre-flight
`openCompleter` still requires both.
- Lossy sends drop the oldest queued payload under sustained
backpressure instead of parking unbounded, returning normally —
rationale under the table, doc note on `publish(data:)`.
## New public API
`RoomDelegate.room(_:didUpdateBufferStatus:of:)` + `DataChannelKind`:
transition-only backpressure reporting, the analogue of js's
`DCBufferStatusChanged`.
## Notes
- Supersedes the `DataChannelPair` hunk of #1092 (same byteCount
finding, fixed on main); its other hunks don't overlap.
- AGENTS.md now documents the data-channel exception to the
`liveKitWebRTC`-queue rule (those calls are internally proxied by
libwebrtc — codifies what main already did).
- Verified against a local server across the reconnect/replay matrix,
E2EE, data-track (incl. stress), streams, RPC, room and ObjC suites;
microbenchmarks deferred (BM-DC's 1 ms quantisation can't resolve this
change).
---------
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
The data packet and data stream paths copy payloads they do not need to copy, and two of them do work quadratic in the payload. None of it changes results; each commit removes one of them:
LKRTCDataBuffer.databuilds a freshNSDatafrom the native buffer on every read (webrtc-sdkRTCDataChannel.mm), andDataChannelPairread it just for.count: aftersendData, entering the retry buffer, leaving it.PublishDataRequestnow carries the encoded size.NanopbMsg.init(serializedBytes:)copied its input into anArraybefore decoding. nanopb copies every string and bytes field out of the input while decoding, so it now decodes on the collection's contiguous storage (Data, arrays, slices) and copies only for a collection without any.serializedData()encoded into[UInt8]and copied that intoData; it now encodes intoData.collect()behindreadAll()folded with+, copying the accumulated prefix for every chunk. It appends in place.IncomingStreamManagerreadchunk.contentthree times per chunk (getters copy out of nanopb storage), so every stream chunk was copied three times to deliver one;Room.engine(_:didReceiveUserPacket:)readpacket.payloadonce per delegate. Each reads once.String.chunks(of:)re-sliced the remainder withsubdata(in:)on every step. It returns slices of one encoding, asData.chunks(of:)does; the UTF-8 boundary rule is unchanged.String.byteLengthencoded the string intoDatato count it, andtruncate(maxBytes:)built a prefix string plus that copy at every step of a binary search. They useutf8.countand one pass.TranscriptionStreamReceiverre-parsed the stream attributes (a JSON round trip) on every chunk. Once per stream.Measurements
Same test file run against
mainand this branch,-O, macOS 26 / Apple Silicon. Allocations are counted with libmalloc'smalloc_loggerhook and attributed by backtrace, so the counts are exact and deterministic; times are medians.Livekit_DataPacketfromData, 15,360 B payloadserializedData(), 15,360 BserializedData(), 61,440 BreadAll()of 256 × 15 KiB chunks (3.9 MB)readAll()of 1024 chunks (15.7 MB)String.chunks(of: 15 KiB), 4 MiBString.chunks(of: 15 KiB), 16 MiBbyteLength, 15 KiBtruncate(maxBytes: 8192), 15 KiB KoreanEnd to end against a local
livekit-server(two rooms in one process, plaintext), payload-sized allocations perpublish(data:)round trip attributed to LiveKit frames: reliable 15,360 B goes from 14.5 to 9 per packet (theLKRTCDataBuffer.datareads, theArraycopy on decode, theDatacopy on encode, and the reliable sequence-stamp append's reallocation, which stops becauseData(count:)leaves headroom). Counting every allocation of at least half the payload anywhere in the process, reliable 61,440 B goes from 16.8 to 11.0 per packet (1.07 MB to 0.70 MB). A 4 MiB byte stream: process CPU 207 ms to 157 ms, receiver allocation per chunk 5.7 MB to 0.9 MB. A 4 MiB text stream:sendText105 ms to 87 ms.Verification
LiveKitNanopbTests(conformance oracle, edge cases, fuzz, concurrency, leaks),DataStream*,StreamDataTests,StringTests,TranscriptionTests,DataChannel*,PublishDataTests,Rpc*,IncomingStreamManagerTests,OutgoingStreamManagerTests,DataTrack*pass against a locallivekit-server1.13.5.LiveKitCoreTests+LiveKitNanopbTests+LiveKitObjCTestsviaxcodebuildas in CI: pass (one timing-based test unrelated to this change,TaskObserveTests.streamFinishEndsTask, failed once in the full run and passes in isolation).