feat: Use the ice4j push API. (#2195)

* feat: Use the ice4j push API. * feat: Put incoming DTLS packets on a queue. * Start writing to the ICE transport as soon as we have any validated pair. * chore: Bump to ice4j 3.2 --------- Co-authored-by: Jonathan Lennox <[email protected]>
jitsi · Oct 2, 2024 · 0b2e8d6 · 0b2e8d6
1 parent 93b2699
commit 0b2e8d6
Show file tree

Hide file tree

Showing 6 changed files with 129 additions and 58 deletions.
diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/Endpoint.kt
@@ -16,6 +16,7 @@
 
 package org.jitsi.videobridge
 
+import org.ice4j.util.Buffer
 import org.jitsi.config.JitsiConfig
 import org.jitsi.dcsctp4j.DcSctpMessage
 import org.jitsi.dcsctp4j.ErrorKind
@@ -45,13 +46,11 @@ import org.jitsi.nlj.util.NEVER
 import org.jitsi.nlj.util.PacketInfoQueue
 import org.jitsi.nlj.util.RemoteSsrcAssociation
 import org.jitsi.nlj.util.sumOf
-import org.jitsi.rtp.Packet
 import org.jitsi.rtp.UnparsedPacket
 import org.jitsi.rtp.rtcp.RtcpSrPacket
 import org.jitsi.rtp.rtcp.rtcpfb.RtcpFbPacket
 import org.jitsi.rtp.rtcp.rtcpfb.payload_specific_fb.RtcpFbFirPacket
 import org.jitsi.rtp.rtcp.rtcpfb.payload_specific_fb.RtcpFbPliPacket
-import org.jitsi.rtp.rtp.RtpPacket
 import org.jitsi.utils.MediaType
 import org.jitsi.utils.concurrent.RecurringRunnableExecutor
 import org.jitsi.utils.logging2.Logger
@@ -166,7 +165,7 @@ class Endpoint @JvmOverloads constructor(
 
     /* TODO: do we ever want to support useUniquePort for an Endpoint? */
     private val iceTransport = IceTransport(id, iceControlling, false, supportsPrivateAddresses, logger)
-    private val dtlsTransport = DtlsTransport(logger).also { it.cryptex = CryptexConfig.endpoint }
+    private val dtlsTransport = DtlsTransport(logger, id).also { it.cryptex = CryptexConfig.endpoint }
 
     private var cryptex: Boolean = CryptexConfig.endpoint
 
@@ -407,41 +406,34 @@ class Endpoint @JvmOverloads constructor(
 
     private fun setupIceTransport() {
         iceTransport.incomingDataHandler = object : IceTransport.IncomingDataHandler {
-            override fun dataReceived(data: ByteArray, offset: Int, length: Int, receivedTime: Instant) {
-                // DTLS data will be handled by the DtlsTransport, but SRTP data can go
-                // straight to the transceiver
-                if (looksLikeDtls(data, offset, length)) {
-                    // DTLS transport is responsible for making its own copy, because it will manage its own
-                    // buffers
-                    dtlsTransport.dtlsDataReceived(data, offset, length)
+            override fun dataReceived(buffer: Buffer) {
+                if (looksLikeDtls(buffer.buffer, buffer.offset, buffer.length)) {
+                    // DTLS transport is responsible for making its own copy, because it will manage its own buffers
+                    dtlsTransport.enqueueBuffer(buffer)
                 } else {
-                    val copy = ByteBufferPool.getBuffer(
-                        length +
-                            RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET +
-                            Packet.BYTES_TO_LEAVE_AT_END_OF_PACKET
-                    )
-                    System.arraycopy(data, offset, copy, RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET, length)
                     val pktInfo =
-                        PacketInfo(UnparsedPacket(copy, RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET, length)).apply {
-                            this.receivedTime = receivedTime
+                        PacketInfo(UnparsedPacket(buffer.buffer, buffer.offset, buffer.length)).apply {
+                            this.receivedTime = buffer.receivedTime
                         }
                     transceiver.handleIncomingPacket(pktInfo)
                 }
             }
         }
         iceTransport.eventHandler = object : IceTransport.EventHandler {
-            override fun connected() {
+            override fun writeable() {
                 logger.info("ICE connected")
                 transceiver.setOutgoingPacketHandler(object : PacketHandler {
                     override fun processPacket(packetInfo: PacketInfo) {
                         packetInfo.addEvent(SRTP_QUEUE_ENTRY_EVENT)
                         outgoingSrtpPacketQueue.add(packetInfo)
                     }
                 })
-                TaskPools.IO_POOL.execute(iceTransport::startReadingData)
                 TaskPools.IO_POOL.execute(dtlsTransport::startDtlsHandshake)
             }
 
+            override fun connected() {
+            }
+
             override fun failed() {
             }
 

diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/Main.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/Main.kt
@@ -18,7 +18,9 @@ package org.jitsi.videobridge
 
 import org.eclipse.jetty.servlet.ServletHolder
 import org.glassfish.jersey.servlet.ServletContainer
+import org.ice4j.ice.harvest.AbstractUdpListener
 import org.ice4j.ice.harvest.MappingCandidateHarvesters
+import org.ice4j.util.Buffer
 import org.jitsi.config.JitsiConfig
 import org.jitsi.metaconfig.ConfigException
 import org.jitsi.metaconfig.MetaconfigLogger
@@ -29,6 +31,8 @@ import org.jitsi.rest.createServer
 import org.jitsi.rest.enableCors
 import org.jitsi.rest.isEnabled
 import org.jitsi.rest.servletContextHandler
+import org.jitsi.rtp.Packet
+import org.jitsi.rtp.rtp.RtpPacket
 import org.jitsi.shutdown.ShutdownServiceImpl
 import org.jitsi.utils.logging2.LoggerImpl
 import org.jitsi.utils.queue.PacketQueue
@@ -220,6 +224,7 @@ private fun getSystemPropertyDefaults(): Map<String, String> {
 }
 
 private fun startIce4j() {
+    AbstractUdpListener.USE_PUSH_API = true
     // Start the initialization of the mapping candidate harvesters.
     // Asynchronous, because the AWS and STUN harvester may take a long
     // time to initialize.
@@ -239,4 +244,13 @@ private fun setupBufferPools() {
     org.jitsi.rtp.util.BufferPool.returnArray = { ByteBufferPool.returnBuffer(it) }
     org.jitsi.nlj.util.BufferPool.getBuffer = { ByteBufferPool.getBuffer(it) }
     org.jitsi.nlj.util.BufferPool.returnBuffer = { ByteBufferPool.returnBuffer(it) }
+    org.ice4j.util.BufferPool.getBuffer = { len ->
+        val b = ByteBufferPool.getBuffer(len)
+        Buffer(b, 0, b.size)
+    }
+    org.ice4j.util.BufferPool.returnBuffer = { ByteBufferPool.returnBuffer(it.buffer) }
+    org.ice4j.ice.harvest.AbstractUdpListener.BYTES_TO_LEAVE_AT_START_OF_PACKET =
+        RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET
+    org.ice4j.ice.harvest.AbstractUdpListener.BYTES_TO_LEAVE_AT_END_OF_PACKET =
+        Packet.BYTES_TO_LEAVE_AT_END_OF_PACKET
 }
diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/relay/Relay.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/relay/Relay.kt
@@ -15,6 +15,7 @@
  */
 package org.jitsi.videobridge.relay
 
+import org.ice4j.util.Buffer
 import org.jitsi.dcsctp4j.DcSctpMessage
 import org.jitsi.dcsctp4j.ErrorKind
 import org.jitsi.dcsctp4j.SendPacketStatus
@@ -47,7 +48,6 @@ import org.jitsi.nlj.util.LocalSsrcAssociation
 import org.jitsi.nlj.util.PacketInfoQueue
 import org.jitsi.nlj.util.RemoteSsrcAssociation
 import org.jitsi.nlj.util.sumOf
-import org.jitsi.rtp.Packet
 import org.jitsi.rtp.UnparsedPacket
 import org.jitsi.rtp.extensions.looksLikeRtcp
 import org.jitsi.rtp.extensions.looksLikeRtp
@@ -220,7 +220,7 @@ class Relay @JvmOverloads constructor(
         clock = clock
     )
 
-    private val dtlsTransport = DtlsTransport(logger).also { it.cryptex = CryptexConfig.relay }
+    private val dtlsTransport = DtlsTransport(logger, id).also { it.cryptex = CryptexConfig.relay }
 
     private var cryptex = CryptexConfig.relay
 
@@ -365,44 +365,38 @@ class Relay @JvmOverloads constructor(
 
     private fun setupIceTransport() {
         iceTransport.incomingDataHandler = object : IceTransport.IncomingDataHandler {
-            override fun dataReceived(data: ByteArray, offset: Int, length: Int, receivedTime: Instant) {
+            override fun dataReceived(buffer: Buffer) {
                 // DTLS data will be handled by the DtlsTransport, but SRTP data can go
                 // straight to the transceiver
-                if (looksLikeDtls(data, offset, length)) {
-                    // DTLS transport is responsible for making its own copy, because it will manage its own
-                    // buffers
-                    dtlsTransport.dtlsDataReceived(data, offset, length)
+                if (looksLikeDtls(buffer.buffer, buffer.offset, buffer.length)) {
+                    // DTLS transport is responsible for making its own copy, because it will manage its own buffers
+                    dtlsTransport.enqueueBuffer(buffer)
                 } else {
-                    val copy = ByteBufferPool.getBuffer(
-                        length +
-                            RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET +
-                            Packet.BYTES_TO_LEAVE_AT_END_OF_PACKET
-                    )
-                    System.arraycopy(data, offset, copy, RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET, length)
                     val pktInfo =
                         RelayedPacketInfo(
-                            UnparsedPacket(copy, RtpPacket.BYTES_TO_LEAVE_AT_START_OF_PACKET, length),
+                            UnparsedPacket(buffer.buffer, buffer.offset, buffer.length),
                             meshId
                         ).apply {
-                            this.receivedTime = receivedTime
+                            this.receivedTime = buffer.receivedTime
                         }
                     handleMediaPacket(pktInfo)
                 }
             }
         }
         iceTransport.eventHandler = object : IceTransport.EventHandler {
-            override fun connected() {
+            override fun writeable() {
                 logger.info("ICE connected")
                 transceiver.setOutgoingPacketHandler(object : PacketHandler {
                     override fun processPacket(packetInfo: PacketInfo) {
                         packetInfo.addEvent(SRTP_QUEUE_ENTRY_EVENT)
                         outgoingSrtpPacketQueue.add(packetInfo)
                     }
                 })
-                TaskPools.IO_POOL.execute(iceTransport::startReadingData)
                 TaskPools.IO_POOL.execute(dtlsTransport::startDtlsHandshake)
             }
 
+            override fun connected() {}
+
             override fun failed() {}
 
             override fun consentUpdated(time: Instant) {

diff --git a/jvb/src/main/kotlin/org/jitsi/videobridge/transport/dtls/DtlsTransport.kt b/jvb/src/main/kotlin/org/jitsi/videobridge/transport/dtls/DtlsTransport.kt
@@ -16,13 +16,17 @@
 
 package org.jitsi.videobridge.transport.dtls
 
+import org.ice4j.util.Buffer
 import org.jitsi.nlj.dtls.DtlsClient
 import org.jitsi.nlj.dtls.DtlsServer
 import org.jitsi.nlj.dtls.DtlsStack
 import org.jitsi.nlj.srtp.TlsRole
+import org.jitsi.nlj.util.BufferPool
 import org.jitsi.utils.OrderedJsonObject
 import org.jitsi.utils.logging2.Logger
 import org.jitsi.utils.logging2.createChildLogger
+import org.jitsi.utils.queue.PacketQueue
+import org.jitsi.videobridge.util.TaskPools
 import org.jitsi.xmpp.extensions.jingle.DtlsFingerprintPacketExtension
 import org.jitsi.xmpp.extensions.jingle.IceUdpTransportPacketExtension
 import java.util.concurrent.atomic.AtomicBoolean
@@ -39,7 +43,7 @@ import java.util.concurrent.atomic.AtomicBoolean
  * be passed to the [outgoingDataHandler], which should be set by an
  * interested party.
  */
-class DtlsTransport(parentLogger: Logger) {
+class DtlsTransport(parentLogger: Logger, id: String) {
     private val logger = createChildLogger(parentLogger)
 
     private val running = AtomicBoolean(true)
@@ -62,6 +66,26 @@ class DtlsTransport(parentLogger: Logger) {
     /** Whether to advertise cryptex to peers. */
     var cryptex = false
 
+    val dtlsQueue = object : PacketQueue<Buffer>(
+        128,
+        null,
+        "dtls-queue-$id",
+        { buffer: Buffer ->
+            try {
+                dtlsDataReceived(buffer.buffer, buffer.offset, buffer.length)
+                true
+            } catch (e: Exception) {
+                logger.warn("Failed to handle DTLS data", e)
+                false
+            }
+        },
+        TaskPools.IO_POOL,
+    ) {
+        override fun releasePacket(buffer: Buffer) {
+            BufferPool.returnBuffer(buffer.buffer)
+        }
+    }
+
     /**
      * The DTLS stack instance
      */
@@ -170,10 +194,13 @@ class DtlsTransport(parentLogger: Logger) {
         }
     }
 
+    fun enqueueBuffer(buffer: Buffer) = dtlsQueue.add(buffer)
+
     /**
      * Notify this layer that DTLS data has been received from the network
      */
-    fun dtlsDataReceived(data: ByteArray, off: Int, len: Int) = dtlsStack.processIncomingProtocolData(data, off, len)
+    private fun dtlsDataReceived(data: ByteArray, off: Int, len: Int) =
+        dtlsStack.processIncomingProtocolData(data, off, len)
 
     /**
      * Send out DTLS data