Skip to content

Commit

Permalink
Merge queued remote source changes. (#784)
Browse files Browse the repository at this point in the history
* fix: Synchronize access to queuedRemoteSourceChanges.

* fix: Reduce the number queued of source-add/source-removes

Merges consecutive source-add or source-remove operations to reduce the
number of messages sent. In a large conference in which participants
join at a high rate there can be a lot of source-adds queued and it's
more efficient to merge them in one message.
  • Loading branch information
bgrozev authored Aug 11, 2021
1 parent 274d8b4 commit 2f72367
Show file tree
Hide file tree
Showing 2 changed files with 161 additions and 7 deletions.
54 changes: 47 additions & 7 deletions src/main/java/org/jitsi/jicofo/AbstractParticipant.java
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@

import java.util.*;

import org.jetbrains.annotations.*;
import com.google.common.collect.*;
import org.jitsi.jicofo.conference.*;
import org.jitsi.jicofo.conference.source.*;
import org.jitsi.xmpp.extensions.colibri.*;
Expand Down Expand Up @@ -52,7 +52,7 @@ public abstract class AbstractParticipant
/**
* List of remote source addition or removal operations that have not yet been signaled to this participant.
*/
private List<SourcesToAddOrRemove> queuedRemoteSourceChanges = new ArrayList<>();
private final List<SourcesToAddOrRemove> queuedRemoteSourceChanges = new ArrayList<>();

/**
* Returns currently stored map of RTP description to Colibri content name.
Expand Down Expand Up @@ -118,9 +118,23 @@ public void setRTPDescription(List<ContentPacketExtension> jingleContents)
*/
public List<SourcesToAddOrRemove> clearQueuedRemoteSourceChanges()
{
List<SourcesToAddOrRemove> ret = queuedRemoteSourceChanges;
queuedRemoteSourceChanges = new ArrayList<>();
return ret;
synchronized (queuedRemoteSourceChanges)
{
List<SourcesToAddOrRemove> ret = new ArrayList<>(queuedRemoteSourceChanges);
queuedRemoteSourceChanges.clear();
return ret;
}
}

/**
* Gets the list of pending remote sources, without clearing them. For testing.
*/
public List<SourcesToAddOrRemove> getQueuedRemoteSourceChanges()
{
synchronized (queuedRemoteSourceChanges)
{
return new ArrayList<>(queuedRemoteSourceChanges);
}
}

/**
Expand All @@ -130,7 +144,20 @@ public List<SourcesToAddOrRemove> clearQueuedRemoteSourceChanges()
*/
public void queueRemoteSourcesToAdd(ConferenceSourceMap sourcesToAdd)
{
queuedRemoteSourceChanges.add(new SourcesToAddOrRemove(AddOrRemove.Add, sourcesToAdd));
synchronized (queuedRemoteSourceChanges)
{
SourcesToAddOrRemove previous = Iterables.getLast(queuedRemoteSourceChanges, null);
if (previous != null && previous.getAction() == AddOrRemove.Add)
{
// We merge sourcesToAdd with the previous sources queued to be added to reduce the number of
// source-add messages that need to be sent.
queuedRemoteSourceChanges.remove(queuedRemoteSourceChanges.size() - 1);
sourcesToAdd = sourcesToAdd.copy();
sourcesToAdd.add(previous.getSources());
}

queuedRemoteSourceChanges.add(new SourcesToAddOrRemove(AddOrRemove.Add, sourcesToAdd));
}
}

/**
Expand All @@ -140,7 +167,20 @@ public void queueRemoteSourcesToAdd(ConferenceSourceMap sourcesToAdd)
*/
public void queueRemoteSourcesToRemove(ConferenceSourceMap sourcesToRemove)
{
queuedRemoteSourceChanges.add(new SourcesToAddOrRemove(AddOrRemove.Remove, sourcesToRemove));
synchronized (queuedRemoteSourceChanges)
{
SourcesToAddOrRemove previous = Iterables.getLast(queuedRemoteSourceChanges, null);
if (previous != null && previous.getAction() == AddOrRemove.Remove)
{
// We merge sourcesToRemove with the previous sources queued to be remove to reduce the number of
// source-remove messages that need to be sent.
queuedRemoteSourceChanges.remove(queuedRemoteSourceChanges.size() - 1);
sourcesToRemove = sourcesToRemove.copy();
sourcesToRemove.add(previous.getSources());
}

queuedRemoteSourceChanges.add(new SourcesToAddOrRemove(AddOrRemove.Remove, sourcesToRemove));
}
}

/**
Expand Down
114 changes: 114 additions & 0 deletions src/test/kotlin/org/jitsi/jicofo/conference/AbstractParticipantTest.kt
Original file line number Diff line number Diff line change
@@ -0,0 +1,114 @@
/*
* Jicofo, the Jitsi Conference Focus.
*
* Copyright @ 2021-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.jicofo.conference

import io.kotest.core.spec.style.ShouldSpec
import io.kotest.matchers.shouldBe
import org.jitsi.jicofo.AbstractParticipant
import org.jitsi.jicofo.conference.AddOrRemove.Add
import org.jitsi.jicofo.conference.AddOrRemove.Remove
import org.jitsi.jicofo.conference.source.ConferenceSourceMap
import org.jitsi.jicofo.conference.source.EndpointSourceSet
import org.jitsi.jicofo.conference.source.Source
import org.jitsi.utils.MediaType.AUDIO
import org.jitsi.utils.logging2.LoggerImpl
import org.jxmpp.jid.impl.JidCreate

class AbstractParticipantTest : ShouldSpec() {
init {
val jid1 = JidCreate.from("jid1")
val s1 = ConferenceSourceMap(jid1 to EndpointSourceSet(Source(1, AUDIO))).unmodifiable

val jid2 = JidCreate.from("jid2")
val s2 = ConferenceSourceMap(jid2 to EndpointSourceSet(Source(2, AUDIO))).unmodifiable
val s2new = ConferenceSourceMap(
jid2 to EndpointSourceSet(
setOf(
Source(2, AUDIO),
Source(222, AUDIO)
)
)
).unmodifiable

val jid3 = JidCreate.from("jid3")
val s3 = ConferenceSourceMap(jid3 to EndpointSourceSet(Source(3, AUDIO))).unmodifiable

val participant = TestParticipant()

context("Queueing remote sources") {
participant.queueRemoteSourcesToAdd(s1)
participant.queuedRemoteSourceChanges.size shouldBe 1
participant.queuedRemoteSourceChanges[0].action shouldBe Add
participant.queuedRemoteSourceChanges[0].sources.toMap() shouldBe s1.toMap()

// Consecutive source-adds should be merged.
participant.queueRemoteSourcesToAdd(s2)
participant.queuedRemoteSourceChanges.size shouldBe 1
participant.queuedRemoteSourceChanges[0].action shouldBe Add
participant.queuedRemoteSourceChanges[0].sources.toMap() shouldBe
s1.copy().apply { add(s2) }.toMap()

// Consecutive source-adds should be merged.
participant.queueRemoteSourcesToAdd(s2new)
participant.queuedRemoteSourceChanges.size shouldBe 1
participant.queuedRemoteSourceChanges[0].action shouldBe Add
participant.queuedRemoteSourceChanges[0].sources.toMap() shouldBe
s1.copy().apply { add(s2); add(s2new) }.toMap()

// A source-remove after a series of source-adds should be a new entry.
participant.queueRemoteSourcesToRemove(s2new)
participant.queuedRemoteSourceChanges.size shouldBe 2
participant.queuedRemoteSourceChanges[0].action shouldBe Add
participant.queuedRemoteSourceChanges[0].sources.toMap() shouldBe
s1.copy().apply { add(s2); add(s2new) }.toMap()
participant.queuedRemoteSourceChanges[1].action shouldBe Remove
participant.queuedRemoteSourceChanges[1].sources.toMap() shouldBe s2new.toMap()

// A source-add following source-remove should be a new entry.
participant.queueRemoteSourcesToAdd(s3)
participant.queuedRemoteSourceChanges.size shouldBe 3
participant.queuedRemoteSourceChanges[0].action shouldBe Add
participant.queuedRemoteSourceChanges[0].sources.toMap() shouldBe
s1.copy().apply { add(s2); add(s2new) }.toMap()
participant.queuedRemoteSourceChanges[1].action shouldBe Remove
participant.queuedRemoteSourceChanges[1].sources.toMap() shouldBe s2new.toMap()
participant.queuedRemoteSourceChanges[2].action shouldBe Add
participant.queuedRemoteSourceChanges[2].sources.toMap() shouldBe s3.toMap()

// Consecutive source-adds should be merged.
participant.queueRemoteSourcesToAdd(s1)
participant.queuedRemoteSourceChanges.size shouldBe 3
participant.queuedRemoteSourceChanges[0].action shouldBe Add
participant.queuedRemoteSourceChanges[0].sources.toMap() shouldBe
s1.copy().apply { add(s2); add(s2new) }.toMap()
participant.queuedRemoteSourceChanges[1].action shouldBe Remove
participant.queuedRemoteSourceChanges[1].sources.toMap() shouldBe s2new.toMap()
participant.queuedRemoteSourceChanges[2].action shouldBe Add
participant.queuedRemoteSourceChanges[2].sources.toMap() shouldBe
s3.copy().apply { add(s1) }.toMap()

participant.clearQueuedRemoteSourceChanges()
participant.queuedRemoteSourceChanges shouldBe emptyList()
}
}
}

class TestParticipant : AbstractParticipant(LoggerImpl("AbstractParticipantTest")) {
override fun getSources() = ConferenceSourceMap()
override fun isSessionEstablished() = true
}

0 comments on commit 2f72367

Please sign in to comment.