Base A/V synchronization on sync_labels.

webrtc/video/call.cc

Index: webrtc/video/call.cc
diff --git a/webrtc/video/call.cc b/webrtc/video/call.cc
index cde41bc77a69d589c2cdb18620fbb981371ef9db..84c08609fbd39d8dfeff426d38fefd076345909e 100644
--- a/webrtc/video/call.cc
+++ b/webrtc/video/call.cc
@@ -65,6 +65,60 @@ class CpuOveruseObserverProxy : public webrtc::CpuOveruseObserver {
   LoadObserver* overuse_callback_ GUARDED_BY(crit_);
 };
 
+class SyncGroup {

[the sun, 2015/06/11 11:52:04]

Simplify as discussed offline.

[pbos-webrtc, 2015/06/11 14:48:53]

Done.

+ public:
+  explicit SyncGroup(VoiceEngine* voice_engine) : voice_engine_(voice_engine) {
+    DCHECK(voice_engine != nullptr);
+  }
+
+  ~SyncGroup() {
+    DCHECK(audio_channel_map_.empty());
+    DCHECK(video_streams_.empty());
+  }
+
+  void AddAudioReceiveStream(AudioReceiveStream* receive_stream,
+                             int channel_id) {
+    if (sync_channel_ == -1)
+      SetSyncChannel(channel_id);
+    audio_channel_map_[receive_stream] = channel_id;
+  }
+
+  void RemoveAudioReceiveStream(AudioReceiveStream* receive_stream) {
+    DCHECK(audio_channel_map_.find(receive_stream) != audio_channel_map_.end());
+    int old_channel = audio_channel_map_[receive_stream];
+    int current_channel = sync_channel_;
+    audio_channel_map_.erase(receive_stream);
+    if (old_channel != current_channel)
+      return;
+    SetSyncChannel(
+        audio_channel_map_.empty() ? -1 : audio_channel_map_.begin()->second);
+  }
+
+  void AddVideoReceiveStream(VideoReceiveStream* receive_stream) {
+    video_streams_.insert(receive_stream);
+    if (sync_channel_ == -1)
+      return;
+    receive_stream->SetSyncChannel(voice_engine_, sync_channel_);
+  }
+
+  void RemoveVideoReceiveStream(VideoReceiveStream* receive_stream) {
+    receive_stream->SetSyncChannel(voice_engine_, -1);
+    video_streams_.erase(receive_stream);
+  }
+
+ private:
+  void SetSyncChannel(int channel_id) {
+    sync_channel_ = channel_id;
+    for (VideoReceiveStream* video_stream : video_streams_)
+      video_stream->SetSyncChannel(voice_engine_, channel_id);
+  }
+
+  VoiceEngine* const voice_engine_;
+  int sync_channel_ = -1;
+  std::map<AudioReceiveStream*, int> audio_channel_map_;
+  std::set<VideoReceiveStream*> video_streams_;
+};
+
 class Call : public webrtc::Call, public PacketReceiver {
  public:
   explicit Call(const Call::Config& config);
@@ -109,6 +163,18 @@ class Call : public webrtc::Call, public PacketReceiver {
   void SetBitrateControllerConfig(
       const webrtc::Call::Config::BitrateConfig& bitrate_config);
 
+  void AddAudioReceiveStreamSync(AudioReceiveStream* stream,
+                                 const std::string& sync_group,
+                                 int channel_id);
+  void RemoveAudioReceiveStreamSync(AudioReceiveStream* stream);
+
+  void AddVideoReceiveStreamSync(VideoReceiveStream* stream,
+                                 const std::string& sync_group);
+  void RemoveVideoReceiveStreamSync(VideoReceiveStream* stream);
+
+  SyncGroup* FindSyncGroup(const std::string& sync_group)
+      EXCLUSIVE_LOCKS_REQUIRED(sync_group_crit_);
+
   const int num_cpu_cores_;
   const rtc::scoped_ptr<ProcessThread> module_process_thread_;
   const rtc::scoped_ptr<ChannelGroup> channel_group_;
@@ -135,6 +201,13 @@ class Call : public webrtc::Call, public PacketReceiver {
   std::map<uint32_t, VideoSendStream*> video_send_ssrcs_ GUARDED_BY(send_crit_);
   std::set<VideoSendStream*> video_send_streams_ GUARDED_BY(send_crit_);
 
+  rtc::CriticalSection sync_group_crit_;
+  std::map<VideoReceiveStream*, std::string> video_sync_groups_
+      GUARDED_BY(sync_group_crit_);
+  std::map<AudioReceiveStream*, std::string> audio_sync_groups_
+      GUARDED_BY(sync_group_crit_);
+  std::map<std::string, SyncGroup*> sync_groups_ GUARDED_BY(sync_group_crit_);
+
   rtc::scoped_ptr<CpuOveruseObserverProxy> overuse_observer_proxy_;
 
   VideoSendStream::RtpStateMap suspended_video_send_ssrcs_;
@@ -192,6 +265,8 @@ Call::~Call() {
   CHECK_EQ(0u, audio_receive_ssrcs_.size());
   CHECK_EQ(0u, video_receive_ssrcs_.size());
   CHECK_EQ(0u, video_receive_streams_.size());
+  for (auto& kv : sync_groups_)
+    delete kv.second;
 
   channel_group_->DeleteChannel(base_channel_id_);
   module_process_thread_->Stop();
@@ -214,6 +289,8 @@ webrtc::AudioReceiveStream* Call::CreateAudioReceiveStream(
   LOG(LS_INFO) << "CreateAudioReceiveStream: " << config.ToString();
   AudioReceiveStream* receive_stream = new AudioReceiveStream(
       channel_group_->GetRemoteBitrateEstimator(), config);
+  AddAudioReceiveStreamSync(receive_stream, config.sync_group,
+                            config.channel_id);
   {
     WriteLockScoped write_lock(*receive_crit_);
     DCHECK(audio_receive_ssrcs_.find(config.rtp.remote_ssrc) ==
@@ -229,6 +306,7 @@ void Call::DestroyAudioReceiveStream(
   DCHECK(receive_stream != nullptr);
   AudioReceiveStream* audio_receive_stream =
       static_cast<AudioReceiveStream*>(receive_stream);
+  RemoveAudioReceiveStreamSync(audio_receive_stream);
   {
     WriteLockScoped write_lock(*receive_crit_);
     size_t num_deleted = audio_receive_ssrcs_.erase(
@@ -309,6 +387,7 @@ webrtc::VideoReceiveStream* Call::CreateVideoReceiveStream(
       num_cpu_cores_, base_channel_id_, channel_group_.get(),
       rtc::AtomicOps::Increment(&next_channel_id_), config,
       config_.send_transport, config_.voice_engine);
+  AddVideoReceiveStreamSync(receive_stream, config.sync_group);
 
   // This needs to be taken before receive_crit_ as both locks need to be held
   // while changing network state.
@@ -333,7 +412,6 @@ void Call::DestroyVideoReceiveStream(
     webrtc::VideoReceiveStream* receive_stream) {
   TRACE_EVENT0("webrtc", "Call::DestroyVideoReceiveStream");
   DCHECK(receive_stream != nullptr);
-
   VideoReceiveStream* receive_stream_impl = nullptr;
   {
     WriteLockScoped write_lock(*receive_crit_);
@@ -353,6 +431,7 @@ void Call::DestroyVideoReceiveStream(
     video_receive_streams_.erase(receive_stream_impl);
   }
   CHECK(receive_stream_impl != nullptr);
+  RemoveVideoReceiveStreamSync(receive_stream_impl);
   delete receive_stream_impl;
 }
 
@@ -428,6 +507,53 @@ void Call::SignalNetworkState(NetworkState state) {
   }
 }
 
+void Call::AddAudioReceiveStreamSync(AudioReceiveStream* stream,
+                                     const std::string& sync_group,
+                                     int channel_id) {
+  if (!config_.voice_engine || sync_group.empty())
+    return;
+  rtc::CritScope lock(&sync_group_crit_);
+  audio_sync_groups_[stream] = sync_group;
+  SyncGroup* group = FindSyncGroup(sync_group);
+  group->AddAudioReceiveStream(stream, channel_id);
+}
+
+void Call::RemoveAudioReceiveStreamSync(AudioReceiveStream* stream) {
+  if (!config_.voice_engine)
+    return;
+  rtc::CritScope lock(&sync_group_crit_);
+  SyncGroup* group = FindSyncGroup(audio_sync_groups_[stream]);
+  group->RemoveAudioReceiveStream(stream);
+}
+
+void Call::AddVideoReceiveStreamSync(VideoReceiveStream* stream,
+                                     const std::string& sync_group) {
+  if (!config_.voice_engine || sync_group.empty())
+    return;
+  rtc::CritScope lock(&sync_group_crit_);
+  video_sync_groups_[stream] = sync_group;
+  SyncGroup* group = FindSyncGroup(sync_group);
+  group->AddVideoReceiveStream(stream);
+}
+
+void Call::RemoveVideoReceiveStreamSync(VideoReceiveStream* stream) {
+  if (!config_.voice_engine)
+    return;
+  rtc::CritScope lock(&sync_group_crit_);
+  SyncGroup* group = FindSyncGroup(video_sync_groups_[stream]);
+  group->RemoveVideoReceiveStream(stream);
+}
+
+SyncGroup* Call::FindSyncGroup(const std::string& sync_group) {
+  DCHECK(!sync_group.empty());
+
+  if (sync_groups_.find(sync_group) == sync_groups_.end())
+    sync_groups_[sync_group] = new SyncGroup(config_.voice_engine);
+
+  SyncGroup* group = sync_groups_[sync_group];
+  return group;
+}
+
 PacketReceiver::DeliveryStatus Call::DeliverRtcp(MediaType media_type,
                                                  const uint8_t* packet,
                                                  size_t length) {