Make `publish_data` wait until the DataChannel's bufferedAmount becomes low. #545

typester · 2025-01-11T00:03:08Z

This pull request modifies publish_data to wait until the bufferedAmount of a DataChannel falls below a specified threshold before sending data, to support DataStream functionality.

The main changes are as follows:

Expose the bufferedAmount property from libwebrtc's DataChannel.
Forward the bufferedAmountChange callback of the DataChannel to an RtcEvent, allowing the rtc session to retrieve the current bufferedAmount.
Expose the ability for users to set and get the buffered_amount_low_threshold via FFI. (Default: 0)
Update publish_data to monitor the values in 2 and 3 and wait to send data until the bufferedAmount falls below the specified threshold.

Currently, the logic to wait for the buffer to become low has only been implemented for reliable DataChannels. In fact, all the features added in this update are exclusive to reliable DataChannels. Implementing this for lossy DataChannels didn’t make sense to me, but it might be worth discussing further.

ilo-nanpa · 2025-01-11T00:03:12Z

it seems like you haven't added any nanpa changeset files to this PR.

if this pull request includes changes to code, make sure to add a changeset, by writing a file to .nanpa/<unique-name>.kdl:

minor type="added" "Introduce frobnication algorithm"

refer to the manpage for more information.

…_data

bcherry · 2025-01-14T05:30:50Z

livekit/src/rtc_engine/rtc_session.rs

-        rtc_events::forward_dc_events(&mut lossy_dc, rtc_emitter.clone());
-        rtc_events::forward_dc_events(&mut reliable_dc, rtc_emitter);
+        rtc_events::forward_dc_events(&mut lossy_dc, DataPacketKind::Lossy, rtc_emitter.clone());
+        rtc_events::forward_dc_events(&mut reliable_dc, DataPacketKind::Reliable, rtc_emitter);


was this a separate bug?

No, I just added it as a new argument because the newly added Event handler needed DataPacketKind.

lukasIO · 2025-01-14T07:28:31Z

livekit/src/rtc_engine/rtc_session.rs

+            RtcEvent::DataChannelBufferedAmountChange { sent: _, amount, kind } => {
+                match kind {
+                    DataPacketKind::Lossy => {
+                        // Do nothing at this moment


I agree with your assessment that this issue isn't as pressing on the lossy channel, but one thing we'd want to avoid is that the data channel just errors out if the buffered_amount surpassing an internal maximum value. I think @theomonnom mentioned that right now it will raise an error in this case?
For that reason I think it could also make sense to include the same logic for the lossy data channel

Yes, I think we should use the same logic to avoid internal RtcError

lukasIO · 2025-01-14T07:31:15Z

livekit/src/rtc_engine/rtc_session.rs

@@ -958,6 +991,9 @@ impl SessionInner {
        kind: DataPacketKind,
    ) -> Result<(), EngineError> {
        self.ensure_publisher_connected(kind).await?;
+        if kind == DataPacketKind::Reliable {
+            self.wait_buffer_low().await?;
+        }


Is it guaranteed that successive calls to publish_data are handled in order here?
I'm imagining a scenario where a user doesn't await publish_data and just fires a bunch of them in very quick succession, and we'd want to make sure that the next self.data_channel().send(...) is still processed in the correct order.

I think there are no issues with the usage of notify, but I found one part that isn't ideal. I'll explain it in the comment below.

lukasIO · 2025-01-14T07:43:20Z

livekit/src/rtc_engine/rtc_session.rs

+        if amount <= threshold {
+            return Ok(());
+        }
+        self.reliable_dc_buffered_amount_low_notify.notified().await;


I'm not very familiar with tokio notify, so I'll ask a couple of questions that might be a bit naive:

what if wait_for_buffer_low is called exactly when amount drops under threshold? Is there a potential for a race here, where the new call will succeed before previous calls to wait_buffer_low will be notified ?

if there are multiple calls to self.reliable_dc_buffered_amount_low_notify.notified().await because the threshold is exceeded, how would subsequent callers get notified? the buffered_amount_changed has a notify_once so it will only notify one caller when the buffered amount changes. I'm thinking of a scenario where the buffer drops below the threshold in buffere_amount_changed and there would be enough headroom to forward all pending packets. We would then still wait for each packet to be sent first, because we don't receive a buffered_amount_changed event until another packet has been sent, right? I'm not sure if that is actually a problem but I think at least we'd have to be sure that buffered_amount_changed is reliably fired for every packet

I'm not sure if this is the same with your concern, but I found one part of my code that isn't ideal. Following is the content of wait_for_buffer_low. The code checks the threshold and immediately returns if there's no issue, but there's a possibility that a notify waiter already exists when this code is executed. This isn't ideal, so let me fix it.

rust-sdks/livekit/src/rtc_engine/rtc_session.rs

Lines 1008 to 1012 in 8887378

if amount <= threshold {

return Ok(());

}

self.reliable_dc_buffered_amount_low_notify.notified().await;

Ok(())

I'm thinking of a scenario where the buffer drops below the threshold in buffere_amount_changed and there would be enough headroom to forward all pending packets. We would then still wait for each packet to be sent first, because we don't receive a buffered_amount_changed event until another packet has been sent, right? I'm not sure if that is actually but I think at least we'd have to be sure that buffered_amount_changed is reliably fired for every packet

Yes, with my current code, calling publish_data multiple times could result in each call waiting for a notify. However, since DataChannel.send sends at least one buffer amount change event every time, I don't think it will end up waiting indefinitely. That said, as you pointed out, I agree that more testing is needed in this point.

lukasIO · 2025-01-14T08:06:43Z

Expose the ability for users to set and get the buffered_amount_low_threshold via FFI. (Default: 0)

It's great that it's configurable, but we might want to use a different threshold default value. This discussion here might be helpful for choosing some sane default values?

…ort Lossy kind as well

typester · 2025-01-15T19:46:05Z

I have made the following changes:

Updated the FFI functions to set and get the buffered amount low threshold, extending support to lossy data channels in addition to the already supported reliable data channels.
Created a dedicated worker task for data channels to enable a more robust message-sending mechanism.

Regarding 2, I believe that lowering latency is not an option to support lossy data channels. However, in my previous implementation, when multiple data packets were sent in a short period, a slight delay occurred even if the buffer had sufficient capacity.

To eliminate this delay, I created a new worker dedicated to data channels, which now manages the buffer by itself.

typester · 2025-01-15T19:59:41Z

For the default value for buffered amount low threshold, I am still not sure what value should be ideal. 🤔

lukasIO · 2025-01-16T08:10:08Z

For a default value, maybe just half of the max value? So 8192 ?

theomonnom · 2025-01-16T11:44:14Z

For a default value, maybe just half of the max value? So 8192 ?

Isn't the max value 65536?

theomonnom · 2025-01-16T11:46:17Z

livekit-ffi/protocol/ffi.proto

+    GetDataChannelBufferedAmountLowThresholdRequest get_data_channel_buffered_amount_low_threshold = 46;
+    SetDataChannelBufferedAmountLowThresholdRequest set_data_channel_buffered_amount_low_threshold = 47;


I don't think we should "request" the buffered amount low threshold. Since datachannels are part of the roomm, let's add it to our RoomInfo?

The issue if we do a request, then on the python side, getting this threshold is going to be an async operation but this isn't ideal.

Ideally it is just:

def buffered_amount_low_threshold(self) -> int: return self._info.buffered_amount_low_threshold

That super makes sense to me. Let me fix it

theomonnom · 2025-01-16T11:52:54Z

livekit/src/rtc_engine/rtc_session.rs

+            if let Err(_) = tx.send(result) {
+                log::error!("failed to send publish_data result");
+            }


I think we can ignore this failure? Not a big deal if the user cancel their call to publish_data

Fixed in 0aad6d1

theomonnom

Otherwise lgtm!

lukasIO · 2025-01-16T12:40:50Z

For a default value, maybe just half of the max value? So 8192 ?

Isn't the max value 65536?

that's the theoretic max value for a data packet, I think?
But I was also missing a couple of zeros in my value, meant kB.

Apparently the maximum bufferedAmount is 16MB in Chrome, so my suggestion was to use half of that as the threshold value, but we might also just do something a lot smaller like 1 or 2MB to be safe.

…t to RoomInfo

typester · 2025-01-16T21:05:00Z

Updated:

Removed Set function for buffered_amount_low_threshold from FFI, instead, added it to RoomInfo.
Added BufferedAmountLowThreasholdUpdated event so that FFI client can update room info

theomonnom · 2025-01-16T21:35:05Z

livekit-ffi/protocol/room.proto

+  required DataChannelOptions lossy_dc_options = 4;
+  required DataChannelOptions reliable_dc_options = 5;


nit since we have one field I would have flatten the dc options

Will fix. Thank you!

Fixed in 6c73254.

theomonnom

lgtm!

typester added 2 commits January 9, 2025 09:48

expose buffered_amount method to Rust

e502e30

test to implement wait_for_dc_buffer_low

7a4be4f

typester added 7 commits January 13, 2025 12:21

remove wait_for_low function, add functionality to wait it in publish…

ee64923

…_data

test FFI implementation

82d6298

add callback

fb93565

revert unused changes

cb4e34c

not necessary to make this async

3a2fb5e

update lock

6e0cd81

add nanpa changeset

8887378

typester changed the title ~~WIP: DataChannel related utility functions~~ Make publish_data wait until the DataChannel's bufferedAmount becomes low. Jan 14, 2025

typester requested review from theomonnom, bcherry and lukasIO January 14, 2025 02:05

bcherry reviewed Jan 14, 2025

View reviewed changes

typester marked this pull request as ready for review January 14, 2025 06:26

lukasIO reviewed Jan 14, 2025

View reviewed changes

typester added 2 commits January 15, 2025 10:59

create dc_task for more reliable data publishing

a69387b

change get/set dc buffered_amount_low_threshold FFI functions to supp…

d9cb564

…ort Lossy kind as well

typester added 2 commits January 15, 2025 11:47

fmt

5c7731e

add logs if buffer amount become unexpected value

ff09a80

typester requested review from lukasIO and bcherry January 15, 2025 19:59

theomonnom reviewed Jan 16, 2025

View reviewed changes

theomonnom approved these changes Jan 16, 2025

View reviewed changes

typester added 4 commits January 16, 2025 09:52

Merge remote-tracking branch 'origin/main' into typester/data-stream

5a3773d

set default threshold to 2MB

ab0bc9c

fmt

08f5f5c

ignore error here

0aad6d1

lukasIO approved these changes Jan 16, 2025

View reviewed changes

typester added 3 commits January 16, 2025 10:55

add buffered_amount_low_threshold in RoomInfo

ca69743

remove Get ffi function for dc buffered_low_threshold, instead, add i…

8e5e9d8

…t to RoomInfo

update changeset

1a42a4f

typester requested a review from theomonnom January 16, 2025 21:05

theomonnom reviewed Jan 16, 2025

View reviewed changes

theomonnom approved these changes Jan 16, 2025

View reviewed changes

typester added 3 commits January 16, 2025 13:57

flatten DataChannelOptions in protobuf

6c73254

fmt

88a623e

Merge remote-tracking branch 'origin/main' into typester/data-stream

ae7d1df

typester merged commit f162413 into main Jan 17, 2025
11 of 16 checks passed

typester deleted the typester/data-stream branch January 17, 2025 17:07

typester mentioned this pull request Feb 5, 2025

Fix webrtc crate build issue #569

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Make `publish_data` wait until the DataChannel's bufferedAmount becomes low. #545

Make `publish_data` wait until the DataChannel's bufferedAmount becomes low. #545

typester commented Jan 11, 2025 •

edited

Loading

ilo-nanpa bot commented Jan 11, 2025

bcherry Jan 14, 2025

typester Jan 14, 2025

lukasIO Jan 14, 2025

theomonnom Jan 16, 2025

lukasIO Jan 14, 2025 •

edited

Loading

typester Jan 14, 2025

lukasIO Jan 14, 2025 •

edited

Loading

typester Jan 14, 2025

lukasIO commented Jan 14, 2025

typester commented Jan 15, 2025

typester commented Jan 15, 2025

lukasIO commented Jan 16, 2025

theomonnom commented Jan 16, 2025 •

edited

Loading

theomonnom Jan 16, 2025 •

edited

Loading

typester Jan 16, 2025

theomonnom Jan 16, 2025

typester Jan 16, 2025

theomonnom left a comment •

edited

Loading

lukasIO commented Jan 16, 2025

typester commented Jan 16, 2025

theomonnom Jan 16, 2025 •

edited

Loading

typester Jan 16, 2025

typester Jan 16, 2025

theomonnom left a comment

	if amount <= threshold {
	return Ok(());
	}
	self.reliable_dc_buffered_amount_low_notify.notified().await;
	Ok(())

		GetDataChannelBufferedAmountLowThresholdRequest get_data_channel_buffered_amount_low_threshold = 46;
		SetDataChannelBufferedAmountLowThresholdRequest set_data_channel_buffered_amount_low_threshold = 47;

		required DataChannelOptions lossy_dc_options = 4;
		required DataChannelOptions reliable_dc_options = 5;

Make publish_data wait until the DataChannel's bufferedAmount becomes low. #545

Make publish_data wait until the DataChannel's bufferedAmount becomes low. #545

Conversation

typester commented Jan 11, 2025 • edited Loading

ilo-nanpa bot commented Jan 11, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukasIO Jan 14, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukasIO Jan 14, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukasIO commented Jan 14, 2025

typester commented Jan 15, 2025

typester commented Jan 15, 2025

lukasIO commented Jan 16, 2025

theomonnom commented Jan 16, 2025 • edited Loading

theomonnom Jan 16, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

theomonnom left a comment • edited Loading

Choose a reason for hiding this comment

lukasIO commented Jan 16, 2025

typester commented Jan 16, 2025

theomonnom Jan 16, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

theomonnom left a comment

Choose a reason for hiding this comment

Make `publish_data` wait until the DataChannel's bufferedAmount becomes low. #545

Make `publish_data` wait until the DataChannel's bufferedAmount becomes low. #545

typester commented Jan 11, 2025 •

edited

Loading

lukasIO Jan 14, 2025 •

edited

Loading

lukasIO Jan 14, 2025 •

edited

Loading

theomonnom commented Jan 16, 2025 •

edited

Loading

theomonnom Jan 16, 2025 •

edited

Loading

theomonnom left a comment •

edited

Loading

theomonnom Jan 16, 2025 •

edited

Loading