Control latency of lossy data channel (#1754)

Co-authored-by: lukasIO <mail@lukasseiler.de>

Author: cnderrauber

.changeset/thick-ghosts-chew.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+Control latency of lossy data channel

src/room/RTCEngine.ts

@@ -92,6 +92,8 @@ const reliableDataChannel = '_reliable';
 const minReconnectWait = 2 * 1000;
 const leaveReconnect = 'leave-reconnect';
 const reliabeReceiveStateTTL = 30_000;
+const lossyDataChannelBufferThresholdMin = 8 * 1024;
+const lossyDataChannelBufferThresholdMax = 256 * 1024;
 
 enum PCState {
   New,
@@ -203,6 +205,14 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
 
   private reliableReceivedState: TTLMap<string, number> = new TTLMap(reliabeReceiveStateTTL);
 
+  private lossyDataStatCurrentBytes: number = 0;
+
+  private lossyDataStatByterate: number = 0;
+
+  private lossyDataStatInterval: ReturnType<typeof setInterval> | undefined;
+
+  private lossyDataDropCount: number = 0;
+
   private midToTrackId: { [key: string]: string } = {};
 
   /** used to indicate whether the browser is currently waiting to reconnect */
@@ -312,6 +322,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
       this.removeAllListeners();
       this.deregisterOnLineListener();
       this.clearPendingReconnect();
+      this.cleanupLossyDataStats();
       await this.cleanupPeerConnections();
       await this.cleanupClient();
     } finally {
@@ -347,6 +358,16 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     this.reliableReceivedState.clear();
   }
 
+  cleanupLossyDataStats() {
+    this.lossyDataStatByterate = 0;
+    this.lossyDataStatCurrentBytes = 0;
+    if (this.lossyDataStatInterval) {
+      clearInterval(this.lossyDataStatInterval);
+      this.lossyDataStatInterval = undefined;
+    }
+    this.lossyDataDropCount = 0;
+  }
+
   async cleanupClient() {
     await this.client.close();
     this.client.resetCallbacks();
@@ -713,6 +734,22 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     // handle buffer amount low events
     this.lossyDC.onbufferedamountlow = this.handleBufferedAmountLow;
     this.reliableDC.onbufferedamountlow = this.handleBufferedAmountLow;
+
+    this.cleanupLossyDataStats();
+    this.lossyDataStatInterval = setInterval(() => {
+      this.lossyDataStatByterate = this.lossyDataStatCurrentBytes;
+      this.lossyDataStatCurrentBytes = 0;
+
+      const dc = this.dataChannelForKind(DataPacket_Kind.LOSSY);
+      if (dc) {
+        // control buffered latency to ~100ms
+        const threshold = this.lossyDataStatByterate / 10;
+        dc.bufferedAmountLowThreshold = Math.min(
+          Math.max(threshold, lossyDataChannelBufferThresholdMin),
+          lossyDataChannelBufferThresholdMax,
+        );
+      }
+    }, 1000);
   }
 
   private handleDataChannel = async ({ channel }: RTCDataChannelEvent) => {
@@ -1282,12 +1319,24 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
 
     const msg = packet.toBinary();
 
-    await this.waitForBufferStatusLow(kind);
-
     const dc = this.dataChannelForKind(kind);
     if (dc) {
       if (kind === DataPacket_Kind.RELIABLE) {
+        await this.waitForBufferStatusLow(kind);
         this.reliableMessageBuffer.push({ data: msg, sequence: packet.sequence });
+      } else {
+        // lossy channel, drop messages to reduce latency
+        if (!this.isBufferStatusLow(kind)) {
+          this.lossyDataDropCount += 1;
+          if (this.lossyDataDropCount % 100 === 0) {
+            this.log.warn(
+              `dropping lossy data channel messages, total dropped: ${this.lossyDataDropCount}`,
+              this.logContext,
+            );
+          }
+          return;
+        }
+        this.lossyDataStatCurrentBytes += msg.byteLength;
       }
 
       if (this.attemptingReconnect) {