If the number of handlers is large, split Conference.sendOut on a task queue.

src/main/java/org/jitsi/videobridge/Conference.java

@@ -1146,6 +1146,11 @@ public ConferenceShim getShim()
         return shim;
     }
 
+    /** The maximum number of packet handlers we want to execute in one task.
+     *   TODO: this number is pulled out of the air; tune it.
+     */
+    public static final int MAX_HANDLERS_PER_TASK = 5;
+
     /**
      * Broadcasts the packet to all endpoints and tentacles that want it.
      *
@@ -1154,45 +1159,73 @@ public ConferenceShim getShim()
     private void sendOut(PacketInfo packetInfo)
     {
         String sourceEndpointId = packetInfo.getEndpointId();
-        // We want to avoid calling 'clone' for the last receiver of this packet
-        // since it's unnecessary.  To do so, we'll wait before we clone and send
-        // to an interested handler until after we've determined another handler
-        // is also interested in the packet.  We'll give the last handler the
-        // original packet (without cloning).
-        PotentialPacketHandler prevHandler = null;
-        for (Endpoint endpoint : endpointsCache)
+
+        List<Runnable> tasks = new ArrayList<>();
+        PacketInfoDistributor distributor = new PacketInfoDistributor(packetInfo, 0);
+
+        for (Endpoint e: endpointsCache)
         {
-            if (endpoint.getID().equals(sourceEndpointId))
+            if (e.getID().equals(sourceEndpointId))
             {
                 continue;
             }
 
-            if (endpoint.wants(packetInfo))
-            {
-                if (prevHandler != null)
-                {
-                    prevHandler.send(packetInfo.clone());
-                }
-                prevHandler = endpoint;
-            }
+            tasks.add(() -> doSendOut(distributor, e));
         }
-        if (tentacle != null && tentacle.wants(packetInfo))
+        if (tentacle != null)
         {
-            if (prevHandler != null)
+            tasks.add(() -> doSendOut(distributor, tentacle));
+        }
+
+        distributor.setCount(tasks.size());
+
+        if (tasks.isEmpty()) {
+            return;
+        }
+        else if (tasks.size() == 1) {
+            tasks.get(0).run();
+            return;
+        }
+
+        List<Future<?>> taskFutures = new ArrayList<>();
+
+        for (Runnable task : tasks)
+        {
+            taskFutures.add(TaskPools.CPU_POOL.submit(task));
+        }
+
+        /* This doesn't work - these tasks will be on the same task pool as this
+           code, causing deadlocks. */
+        /*
+        for (Future<?> task: taskFutures)
+        {
+            try
             {
-                prevHandler.send(packetInfo.clone());
+                task.get();
+            }
+            catch (InterruptedException | ExecutionException e)
+            {
+                logger.warn("Exception waiting for sendOut", e);
             }
-            prevHandler = tentacle;
         }
+        */
+    }
 
-        if (prevHandler != null)
+    /**
+     * Handles a packet for one packet handler.
+     *
+     * @param distributor A distributor for the packet
+     * @param handler the packet handlers to send it to
+     */
+    private static void doSendOut(PacketInfoDistributor distributor, PotentialPacketHandler handler)
+    {
+        if (handler.wants(distributor.previewPacketInfo()))
         {
-            prevHandler.send(packetInfo);
+            distributor.usePacketInfoReference(handler::send);
         }
         else
         {
-            // No one wanted the packet, so the buffer is now free!
-            ByteBufferPool.returnBuffer(packetInfo.getPacket().getBuffer());
+            distributor.releasePacketInfoReference();
         }
     }
 

src/main/java/org/jitsi/videobridge/Endpoint.java

@@ -386,7 +386,7 @@ public double getRtt()
      * {@inheritDoc}
      */
     @Override
-    public boolean wants(PacketInfo packetInfo)
+    public synchronized boolean wants(PacketInfo packetInfo)
     {
         if (!isTransportConnected())
         {
@@ -441,7 +441,7 @@ else if (packet instanceof RtcpFbPliPacket ||
      * TODO Brian
      */
     @Override
-    public void send(PacketInfo packetInfo)
+    public synchronized void send(PacketInfo packetInfo)
     {
         Packet packet = packetInfo.getPacket();
         if (packet instanceof VideoRtpPacket)

src/main/java/org/jitsi/videobridge/octo/OctoTentacle.java

@@ -145,7 +145,7 @@ public void addPayloadType(PayloadType payloadType)
      * {@inheritDoc}
      */
     @Override
-    public void send(PacketInfo packetInfo)
+    public synchronized void send(PacketInfo packetInfo)
     {
         Packet packet = packetInfo.getPacket();
         if (packet != null)

src/main/java/org/jitsi/videobridge/util/PacketInfoDistributor.java

@@ -0,0 +1,125 @@
+/*
+ * Copyright @ 2018 - present 8x8, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.jitsi.videobridge.util;
+
+import org.jitsi.nlj.*;
+
+import java.util.function.*;
+
+public class PacketInfoDistributor
+{
+    private final PacketInfo packetInfo;
+    private int totalReferences;
+    private int outstandingReferences;
+
+    private Consumer<PacketInfo> prevConsumer = null;
+
+    public PacketInfoDistributor(PacketInfo pi, int count)
+    {
+        packetInfo = pi;
+        totalReferences = count;
+        outstandingReferences = count;
+        /* We create a PacketInfoDistributor on each packet so we don't want to instantiate a new logger each time. */
+    }
+
+    public void setCount(int count)
+    {
+        if (outstandingReferences < totalReferences)
+        {
+            throw new IllegalStateException("Count set after references have been taken");
+        }
+        totalReferences = count;
+        outstandingReferences = count;
+    }
+
+    public synchronized PacketInfo previewPacketInfo()
+    {
+        if (outstandingReferences <= 0)
+        {
+            throw new IllegalStateException("Packet previewed after all references taken");
+        }
+        return packetInfo;
+    }
+
+    public void usePacketInfoReference(Consumer<PacketInfo> consumer)
+    {
+        // We want to avoid calling 'clone' for the last receiver of this packet
+        // since it's unnecessary.  To do so, we'll wait before we clone and send
+        // to an interested handler until after we've determined another handler
+        // is also interested in the packet.  We'll give the last handler the
+        // original packet (without cloning).
+        Consumer<PacketInfo> c1 = null, c2 = null;
+        PacketInfo p1 = null, p2 = null;
+        synchronized(this) {
+            outstandingReferences--;
+            if (outstandingReferences < 0)
+            {
+                throw new IllegalStateException("Too many references taken");
+            }
+            if (prevConsumer != null) {
+                c1 = prevConsumer;
+                p1 = packetInfo.clone();
+            }
+            prevConsumer = consumer;
+            if (outstandingReferences == 0)
+            {
+                c2 = prevConsumer;
+                p2 = packetInfo;
+            }
+        }
+
+        if (c1 != null)
+        {
+            c1.accept(p1);
+        }
+        if (c2 != null)
+        {
+            c2.accept(p2);
+        }
+    }
+
+    public void releasePacketInfoReference()
+    {
+        Consumer<PacketInfo> c = null;
+        PacketInfo p = null;
+        synchronized (this)
+        {
+            outstandingReferences--;
+            if (outstandingReferences < 0)
+            {
+                throw new IllegalStateException("Too many references taken");
+            }
+            if (outstandingReferences == 0)
+            {
+                if (prevConsumer != null)
+                {
+                    c = prevConsumer;
+                    p = packetInfo;
+                }
+                else
+                {
+                    ByteBufferPool.returnBuffer(packetInfo.getPacket().getBuffer());
+                }
+            }
+        }
+        if (c != null)
+        {
+            c.accept(p);
+        }
+    }
+}