TransportController refactoring

talk/session/media/channel.cc

 /*
  * libjingle
  * Copyright 2004 Google Inc.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are met:
  *
  *  1. Redistributions of source code must retain the above copyright notice,
  *     this list of conditions and the following disclaimer.
  *  2. Redistributions in binary form must reproduce the above copyright notice,
@@ skipping 153 matching lines @@
 
 template <class Codec, class Options>
 void RtpSendParametersFromMediaDescription(
     const MediaContentDescriptionImpl<Codec>* desc,
     RtpSendParameters<Codec, Options>* send_params) {
   RtpParametersFromMediaDescription(desc, send_params);
   send_params->max_bandwidth_bps = desc->bandwidth();
 }
 
 BaseChannel::BaseChannel(rtc::Thread* thread,
-                         MediaChannel* media_channel, BaseSession* session,
-                         const std::string& content_name, bool rtcp)
+                         MediaChannel* media_channel,
+                         TransportController* transport_controller,
+                         const std::string& content_name,
+                         bool rtcp)
     : worker_thread_(thread),
-      session_(session),
+      transport_controller_(transport_controller),
       media_channel_(media_channel),
       content_name_(content_name),
-      rtcp_(rtcp),
-      transport_channel_(NULL),
-      rtcp_transport_channel_(NULL),
+      rtcp_transport_enabled_(rtcp),
+      transport_channel_(nullptr),
+      rtcp_transport_channel_(nullptr),
       enabled_(false),
       writable_(false),
       rtp_ready_to_send_(false),
       rtcp_ready_to_send_(false),
       was_ever_writable_(false),
       local_content_direction_(MD_INACTIVE),
       remote_content_direction_(MD_INACTIVE),
       has_received_packet_(false),
       dtls_keyed_(false),
       secure_required_(false),
       rtp_abs_sendtime_extn_id_(-1) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   LOG(LS_INFO) << "Created channel for " << content_name;
 }
 
 BaseChannel::~BaseChannel() {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   Deinit();
   StopConnectionMonitor();
   FlushRtcpMessages();  // Send any outstanding RTCP packets.
   worker_thread_->Clear(this);  // eats any outstanding messages or packets
   // We must destroy the media channel before the transport channel, otherwise
   // the media channel may try to send on the dead transport channel. NULLing
   // is not an effective strategy since the sends will come on another thread.
   delete media_channel_;
-  set_transport_channel(nullptr);
-  set_rtcp_transport_channel(nullptr);
+  // Note that we don't just call set_transport_channel(nullptr) because that
+  // would call a pure virtual method which we can't do from a destructor.
+  if (transport_channel_) {
+    DisconnectFromTransportChannel(transport_channel_);
+    transport_controller_->DestroyTransportChannel_w(
+        transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTP);
+  }
+  if (rtcp_transport_channel_) {
+    DisconnectFromTransportChannel(rtcp_transport_channel_);
+    transport_controller_->DestroyTransportChannel_w(
+        transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTCP);
+  }
   LOG(LS_INFO) << "Destroyed channel";
 }
 
 bool BaseChannel::Init() {
-  if (!SetTransportChannels(session(), rtcp())) {
+  if (!SetTransportChannels(content_name())) {
     return false;
   }
 
   if (!SetDtlsSrtpCiphers(transport_channel(), false)) {
     return false;
   }
-  if (rtcp() && !SetDtlsSrtpCiphers(rtcp_transport_channel(), true)) {
+  if (rtcp_transport_enabled() &&
+      !SetDtlsSrtpCiphers(rtcp_transport_channel(), true)) {
     return false;
   }
 
   // Both RTP and RTCP channels are set, we can call SetInterface on
   // media channel and it can set network options.
   media_channel_->SetInterface(this);
   return true;
 }
 
 void BaseChannel::Deinit() {
   media_channel_->SetInterface(NULL);
 }
 
-bool BaseChannel::SetTransportChannels(BaseSession* session, bool rtcp) {
-  return worker_thread_->Invoke<bool>(Bind(
-      &BaseChannel::SetTransportChannels_w, this, session, rtcp));
+bool BaseChannel::SetTransport(const std::string& transport_name) {
+  if (!SetTransportChannels(transport_name)) {
+    return false;
+  }
+  return true;
 }
 
-bool BaseChannel::SetTransportChannels_w(BaseSession* session, bool rtcp) {
+bool BaseChannel::SetTransportChannels(const std::string& transport_name) {
+  return worker_thread_->Invoke<bool>(
+      Bind(&BaseChannel::SetTransportChannels_w, this, transport_name));
+}
+
+bool BaseChannel::SetTransportChannels_w(const std::string& transport_name) {

[pthatcher1, 2015/08/31 22:01:35]

Maybe we should just call this SetTransport_w, and then we wouldn't need both
SetTransport and SetTransportChannels (we could have just SetTransport).

[Taylor Brandstetter, 2015/09/01 23:53:30]

Done.

   ASSERT(worker_thread_ == rtc::Thread::Current());
 
-  set_transport_channel(session->CreateChannel(
-      content_name(), cricket::ICE_CANDIDATE_COMPONENT_RTP));
+  if (transport_name == transport_name_) {
+    // Nothing to do if transport name isn't changing
+    return true;
+  }
+
+  set_transport_channel(transport_controller_->CreateTransportChannel_w(
+      transport_name, cricket::ICE_CANDIDATE_COMPONENT_RTP));
   if (!transport_channel()) {
     return false;
   }
-  if (rtcp) {
-    set_rtcp_transport_channel(session->CreateChannel(
-        content_name(), cricket::ICE_CANDIDATE_COMPONENT_RTCP));
+  if (rtcp_transport_enabled()) {
+    LOG(LS_INFO) << "Create RTCP TransportChannel for " << content_name()
+                 << " on " << transport_name << " transport ";
+    set_rtcp_transport_channel(transport_controller_->CreateTransportChannel_w(
+        transport_name, cricket::ICE_CANDIDATE_COMPONENT_RTCP));
     if (!rtcp_transport_channel()) {
       return false;
     }
-  } else {
-    set_rtcp_transport_channel(nullptr);
   }
 
+  transport_name_ = transport_name;
   return true;
 }
 
 void BaseChannel::set_transport_channel(TransportChannel* new_tc) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   TransportChannel* old_tc = transport_channel_;
-
-  if (old_tc == new_tc) {
-    return;
-  }

[pthatcher1, 2015/08/31 22:01:35]

Sorry, can you remind me again why we don't return early if the channel didn't
change?  It seems like if the TC is the same, then all the work below would be
useless.

[Taylor Brandstetter, 2015/09/01 23:53:30]

It should now be impossible for "set_transport_channel" to be called with the
same channel. This was broken in the first place (if you recall the discussion
earlier about a dangling reference count), so returning early only ever worked
if both "old_tc" and "new_tc" are nullptr. I'll add that condition back in, and
add an ASSERT for the other, to make things more clear.

   if (old_tc) {
     DisconnectFromTransportChannel(old_tc);
-    session()->DestroyChannel(
-        content_name(), cricket::ICE_CANDIDATE_COMPONENT_RTP);
+    transport_controller_->DestroyTransportChannel_w(
+        transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTP);
   }
 
   transport_channel_ = new_tc;
 
   if (new_tc) {
     ConnectToTransportChannel(new_tc);
+    for (const auto& pair : socket_options_) {
+      new_tc->SetOption(pair.first, pair.second);
+    }
   }
+
+  // Update aggregate writable/ready-to-send state between RTP and RTCP upon
+  // setting new channel
+  UpdateWritableState_w();
+  SetReadyToSend(false, new_tc && new_tc->writable());
 }
 
 void BaseChannel::set_rtcp_transport_channel(TransportChannel* new_tc) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   TransportChannel* old_tc = rtcp_transport_channel_;
-
-  if (old_tc == new_tc) {
-    return;
-  }
   if (old_tc) {
     DisconnectFromTransportChannel(old_tc);
-    session()->DestroyChannel(
-        content_name(), cricket::ICE_CANDIDATE_COMPONENT_RTCP);
+    transport_controller_->DestroyTransportChannel_w(
+        transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTCP);
   }
 
   rtcp_transport_channel_ = new_tc;
 
   if (new_tc) {
     ConnectToTransportChannel(new_tc);
+    for (const auto& pair : rtcp_socket_options_) {
+      new_tc->SetOption(pair.first, pair.second);
+    }
   }
+
+  // Update aggregate writable/ready-to-send state between RTP and RTCP upon
+  // setting new channel
+  UpdateWritableState_w();
+  SetReadyToSend(true, new_tc && new_tc->writable());
 }
 
 void BaseChannel::ConnectToTransportChannel(TransportChannel* tc) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   tc->SignalWritableState.connect(this, &BaseChannel::OnWritableState);
   tc->SignalReadPacket.connect(this, &BaseChannel::OnChannelRead);
   tc->SignalReadyToSend.connect(this, &BaseChannel::OnReadyToSend);
 }
 
@@ skipping 98 matching lines @@
                            rtc::DiffServCodePoint dscp) {
   return SendPacket(true, packet, dscp);
 }
 
 int BaseChannel::SetOption(SocketType type, rtc::Socket::Option opt,
                            int value) {
   TransportChannel* channel = NULL;
   switch (type) {
     case ST_RTP:
       channel = transport_channel_;
+      socket_options_.push_back(
+          std::pair<rtc::Socket::Option, int>(opt, value));
       break;
     case ST_RTCP:
       channel = rtcp_transport_channel_;
+      rtcp_socket_options_.push_back(
+          std::pair<rtc::Socket::Option, int>(opt, value));
       break;
   }
   return channel ? channel->SetOption(opt, value) : -1;
 }
 
 void BaseChannel::OnWritableState(TransportChannel* channel) {
   ASSERT(channel == transport_channel_ || channel == rtcp_transport_channel_);
-  if (transport_channel_->writable()
-      && (!rtcp_transport_channel_ || rtcp_transport_channel_->writable())) {
-    ChannelWritable_w();
-  } else {
-    ChannelNotWritable_w();
-  }
+  UpdateWritableState_w();
 }
 
 void BaseChannel::OnChannelRead(TransportChannel* channel,
                                 const char* data, size_t len,
                                 const rtc::PacketTime& packet_time,
                                 int flags) {
   // OnChannelRead gets called from P2PSocket; now pass data to MediaEngine
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   // When using RTCP multiplexing we might get RTCP packets on the RTP
   // transport. We feed RTP traffic into the demuxer to determine if it is RTCP.
   bool rtcp = PacketIsRtcp(channel, data, len);
   rtc::Buffer packet(data, len);
   HandlePacket(rtcp, &packet, packet_time);
 }
 
 void BaseChannel::OnReadyToSend(TransportChannel* channel) {
-  SetReadyToSend(channel, true);
+  ASSERT(channel == transport_channel_ || channel == rtcp_transport_channel_);
+  SetReadyToSend(channel == rtcp_transport_channel_, true);
 }
 
-void BaseChannel::SetReadyToSend(TransportChannel* channel, bool ready) {
-  ASSERT(channel == transport_channel_ || channel == rtcp_transport_channel_);
-  if (channel == transport_channel_) {
+void BaseChannel::SetReadyToSend(bool rtcp, bool ready) {
+  if (rtcp) {
+    rtcp_ready_to_send_ = ready;
+  } else {
     rtp_ready_to_send_ = ready;
   }
-  if (channel == rtcp_transport_channel_) {
-    rtcp_ready_to_send_ = ready;
-  }
 
-  if (!ready) {
+  if (rtp_ready_to_send_ &&
+      // In the case of rtcp mux |rtcp_transport_channel_| will be null.
+      (rtcp_ready_to_send_ || !rtcp_transport_channel_)) {
+    // Notify the MediaChannel when both rtp and rtcp channel can send.
+    media_channel_->OnReadyToSend(true);
+  } else {
     // Notify the MediaChannel when either rtp or rtcp channel can't send.
     media_channel_->OnReadyToSend(false);
-  } else if (rtp_ready_to_send_ &&
-             // In the case of rtcp mux |rtcp_transport_channel_| will be null.
-             (rtcp_ready_to_send_ || !rtcp_transport_channel_)) {
-    // Notify the MediaChannel when both rtp and rtcp channel can send.
-    media_channel_->OnReadyToSend(true);
   }
 }
 
 bool BaseChannel::PacketIsRtcp(const TransportChannel* channel,
                                const char* data, size_t len) {
   return (channel == rtcp_transport_channel_ ||
           rtcp_mux_filter_.DemuxRtcp(data, static_cast<int>(len)));
 }
 
 bool BaseChannel::SendPacket(bool rtcp, rtc::Buffer* packet,
@@ skipping 101 matching lines @@
     return false;
   }
 
   // Bon voyage.
   int ret =
       channel->SendPacket(packet->data<char>(), packet->size(), options,
                           (secure() && secure_dtls()) ? PF_SRTP_BYPASS : 0);
   if (ret != static_cast<int>(packet->size())) {
     if (channel->GetError() == EWOULDBLOCK) {
       LOG(LS_WARNING) << "Got EWOULDBLOCK from socket.";
-      SetReadyToSend(channel, false);
+      SetReadyToSend(rtcp, false);
     }
     return false;
   }
   return true;
 }
 
 bool BaseChannel::WantsPacket(bool rtcp, rtc::Buffer* packet) {
   // Protect ourselves against crazy data.
   if (!ValidPacket(rtcp, packet)) {
     LOG(LS_ERROR) << "Dropping incoming " << content_name_ << " "
@@ skipping 130 matching lines @@
       muted_streams_.erase(ssrc);
   }
   return ret;
 }
 
 bool BaseChannel::IsStreamMuted_w(uint32 ssrc) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   return muted_streams_.find(ssrc) != muted_streams_.end();
 }
 
+void BaseChannel::UpdateWritableState_w() {
+  if (transport_channel_ && transport_channel_->writable() &&
+      (!rtcp_transport_channel_ || rtcp_transport_channel_->writable())) {
+    ChannelWritable_w();
+  } else {
+    ChannelNotWritable_w();
+  }
+}
+
 void BaseChannel::ChannelWritable_w() {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   if (writable_)
     return;
 
-  LOG(LS_INFO) << "Channel socket writable ("
-               << transport_channel_->content_name() << ", "
-               << transport_channel_->component() << ")"
+  LOG(LS_INFO) << "Channel writable (" << content_name_ << ")"
                << (was_ever_writable_ ? "" : " for the first time");
 
   std::vector<ConnectionInfo> infos;
   transport_channel_->GetStats(&infos);
   for (std::vector<ConnectionInfo>::const_iterator it = infos.begin();
        it != infos.end(); ++it) {
     if (it->best_connection) {
       LOG(LS_INFO) << "Using " << it->local_candidate.ToSensitiveString()
                    << "->" << it->remote_candidate.ToSensitiveString();
       break;
     }
   }
 
   // If we're doing DTLS-SRTP, now is the time.
   if (!was_ever_writable_ && ShouldSetupDtlsSrtp()) {
     if (!SetupDtlsSrtp(false)) {
-      SignalDtlsSetupFailure(this, false);
+      SignalDtlsSetupFailure_w(false);
       return;
     }
 
     if (rtcp_transport_channel_) {
       if (!SetupDtlsSrtp(true)) {
-        SignalDtlsSetupFailure(this, true);
+        SignalDtlsSetupFailure_w(true);
         return;
       }
     }
   }
 
   was_ever_writable_ = true;
   writable_ = true;
   ChangeState();
 }
 
@@ skipping 22 matching lines @@
 
 bool BaseChannel::ShouldSetupDtlsSrtp() const {
   return true;
 }
 
 // This function returns true if either DTLS-SRTP is not in use
 // *or* DTLS-SRTP is successfully set up.
 bool BaseChannel::SetupDtlsSrtp(bool rtcp_channel) {
   bool ret = false;
 
-  TransportChannel *channel = rtcp_channel ?
-      rtcp_transport_channel_ : transport_channel_;
+  TransportChannel* channel =
+      rtcp_channel ? rtcp_transport_channel_ : transport_channel_;
 
   // No DTLS
   if (!channel->IsDtlsActive())
     return true;
 
   std::string selected_cipher;
 
   if (!channel->GetSrtpCipher(&selected_cipher)) {
     LOG(LS_ERROR) << "No DTLS-SRTP selected cipher";
     return false;
@@ skipping 74 matching lines @@
     dtls_keyed_ = true;
 
   return ret;
 }
 
 void BaseChannel::ChannelNotWritable_w() {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   if (!writable_)
     return;
 
-  LOG(LS_INFO) << "Channel socket not writable ("
-               << transport_channel_->content_name() << ", "
-               << transport_channel_->component() << ")";
+  LOG(LS_INFO) << "Channel not writable (" << content_name_ << ")";
   writable_ = false;
   ChangeState();
 }
 
 bool BaseChannel::SetRtpTransportParameters_w(
     const MediaContentDescription* content,
     ContentAction action,
     ContentSource src,
     std::string* error_desc) {
   if (action == CA_UPDATE) {
@@ skipping 78 matching lines @@
 }
 
 void BaseChannel::ActivateRtcpMux() {
   worker_thread_->Invoke<void>(Bind(
       &BaseChannel::ActivateRtcpMux_w, this));
 }
 
 void BaseChannel::ActivateRtcpMux_w() {
   if (!rtcp_mux_filter_.IsActive()) {
     rtcp_mux_filter_.SetActive();
-    set_rtcp_transport_channel(NULL);
+    set_rtcp_transport_channel(nullptr);
+    rtcp_transport_enabled_ = false;
   }
 }
 
 bool BaseChannel::SetRtcpMux_w(bool enable, ContentAction action,
                                ContentSource src,
                                std::string* error_desc) {
   bool ret = false;
   switch (action) {
     case CA_OFFER:
       ret = rtcp_mux_filter_.SetOffer(enable, src);
       break;
     case CA_PRANSWER:
       ret = rtcp_mux_filter_.SetProvisionalAnswer(enable, src);
       break;
     case CA_ANSWER:
       ret = rtcp_mux_filter_.SetAnswer(enable, src);
       if (ret && rtcp_mux_filter_.IsActive()) {
         // We activated RTCP mux, close down the RTCP transport.
-        set_rtcp_transport_channel(NULL);
+        LOG(LS_INFO) << "Enabling rtcp-mux for " << content_name()
+                     << " by destroying RTCP transport channel for "
+                     << transport_name();
+        set_rtcp_transport_channel(nullptr);
+        rtcp_transport_enabled_ = false;
       }
       break;
     case CA_UPDATE:
       // No RTCP mux info.
       ret = true;
       break;
     default:
       break;
   }
   if (!ret) {
@@ skipping 206 matching lines @@
   worker_thread_->Clear(this, MSG_RTCPPACKET, &rtcp_messages);
   for (rtc::MessageList::iterator it = rtcp_messages.begin();
        it != rtcp_messages.end(); ++it) {
     worker_thread_->Send(this, MSG_RTCPPACKET, it->pdata);
   }
 }
 
 VoiceChannel::VoiceChannel(rtc::Thread* thread,
                            MediaEngineInterface* media_engine,
                            VoiceMediaChannel* media_channel,
-                           BaseSession* session,
+                           TransportController* transport_controller,
                            const std::string& content_name,
                            bool rtcp)
-    : BaseChannel(thread, media_channel, session, content_name,
+    : BaseChannel(thread,
+                  media_channel,
+                  transport_controller,
+                  content_name,
                   rtcp),
       media_engine_(media_engine),
       received_media_(false) {
 }
 
 VoiceChannel::~VoiceChannel() {
   StopAudioMonitor();
   StopMediaMonitor();
   // this can't be done in the base class, since it calls a virtual
   DisableMedia_w();
@@ skipping 375 matching lines @@
       break;
   }
 }
 
 void VoiceChannel::GetSrtpCiphers(std::vector<std::string>* ciphers) const {
   GetSupportedAudioCryptoSuites(ciphers);
 }
 
 VideoChannel::VideoChannel(rtc::Thread* thread,
                            VideoMediaChannel* media_channel,
-                           BaseSession* session,
+                           TransportController* transport_controller,
                            const std::string& content_name,
                            bool rtcp)
-    : BaseChannel(thread, media_channel, session, content_name,
+    : BaseChannel(thread,
+                  media_channel,
+                  transport_controller,
+                  content_name,
                   rtcp),
       renderer_(NULL),
       previous_we_(rtc::WE_CLOSE) {
 }
 
 bool VideoChannel::Init() {
   if (!BaseChannel::Init()) {
     return false;
   }
   media_channel()->SignalMediaError.connect(
@@ skipping 414 matching lines @@
   }
 }
 
 
 void VideoChannel::GetSrtpCiphers(std::vector<std::string>* ciphers) const {
   GetSupportedVideoCryptoSuites(ciphers);
 }
 
 DataChannel::DataChannel(rtc::Thread* thread,
                          DataMediaChannel* media_channel,
-                         BaseSession* session,
+                         TransportController* transport_controller,
                          const std::string& content_name,
                          bool rtcp)
-    : BaseChannel(thread, media_channel, session, content_name, rtcp),
+    : BaseChannel(thread,
+                  media_channel,
+                  transport_controller,
+                  content_name,
+                  rtcp),
       data_channel_type_(cricket::DCT_NONE),
       ready_to_send_data_(false) {
 }
 
 DataChannel::~DataChannel() {
   StopMediaMonitor();
   // this can't be done in the base class, since it calls a virtual
   DisableMedia_w();
 
   Deinit();
@@ skipping 317 matching lines @@
   return (data_channel_type_ == DCT_RTP);
 }
 
 void DataChannel::OnStreamClosedRemotely(uint32 sid) {
   rtc::TypedMessageData<uint32>* message =
       new rtc::TypedMessageData<uint32>(sid);
   signaling_thread()->Post(this, MSG_STREAMCLOSEDREMOTELY, message);
 }
 
 }  // namespace cricket