Added network thread to rtc::BaseChannel

webrtc/pc/channel.cc

 /*
  *  Copyright 2004 The WebRTC project authors. All Rights Reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
@@ skipping 19 matching lines @@
 using rtc::Bind;
 
 namespace {
 // See comment below for why we need to use a pointer to a unique_ptr.
 bool SetRawAudioSink_w(VoiceMediaChannel* channel,
                        uint32_t ssrc,
                        std::unique_ptr<webrtc::AudioSinkInterface>* sink) {
   channel->SetRawAudioSink(ssrc, std::move(*sink));
   return true;
 }
+
+struct SendingPacketMessageData : public rtc::MessageData {
+  rtc::CopyOnWriteBuffer packet;
+  rtc::PacketOptions options;
+};
+
+struct ReceivedPacketMessageData : public rtc::MessageData {
+  rtc::CopyOnWriteBuffer packet;
+  rtc::PacketTime packet_time;
+};
+
+struct ChangeState : public rtc::MessageData {
+  bool send;
+  bool recv;
+};
+
+struct NetworkRouteChanged : public rtc::MessageData {
+  std::string transport_name;
+  rtc::NetworkRoute network_route;
+};
+
 }  // namespace
 
 enum {
   MSG_EARLYMEDIATIMEOUT = 1,
-  MSG_RTPPACKET,
-  MSG_RTCPPACKET,
+  MSG_SENDING_RTP_PACKET,
+  MSG_SENDING_RTCP_PACKET,
   MSG_CHANNEL_ERROR,
   MSG_READYTOSENDDATA,
   MSG_DATARECEIVED,
   MSG_FIRSTPACKETRECEIVED,
   MSG_STREAMCLOSEDREMOTELY,
+  MSG_RECEIVED_RTP_PACKET,
+  MSG_RECEIVED_RTCP_PACKET,
+  MSG_CHANGE_STATE,
+  MSG_NOT_READY_TO_SEND,
+  MSG_READY_TO_SEND,
+  MSG_NETWORK_ROUTE_CHANGED,
 };
 
 // Value specified in RFC 5764.
 static const char kDtlsSrtpExporterLabel[] = "EXTRACTOR-dtls_srtp";
 
 static const int kAgcMinus10db = -10;
 
 static void SafeSetError(const std::string& message, std::string* error_desc) {
   if (error_desc) {
     *error_desc = message;
   }
 }
 
-struct PacketMessageData : public rtc::MessageData {
-  rtc::CopyOnWriteBuffer packet;
-  rtc::PacketOptions options;
-};
-
 struct VoiceChannelErrorMessageData : public rtc::MessageData {
   VoiceChannelErrorMessageData(uint32_t in_ssrc,
                                VoiceMediaChannel::Error in_error)
       : ssrc(in_ssrc), error(in_error) {}
   uint32_t ssrc;
   VoiceMediaChannel::Error error;
 };
 
 struct VideoChannelErrorMessageData : public rtc::MessageData {
   VideoChannelErrorMessageData(uint32_t in_ssrc,
@@ skipping 56 matching lines @@
 }
 
 template <class Codec>
 void RtpSendParametersFromMediaDescription(
     const MediaContentDescriptionImpl<Codec>* desc,
     RtpSendParameters<Codec>* send_params) {
   RtpParametersFromMediaDescription(desc, send_params);
   send_params->max_bandwidth_bps = desc->bandwidth();
 }
 
-BaseChannel::BaseChannel(rtc::Thread* thread,
+BaseChannel::BaseChannel(rtc::Thread* worker_thread,
+                         rtc::Thread* network_thread,
                          MediaChannel* media_channel,
                          TransportController* transport_controller,
                          const std::string& content_name,
                          bool rtcp)
-    : worker_thread_(thread),
+    : worker_thread_(worker_thread),
+      network_thread_(network_thread),
       transport_controller_(transport_controller),
       media_channel_(media_channel),
       content_name_(content_name),
       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());
+  if (transport_controller) {
+    RTC_DCHECK_EQ(network_thread, transport_controller->worker_thread());
+  }
   LOG(LS_INFO) << "Created channel for " << content_name;
 }
 
 BaseChannel::~BaseChannel() {
   TRACE_EVENT0("webrtc", "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_;
   // 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);
-  }
+  network_thread_->Invoke<void>([this] {

[pthatcher1, 2016/04/29 23:36:23]

It seems risky to have the BaseChannel destructor doing a synchronous call on
the worker thread. 

Is the worker thread even allowed to block on the network thread?  Is the
network thread allowed to block on the worker thread?  If they can both block on
each other, that could lead to deadlock.

[pthatcher1, 2016/04/29 23:36:23]

I don't think we're allowed to invoke lambdas because then they are stored in a
queue, which is against the chrome style guide, last I checked.  That's done
several times in this file.

[danilchap, 2016/05/02 14:50:34]

Wasn't sure if storing in sendlist queue was counted as storing: style reason
against lambdas because of object lifetime mistakes. With synchronous calls I
considered it is not an issue.
Lambdas Updated to rtc::Bind everywhere in this file.

Wonder if in unitests it is better to use Bind for consistency or lambdas for
readability.

[danilchap, 2016/05/02 14:50:34]

Worker can be blocked on network thread, but network thread can't be blocked on
worker thread. ChannelManager was updated to allow blocking calls on worker
thread and disallow them on network thread.

I assumed this is safe because that how ChannelManager is destructed: it does a
synchronous call on signaling thread to worker thread.

[pthatcher1, 2016/05/11 04:50:01]

That's a good point.  It's probably worth bringing up to webrtc-eng.  If we
decide that it's OK to make a synchronous call with lambdas, then we should do
that everywhere we make a synchronous call (and be consistent).  On the other
hand, some people want to get rid of all synchronous calls.  Would you like to
bring it up on webrtc-eng?
I think for unit tests, we can be a little more loose and use lambdas.  

[pthatcher1, 2016/05/11 04:50:01]

That's probably the only thing that makes sense.  But we should make a big
comment at the top of the file with a big warning about this, explaining what
can block on what and what _w and _n and _s mean.  

[danilchap, 2016/05/11 12:19:16]

Expanded comment above BaseChannel definition

+    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);
+    }
+    network_thread_->Clear(this);
+  });
   LOG(LS_INFO) << "Destroyed channel";
 }
 
 bool BaseChannel::Init() {
-  if (!SetTransport(content_name())) {
-    return false;
-  }
+  bool setup_transport = network_thread_->Invoke<bool>([this] {
+    if (!SetTransport_n(content_name())) {
+      return false;
+    }
 
-  if (!SetDtlsSrtpCryptoSuites(transport_channel(), false)) {
-    return false;
-  }
-  if (rtcp_transport_enabled() &&
-      !SetDtlsSrtpCryptoSuites(rtcp_transport_channel(), true)) {
+    if (!SetDtlsSrtpCryptoSuites(transport_channel(), false)) {

[pthatcher1, 2016/04/29 23:36:23]

I think SetDtlsSrtpCryptoSuites, rtcp_transport_enabled(), and
SetDtlsSrtpCryptoSuites can only be accessed on the signaling thread, not the
network thread.

[danilchap, 2016/05/02 14:50:34]

Old code access them on worker, not signaling thread.
So it seems legal to move this calls to network thread:
rtcp_transport_enabled set and accessed on network thread only (except
BaseChannel contructor)
SetDtlsSrtpCryptoSuites calls TransportChannel, I assume that is better be done
on network thread. SetDtlsSrtpCryptoSuites also call some static functions that
can be be called on any thread.

[pthatcher1, 2016/05/11 04:50:01]

Ah, true.  I was thrown off by Init().  It should be named Init_w().
Yes, you are correct.   We should call it SetDtlsSrtpCryptoSuites_n, then. 

SetDtlsSrtpCryptoSuites also call some static functions that

[danilchap, 2016/05/11 12:19:16]

Init renamed to Init_w, few more function got thread suffixes too.
Discovered few member accesses on wrong threads, added Invokes.

+      return false;
+    }
+    if (rtcp_transport_enabled() &&
+        !SetDtlsSrtpCryptoSuites(rtcp_transport_channel(), true)) {
+      return false;
+    }
+    return true;
+  });
+  if (!setup_transport) {
     return false;
   }
 
   // Both RTP and RTCP channels are set, we can call SetInterface on
   // media channel and it can set network options.
+  RTC_DCHECK(worker_thread_->IsCurrent());
   media_channel_->SetInterface(this);
   return true;
 }
 
 void BaseChannel::Deinit() {
+  RTC_DCHECK(worker_thread_->IsCurrent());
   media_channel_->SetInterface(NULL);
 }
 
 bool BaseChannel::SetTransport(const std::string& transport_name) {
-  return worker_thread_->Invoke<bool>(
-      Bind(&BaseChannel::SetTransport_w, this, transport_name));
+  return network_thread_->Invoke<bool>(
+      Bind(&BaseChannel::SetTransport_n, this, transport_name));
 }
 
-bool BaseChannel::SetTransport_w(const std::string& transport_name) {
-  ASSERT(worker_thread_ == rtc::Thread::Current());
+bool BaseChannel::SetTransport_n(const std::string& transport_name) {
+  RTC_DCHECK(network_thread_->IsCurrent());
 
   if (transport_name == transport_name_) {
     // Nothing to do if transport name isn't changing
     return true;
   }
 
   // When using DTLS-SRTP, we must reset the SrtpFilter every time the transport
   // changes and wait until the DTLS handshake is complete to set the newly
   // negotiated parameters.
   if (ShouldSetupDtlsSrtp()) {
@@ skipping 28 matching lines @@
     // We can only update the RTCP ready to send after set_transport_channel has
     // handled channel writability.
     SetReadyToSend(
         true, rtcp_transport_channel() && rtcp_transport_channel()->writable());
   }
   transport_name_ = transport_name;
   return true;
 }
 
 void BaseChannel::set_transport_channel(TransportChannel* new_tc) {
-  ASSERT(worker_thread_ == rtc::Thread::Current());
+  RTC_DCHECK(network_thread_->IsCurrent());
 
   TransportChannel* old_tc = transport_channel_;
   if (!old_tc && !new_tc) {
     // Nothing to do
     return;
   }
   ASSERT(old_tc != new_tc);
 
   if (old_tc) {
     DisconnectFromTransportChannel(old_tc);
     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();
+  UpdateWritableState_n();
   SetReadyToSend(false, new_tc && new_tc->writable());
 }
 
 void BaseChannel::set_rtcp_transport_channel(TransportChannel* new_tc,
                                              bool update_writablity) {
-  ASSERT(worker_thread_ == rtc::Thread::Current());
+  RTC_DCHECK(network_thread_->IsCurrent());
 
   TransportChannel* old_tc = rtcp_transport_channel_;
   if (!old_tc && !new_tc) {
     // Nothing to do
     return;
   }
   ASSERT(old_tc != new_tc);
 
   if (old_tc) {
     DisconnectFromTransportChannel(old_tc);
     transport_controller_->DestroyTransportChannel_w(
         transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTCP);
   }
 
   rtcp_transport_channel_ = new_tc;
 
   if (new_tc) {
     RTC_CHECK(!(ShouldSetupDtlsSrtp() && srtp_filter_.IsActive()))
         << "Setting RTCP for DTLS/SRTP after SrtpFilter is active "
         << "should never happen.";
     ConnectToTransportChannel(new_tc);
     for (const auto& pair : rtcp_socket_options_) {
       new_tc->SetOption(pair.first, pair.second);
     }
   }
 
   if (update_writablity) {
     // Update aggregate writable/ready-to-send state between RTP and RTCP upon
     // setting new channel
-    UpdateWritableState_w();
+    UpdateWritableState_n();
     SetReadyToSend(true, new_tc && new_tc->writable());
   }
 }
 
 void BaseChannel::ConnectToTransportChannel(TransportChannel* tc) {
-  ASSERT(worker_thread_ == rtc::Thread::Current());
+  RTC_DCHECK(network_thread_->IsCurrent());
 
   tc->SignalWritableState.connect(this, &BaseChannel::OnWritableState);
   tc->SignalReadPacket.connect(this, &BaseChannel::OnChannelRead);
   tc->SignalReadyToSend.connect(this, &BaseChannel::OnReadyToSend);
   tc->SignalDtlsState.connect(this, &BaseChannel::OnDtlsState);
   tc->SignalSelectedCandidatePairChanged.connect(
       this, &BaseChannel::OnSelectedCandidatePairChanged);
 }
 
 void BaseChannel::DisconnectFromTransportChannel(TransportChannel* tc) {
-  ASSERT(worker_thread_ == rtc::Thread::Current());
+  RTC_DCHECK(network_thread_->IsCurrent());
 
   tc->SignalWritableState.disconnect(this);
   tc->SignalReadPacket.disconnect(this);
   tc->SignalReadyToSend.disconnect(this);
   tc->SignalDtlsState.disconnect(this);
 }
 
 bool BaseChannel::Enable(bool enable) {
   worker_thread_->Invoke<void>(Bind(
       enable ? &BaseChannel::EnableMedia_w : &BaseChannel::DisableMedia_w,
@@ skipping 32 matching lines @@
                                    std::string* error_desc) {
   TRACE_EVENT0("webrtc", "BaseChannel::SetRemoteContent");
   return InvokeOnWorker(Bind(&BaseChannel::SetRemoteContent_w,
                              this, content, action, error_desc));
 }
 
 void BaseChannel::StartConnectionMonitor(int cms) {
   // We pass in the BaseChannel instead of the transport_channel_
   // because if the transport_channel_ changes, the ConnectionMonitor
   // would be pointing to the wrong TransportChannel.
-  connection_monitor_.reset(new ConnectionMonitor(
-      this, worker_thread(), rtc::Thread::Current()));
+  connection_monitor_.reset(
+      new ConnectionMonitor(this, network_thread(), rtc::Thread::Current()));

[pthatcher1, 2016/04/29 23:36:23]

We should comment that this is because GetConnectionStats will get called on
this thread, which needs to be the network thread.

[danilchap, 2016/05/02 14:50:34]

Done.

   connection_monitor_->SignalUpdate.connect(
       this, &BaseChannel::OnConnectionMonitorUpdate);
   connection_monitor_->Start(cms);
 }
 
 void BaseChannel::StopConnectionMonitor() {
   if (connection_monitor_) {
     connection_monitor_->Stop();
     connection_monitor_.reset();
   }
 }
 
 bool BaseChannel::GetConnectionStats(ConnectionInfos* infos) {
-  ASSERT(worker_thread_ == rtc::Thread::Current());
+  RTC_DCHECK(network_thread_->IsCurrent());
   return transport_channel_->GetStats(infos);
 }
 
 bool BaseChannel::IsReadyToReceive() const {
   // Receive data if we are enabled and have local content,
   return enabled() && IsReceiveContentDirection(local_content_direction_);
 }
 
 bool BaseChannel::IsReadyToSend() const {
   // Send outgoing data if we are enabled, have local and remote content,
   // and we have had some form of connectivity.
   return enabled() && IsReceiveContentDirection(remote_content_direction_) &&
          IsSendContentDirection(local_content_direction_) &&
          was_ever_writable() &&
          (srtp_filter_.IsActive() || !ShouldSetupDtlsSrtp());
 }
 
 bool BaseChannel::SendPacket(rtc::CopyOnWriteBuffer* packet,
                              const rtc::PacketOptions& options) {
   return SendPacket(false, packet, options);
 }
 
 bool BaseChannel::SendRtcp(rtc::CopyOnWriteBuffer* packet,
                            const rtc::PacketOptions& options) {
   return SendPacket(true, packet, options);
 }
 
 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;
+  return network_thread_->Invoke<int>([this, type, opt, value] {
+    TransportChannel* channel = nullptr;
+    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;

[pthatcher1, 2016/04/29 23:36:23]

Only this last part should be on the network thread.  The rest should be on the
signaling thread.  The old code is wrong, it appears, since it calls the last
part on the signaling thread, which isn't safe.

[danilchap, 2016/05/02 14:50:34]

transport_channels created/destroyed/set on network thread,
socket_options are used to set options on TransportChannel that live on network
thread,
why those two pairs of variables should be accessed on signaling thread?

[pthatcher1, 2016/05/11 04:50:01]

Ah, that's a good point.  transport_channel_ and rtcp_transport_channel_ should
only be accessed on the network thread.   I think the way you did this is the
right way.

+  });
 }
 
 void BaseChannel::OnWritableState(TransportChannel* channel) {
-  ASSERT(channel == transport_channel_ || channel == rtcp_transport_channel_);
-  UpdateWritableState_w();
+  RTC_DCHECK(channel == transport_channel_ ||
+             channel == rtcp_transport_channel_);
+  RTC_DCHECK(network_thread_->IsCurrent());
+  UpdateWritableState_n();
 }
 
 void BaseChannel::OnChannelRead(TransportChannel* channel,
                                 const char* data, size_t len,
                                 const rtc::PacketTime& packet_time,
                                 int flags) {
   TRACE_EVENT0("webrtc", "BaseChannel::OnChannelRead");
   // OnChannelRead gets called from P2PSocket; now pass data to MediaEngine
-  ASSERT(worker_thread_ == rtc::Thread::Current());
+  RTC_DCHECK(network_thread_->IsCurrent());
 
   // 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::CopyOnWriteBuffer packet(data, len);
   HandlePacket(rtcp, &packet, packet_time);
 }
 
 void BaseChannel::OnReadyToSend(TransportChannel* channel) {
   ASSERT(channel == transport_channel_ || channel == rtcp_transport_channel_);
@@ skipping 14 matching lines @@
   if (state != DTLS_TRANSPORT_CONNECTED) {
     srtp_filter_.ResetParams();
   }
 }
 
 void BaseChannel::OnSelectedCandidatePairChanged(
     TransportChannel* channel,
     CandidatePairInterface* selected_candidate_pair,
     int last_sent_packet_id) {
   ASSERT(channel == transport_channel_ || channel == rtcp_transport_channel_);
-  rtc::NetworkRoute network_route;
+  RTC_DCHECK(network_thread_->IsCurrent());
+  NetworkRouteChanged* message = new NetworkRouteChanged;
   if (selected_candidate_pair) {
-    network_route = rtc::NetworkRoute(
+    message->network_route = rtc::NetworkRoute(
         selected_candidate_pair->local_candidate().network_id(),
         selected_candidate_pair->remote_candidate().network_id(),
         last_sent_packet_id);
   }
-  media_channel()->OnNetworkRouteChanged(channel->transport_name(),
-                                         network_route);
+  message->transport_name = channel->transport_name();
+  worker_thread_->Post(this, MSG_NETWORK_ROUTE_CHANGED, message);

[pthatcher1, 2016/04/29 23:36:23]

Can't we do an AsyncInvoke on the worker_thread_ with
media_channel()->OnNetworkRouteChanged and avoid the message?

[danilchap, 2016/05/02 14:50:34]

Didn't know there was AsyncInvoke.
Done.

 }
 
 void BaseChannel::SetReadyToSend(bool rtcp, bool ready) {
+  RTC_DCHECK(network_thread_->IsCurrent());
   if (rtcp) {
     rtcp_ready_to_send_ = ready;
   } else {
     rtp_ready_to_send_ = 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);
-  }
+  bool ready_to_send =
+      (rtp_ready_to_send_ &&
+       // In the case of rtcp mux |rtcp_transport_channel_| will be null.
+       (rtcp_ready_to_send_ || !rtcp_transport_channel_));
+
+  worker_thread_->Post(
+      this, ready_to_send ? MSG_READY_TO_SEND : MSG_NOT_READY_TO_SEND);

[pthatcher1, 2016/04/29 23:36:23]

Same here.

[danilchap, 2016/05/02 14:50:34]

Done.

 }
 
 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::CopyOnWriteBuffer* packet,
                              const rtc::PacketOptions& options) {
-  // SendPacket gets called from MediaEngine, typically on an encoder thread.
-  // If the thread is not our worker thread, we will post to our worker
-  // so that the real work happens on our worker. This avoids us having to
+  // SendPacket gets called from MediaEngine, on a pacer or an encoder thread.
+  // If the thread is not our network thread, we will post to our network
+  // so that the real work happens on our network. This avoids us having to
   // synchronize access to all the pieces of the send path, including
   // SRTP and the inner workings of the transport channels.
   // The only downside is that we can't return a proper failure code if
   // needed. Since UDP is unreliable anyway, this should be a non-issue.
-  if (rtc::Thread::Current() != worker_thread_) {
+  if (!network_thread_->IsCurrent()) {
     // Avoid a copy by transferring the ownership of the packet data.
-    int message_id = (!rtcp) ? MSG_RTPPACKET : MSG_RTCPPACKET;
-    PacketMessageData* data = new PacketMessageData;
+    int message_id = rtcp ? MSG_SENDING_RTCP_PACKET : MSG_SENDING_RTP_PACKET;
+    SendingPacketMessageData* data = new SendingPacketMessageData;

[pthatcher1, 2016/04/29 23:36:23]

Why is this "Sending" instead of "Send".

[danilchap, 2016/05/02 14:50:34]

For no good reason. Changed.

     data->packet = std::move(*packet);
     data->options = options;
-    worker_thread_->Post(this, message_id, data);
+    network_thread_->Post(this, message_id, data);
     return true;
   }
+  TRACE_EVENT0("webrtc", "BaseChannel::SendPacket");
 
   // Now that we are on the correct thread, ensure we have a place to send this
   // packet before doing anything. (We might get RTCP packets that we don't
   // intend to send.) If we've negotiated RTCP mux, send RTCP over the RTP
   // transport.
   TransportChannel* channel = (!rtcp || rtcp_mux_filter_.IsActive()) ?
       transport_channel_ : rtcp_transport_channel_;
   if (!channel || !channel->writable()) {
     return false;
   }
 
   // Protect ourselves against crazy data.
   if (!ValidPacket(rtcp, packet)) {
     LOG(LS_ERROR) << "Dropping outgoing " << content_name_ << " "
                   << PacketType(rtcp)
                   << " packet: wrong size=" << packet->size();
     return false;
   }
 
   rtc::PacketOptions updated_options;
   updated_options = options;
   // Protect if needed.
   if (srtp_filter_.IsActive()) {
+    TRACE_EVENT0("webrtc", "SRTP Encode");
     bool res;
     uint8_t* data = packet->data();
     int len = static_cast<int>(packet->size());
     if (!rtcp) {
     // If ENABLE_EXTERNAL_AUTH flag is on then packet authentication is not done
     // inside libsrtp for a RTP packet. A external HMAC module will be writing
     // a fake HMAC value. This is ONLY done for a RTP packet.
     // Socket layer will update rtp sendtime extension header if present in
     // packet with current time before updating the HMAC.
 #if !defined(ENABLE_EXTERNAL_AUTH)
@@ skipping 47 matching lines @@
   } else if (secure_required_) {
     // This is a double check for something that supposedly can't happen.
     LOG(LS_ERROR) << "Can't send outgoing " << PacketType(rtcp)
                   << " packet when SRTP is inactive and crypto is required";
 
     ASSERT(false);
     return false;
   }
 
   // Bon voyage.
-  int ret =
-      channel->SendPacket(packet->data<char>(), packet->size(), updated_options,
-                          (secure() && secure_dtls()) ? PF_SRTP_BYPASS : 0);
+  int flags = (secure() && secure_dtls()) ? PF_SRTP_BYPASS : PF_NORMAL;
+  int ret = channel->SendPacket(packet->data<char>(), packet->size(),
+                                updated_options, flags);
   if (ret != static_cast<int>(packet->size())) {
     if (channel->GetError() == EWOULDBLOCK) {
       LOG(LS_WARNING) << "Got EWOULDBLOCK from socket.";
       SetReadyToSend(rtcp, false);
     }
     return false;
   }
   return true;
 }
 
 bool BaseChannel::WantsPacket(bool rtcp, const rtc::CopyOnWriteBuffer* packet) {
   // Protect ourselves against crazy data.
   if (!ValidPacket(rtcp, packet)) {
     LOG(LS_ERROR) << "Dropping incoming " << content_name_ << " "
                   << PacketType(rtcp)
                   << " packet: wrong size=" << packet->size();
     return false;
   }
   if (rtcp) {
     // Permit all (seemingly valid) RTCP packets.
     return true;
   }
   // Check whether we handle this payload.
   return bundle_filter_.DemuxPacket(packet->data(), packet->size());
 }
 
 void BaseChannel::HandlePacket(bool rtcp, rtc::CopyOnWriteBuffer* packet,
                                const rtc::PacketTime& packet_time) {
+  RTC_DCHECK(network_thread_->IsCurrent());
   if (!WantsPacket(rtcp, packet)) {
     return;
   }
 
   // We are only interested in the first rtp packet because that
   // indicates the media has started flowing.
   if (!has_received_packet_ && !rtcp) {
     has_received_packet_ = true;
     signaling_thread()->Post(this, MSG_FIRSTPACKETRECEIVED);
   }
 
   // Unprotect the packet, if needed.
   if (srtp_filter_.IsActive()) {
+    TRACE_EVENT0("webrtc", "SRTP Decode");
     char* data = packet->data<char>();
     int len = static_cast<int>(packet->size());
     bool res;
     if (!rtcp) {
       res = srtp_filter_.UnprotectRtp(data, len, &len);
       if (!res) {
         int seq_num = -1;
         uint32_t ssrc = 0;
         GetRtpSeqNum(data, len, &seq_num);
         GetRtpSsrc(data, len, &ssrc);
@@ skipping 23 matching lines @@
     //    channels, so we haven't yet extracted keys, even if DTLS did complete
     //    on the channel that the packets are being sent on. It's really good
     //    practice to wait for both RTP and RTCP to be good to go before sending
     //    media, to prevent weird failure modes, so it's fine for us to just eat
     //    packets here. This is all sidestepped if RTCP mux is used anyway.
     LOG(LS_WARNING) << "Can't process incoming " << PacketType(rtcp)
                     << " packet when SRTP is inactive and crypto is required";
     return;
   }
 
-  // Push it down to the media channel.
-  if (!rtcp) {
-    media_channel_->OnPacketReceived(packet, packet_time);
-  } else {
-    media_channel_->OnRtcpReceived(packet, packet_time);
-  }
+  ReceivedPacketMessageData* message_data = new ReceivedPacketMessageData;
+  message_data->packet = std::move(*packet);
+  message_data->packet_time = packet_time;
+  int message_id = rtcp ? MSG_RECEIVED_RTCP_PACKET : MSG_RECEIVED_RTP_PACKET;
+  worker_thread_->Post(this, message_id, message_data);

[pthatcher1, 2016/04/29 23:36:23]

Same here with AsyncInvoke

[danilchap, 2016/05/02 14:50:34]

Done.

 }
 
 bool BaseChannel::PushdownLocalDescription(
     const SessionDescription* local_desc, ContentAction action,
     std::string* error_desc) {
   const ContentInfo* content_info = GetFirstContent(local_desc);
   const MediaContentDescription* content_desc =
       GetContentDescription(content_info);
   if (content_desc && content_info && !content_info->rejected &&
       !SetLocalContent(content_desc, action, error_desc)) {
@@ skipping 17 matching lines @@
   return true;
 }
 
 void BaseChannel::EnableMedia_w() {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   if (enabled_)
     return;
 
   LOG(LS_INFO) << "Channel enabled";
   enabled_ = true;
-  ChangeState();
+  ChangeState_w();
 }
 
 void BaseChannel::DisableMedia_w() {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   if (!enabled_)
     return;
 
   LOG(LS_INFO) << "Channel disabled";
   enabled_ = false;
-  ChangeState();
+  ChangeState_w();
 }
 
-void BaseChannel::UpdateWritableState_w() {
+void BaseChannel::UpdateWritableState_n() {
   if (transport_channel_ && transport_channel_->writable() &&
       (!rtcp_transport_channel_ || rtcp_transport_channel_->writable())) {
-    ChannelWritable_w();
+    ChannelWritable_n();
   } else {
-    ChannelNotWritable_w();
+    ChannelNotWritable_n();
   }
 }
 
-void BaseChannel::ChannelWritable_w() {
-  ASSERT(worker_thread_ == rtc::Thread::Current());
+void BaseChannel::ChannelWritable_n() {
+  RTC_DCHECK(network_thread_->IsCurrent());
   if (writable_) {
     return;
   }
 
   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;
     }
   }
 
   was_ever_writable_ = true;
-  MaybeSetupDtlsSrtp_w();
+  MaybeSetupDtlsSrtp_n();
   writable_ = true;
   ChangeState();
 }
 
-void BaseChannel::SignalDtlsSetupFailure_w(bool rtcp) {
-  ASSERT(worker_thread() == rtc::Thread::Current());
+void BaseChannel::SignalDtlsSetupFailure_n(bool rtcp) {
+  RTC_DCHECK(network_thread_->IsCurrent());
+  RTC_NOTREACHED();

[pthatcher1, 2016/04/29 23:36:23]

This isn't reached?

[danilchap, 2016/05/02 14:50:34]

In practice it isn't reached.

In theory it can be reached, but then old code would do a disallowed blocking
call on worker thread.
New code does disallowed blocking call on network thread.

Fixed with AsyncInvoke instead of commenting out.

+  // TODO(danilchap): Not allowed to invoke from network thread. Post instead.
   signaling_thread()->Invoke<void>(Bind(
       &BaseChannel::SignalDtlsSetupFailure_s, this, rtcp));
 }
 
 void BaseChannel::SignalDtlsSetupFailure_s(bool rtcp) {
   ASSERT(signaling_thread() == rtc::Thread::Current());
   SignalDtlsSetupFailure(this, rtcp);
 }
 
 bool BaseChannel::SetDtlsSrtpCryptoSuites(TransportChannel* tc, bool rtcp) {
   std::vector<int> crypto_suites;
   // We always use the default SRTP crypto suites for RTCP, but we may use
   // different crypto suites for RTP depending on the media type.
   if (!rtcp) {
     GetSrtpCryptoSuites(&crypto_suites);
   } else {
     GetDefaultSrtpCryptoSuites(&crypto_suites);
   }
   return tc->SetSrtpCryptoSuites(crypto_suites);
 }
 
 bool BaseChannel::ShouldSetupDtlsSrtp() const {
   // Since DTLS is applied to all channels, checking RTP should be enough.
   return transport_channel_ && transport_channel_->IsDtlsActive();
 }
 
 // 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 BaseChannel::SetupDtlsSrtp_n(bool rtcp_channel) {
+  RTC_DCHECK(network_thread_->IsCurrent());
   bool ret = false;
 
   TransportChannel* channel =
       rtcp_channel ? rtcp_transport_channel_ : transport_channel_;
 
   RTC_DCHECK(channel->IsDtlsActive());
 
   int selected_crypto_suite;
 
   if (!channel->GetSrtpCryptoSuite(&selected_crypto_suite)) {
@@ skipping 65 matching lines @@
   }
 
   if (!ret)
     LOG(LS_WARNING) << "DTLS-SRTP key installation failed";
   else
     dtls_keyed_ = true;
 
   return ret;
 }
 
-void BaseChannel::MaybeSetupDtlsSrtp_w() {
+void BaseChannel::MaybeSetupDtlsSrtp_n() {
   if (srtp_filter_.IsActive()) {
     return;
   }
 
   if (!ShouldSetupDtlsSrtp()) {

[pthatcher1, 2016/04/29 23:36:23]

This should be renamed to ShouldSetupDtlsSrtp_n.

[danilchap, 2016/05/02 14:50:34]

Done.

     return;
   }
 
-  if (!SetupDtlsSrtp(false)) {
-    SignalDtlsSetupFailure_w(false);
+  if (!SetupDtlsSrtp_n(false)) {
+    SignalDtlsSetupFailure_n(false);
     return;
   }
 
   if (rtcp_transport_channel_) {
-    if (!SetupDtlsSrtp(true)) {
-      SignalDtlsSetupFailure_w(true);
+    if (!SetupDtlsSrtp_n(true)) {
+      SignalDtlsSetupFailure_n(true);
       return;
     }
   }
 }
 
-void BaseChannel::ChannelNotWritable_w() {
-  ASSERT(worker_thread_ == rtc::Thread::Current());
+void BaseChannel::ChannelNotWritable_n() {
+  RTC_DCHECK(network_thread_->IsCurrent());
   if (!writable_)
     return;
 
   LOG(LS_INFO) << "Channel not writable (" << content_name_ << ")";
   writable_ = false;
   ChangeState();
 }
 
-bool BaseChannel::SetRtpTransportParameters_w(
+bool BaseChannel::SetRtpTransportParameters(
     const MediaContentDescription* content,
     ContentAction action,
     ContentSource src,
     std::string* error_desc) {
   if (action == CA_UPDATE) {
     // These parameters never get changed by a CA_UDPATE.
     return true;
   }
 
   // Cache secure_required_ for belt and suspenders check on SendPacket
-  if (src == CS_LOCAL) {
-    set_secure_required(content->crypto_required() != CT_NONE);
-  }
+  return network_thread_->Invoke<bool>(
+      [this, content, action, src, error_desc] {
+        if (src == CS_LOCAL) {
+          set_secure_required(content->crypto_required() != CT_NONE);
+        }

[pthatcher1, 2016/04/29 23:36:23]

I think set_secure_required needs to be on the signaling thread.

[danilchap, 2016/05/02 14:50:34]

Why?
secure_required accessor is used in tests only,
secure_required_ variable used on send and receive paths on network thread.

 
-  if (!SetSrtp_w(content->cryptos(), action, src, error_desc)) {
-    return false;
-  }
+        if (!SetSrtp_n(content->cryptos(), action, src, error_desc)) {
+          return false;
+        }
 
-  if (!SetRtcpMux_w(content->rtcp_mux(), action, src, error_desc)) {
-    return false;
-  }
+        if (!SetRtcpMux_n(content->rtcp_mux(), action, src, error_desc)) {
+          return false;
+        }
 
-  return true;
+        return true;
+      });
 }
 
 // |dtls| will be set to true if DTLS is active for transport channel and
 // crypto is empty.
 bool BaseChannel::CheckSrtpConfig(const std::vector<CryptoParams>& cryptos,
                                   bool* dtls,
                                   std::string* error_desc) {
   *dtls = transport_channel_->IsDtlsActive();
   if (*dtls && !cryptos.empty()) {
     SafeSetError("Cryptos must be empty when DTLS is active.",
                  error_desc);
     return false;
   }
   return true;
 }
 
-bool BaseChannel::SetSrtp_w(const std::vector<CryptoParams>& cryptos,
+bool BaseChannel::SetSrtp_n(const std::vector<CryptoParams>& cryptos,
                             ContentAction action,
                             ContentSource src,
                             std::string* error_desc) {
   TRACE_EVENT0("webrtc", "BaseChannel::SetSrtp_w");
   if (action == CA_UPDATE) {
     // no crypto params.
     return true;
   }
   bool ret = false;
   bool dtls = false;
@@ skipping 27 matching lines @@
       break;
   }
   if (!ret) {
     SafeSetError("Failed to setup SRTP filter.", error_desc);
     return false;
   }
   return true;
 }
 
 void BaseChannel::ActivateRtcpMux() {
-  worker_thread_->Invoke<void>(Bind(
-      &BaseChannel::ActivateRtcpMux_w, this));
+  network_thread_->Invoke<void>(Bind(&BaseChannel::ActivateRtcpMux_n, this));
 }
 
-void BaseChannel::ActivateRtcpMux_w() {
+void BaseChannel::ActivateRtcpMux_n() {
   if (!rtcp_mux_filter_.IsActive()) {
     rtcp_mux_filter_.SetActive();
     set_rtcp_transport_channel(nullptr, true);
     rtcp_transport_enabled_ = false;
   }
 }
 
-bool BaseChannel::SetRtcpMux_w(bool enable, ContentAction action,
+bool BaseChannel::SetRtcpMux_n(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;
@@ skipping 18 matching lines @@
   if (!ret) {
     SafeSetError("Failed to setup RTCP mux filter.", error_desc);
     return false;
   }
   // |rtcp_mux_filter_| can be active if |action| is CA_PRANSWER or
   // CA_ANSWER, but we only want to tear down the RTCP transport channel if we
   // received a final answer.
   if (rtcp_mux_filter_.IsActive()) {
     // If the RTP transport is already writable, then so are we.
     if (transport_channel_->writable()) {
-      ChannelWritable_w();
+      ChannelWritable_n();
     }
   }
 
   return true;
 }
 
 bool BaseChannel::AddRecvStream_w(const StreamParams& sp) {
   ASSERT(worker_thread() == rtc::Thread::Current());
   return media_channel()->AddRecvStream(sp);
 }
@@ skipping 154 matching lines @@
     const std::vector<RtpHeaderExtension>& extensions) {
   const RtpHeaderExtension* send_time_extension =
       FindHeaderExtension(extensions, kRtpAbsoluteSenderTimeHeaderExtension);
   rtp_abs_sendtime_extn_id_ =
       send_time_extension ? send_time_extension->id : -1;
 }
 
 void BaseChannel::OnMessage(rtc::Message *pmsg) {
   TRACE_EVENT0("webrtc", "BaseChannel::OnMessage");
   switch (pmsg->message_id) {
-    case MSG_RTPPACKET:
-    case MSG_RTCPPACKET: {
-      PacketMessageData* data = static_cast<PacketMessageData*>(pmsg->pdata);
-      SendPacket(pmsg->message_id == MSG_RTCPPACKET, &data->packet,
-                 data->options);
-      delete data;  // because it is Posted
+    case MSG_SENDING_RTP_PACKET:
+    case MSG_SENDING_RTCP_PACKET: {
+      RTC_DCHECK(network_thread_->IsCurrent());
+      SendingPacketMessageData* data =
+          static_cast<SendingPacketMessageData*>(pmsg->pdata);
+      bool rtcp = pmsg->message_id == MSG_SENDING_RTCP_PACKET;
+      SendPacket(rtcp, &data->packet, data->options);
+      delete data;
       break;
     }
     case MSG_FIRSTPACKETRECEIVED: {
       SignalFirstPacketReceived(this);
       break;
     }
+    case MSG_RECEIVED_RTP_PACKET: {
+      RTC_DCHECK(worker_thread_->IsCurrent());
+      ReceivedPacketMessageData* data =
+          static_cast<ReceivedPacketMessageData*>(pmsg->pdata);
+      media_channel_->OnPacketReceived(&data->packet, data->packet_time);
+      delete data;
+      break;
+    }
+    case MSG_RECEIVED_RTCP_PACKET: {
+      RTC_DCHECK(worker_thread_->IsCurrent());
+      ReceivedPacketMessageData* data =
+          static_cast<ReceivedPacketMessageData*>(pmsg->pdata);
+      media_channel_->OnRtcpReceived(&data->packet, data->packet_time);
+      delete data;
+      break;
+    }
+    case MSG_CHANGE_STATE: {
+      RTC_DCHECK(worker_thread_->IsCurrent());
+      ChangeState_w();
+      break;
+    }
+    case MSG_READY_TO_SEND: {
+      RTC_DCHECK(worker_thread_->IsCurrent());
+      media_channel_->OnReadyToSend(true);
+      break;
+    }
+    case MSG_NOT_READY_TO_SEND: {
+      RTC_DCHECK(worker_thread_->IsCurrent());
+      media_channel_->OnReadyToSend(false);
+      break;
+    }
+    case MSG_NETWORK_ROUTE_CHANGED: {
+      RTC_DCHECK(worker_thread_->IsCurrent());
+      NetworkRouteChanged* data =
+          static_cast<NetworkRouteChanged*>(pmsg->pdata);
+      media_channel()->OnNetworkRouteChanged(data->transport_name,
+                                             data->network_route);
+      delete data;
+      break;
+    }
   }
 }
 
 void BaseChannel::FlushRtcpMessages() {
   // Flush all remaining RTCP messages. This should only be called in
   // destructor.
-  ASSERT(rtc::Thread::Current() == worker_thread_);
-  rtc::MessageList rtcp_messages;
-  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);
-  }
+  network_thread_->Invoke<void>([this] {
+    rtc::MessageList rtcp_messages;
+    network_thread_->Clear(this, MSG_SENDING_RTCP_PACKET, &rtcp_messages);
+    for (const auto& message : rtcp_messages) {
+      network_thread_->Send(this, MSG_SENDING_RTCP_PACKET, message.pdata);
+    }
+  });
 }
 
-VoiceChannel::VoiceChannel(rtc::Thread* thread,
+VoiceChannel::VoiceChannel(rtc::Thread* worker_thread,
+                           rtc::Thread* network_thread,
                            MediaEngineInterface* media_engine,
                            VoiceMediaChannel* media_channel,
                            TransportController* transport_controller,
                            const std::string& content_name,
                            bool rtcp)
-    : BaseChannel(thread,
+    : BaseChannel(worker_thread,
+                  network_thread,
                   media_channel,
                   transport_controller,
                   content_name,
                   rtcp),
       media_engine_(media_engine),
       received_media_(false) {}
 
 VoiceChannel::~VoiceChannel() {
   TRACE_EVENT0("webrtc", "VoiceChannel::~VoiceChannel");
   StopAudioMonitor();
@@ skipping 137 matching lines @@
                                 int flags) {
   BaseChannel::OnChannelRead(channel, data, len, packet_time, flags);
 
   // Set a flag when we've received an RTP packet. If we're waiting for early
   // media, this will disable the timeout.
   if (!received_media_ && !PacketIsRtcp(channel, data, len)) {
     received_media_ = true;
   }
 }
 
-void VoiceChannel::ChangeState() {
+void BaseChannel::ChangeState() {
+  RTC_DCHECK(network_thread_->IsCurrent());
+  worker_thread_->Post(this, MSG_CHANGE_STATE);
+}
+
+void VoiceChannel::ChangeState_w() {
   // Render incoming data if we're the active call, and we have the local
   // content. We receive data on the default channel and multiplexed streams.
   bool recv = IsReadyToReceive();
   media_channel()->SetPlayout(recv);
 
   // Send outgoing data if we're the active call, we have the remote content,
   // and we have had some form of connectivity.
   bool send = IsReadyToSend();
   media_channel()->SetSend(send);
 
@@ skipping 13 matching lines @@
   LOG(LS_INFO) << "Setting local voice description";
 
   const AudioContentDescription* audio =
       static_cast<const AudioContentDescription*>(content);
   ASSERT(audio != NULL);
   if (!audio) {
     SafeSetError("Can't find audio content in local description.", error_desc);
     return false;
   }
 
-  if (!SetRtpTransportParameters_w(content, action, CS_LOCAL, error_desc)) {
+  if (!SetRtpTransportParameters(content, action, CS_LOCAL, error_desc)) {
     return false;
   }
 
   AudioRecvParameters recv_params = last_recv_params_;
   RtpParametersFromMediaDescription(audio, &recv_params);
   if (!media_channel()->SetRecvParameters(recv_params)) {
     SafeSetError("Failed to set local audio description recv parameters.",
                  error_desc);
     return false;
   }
   for (const AudioCodec& codec : audio->codecs()) {
     bundle_filter()->AddPayloadType(codec.id);
   }
   last_recv_params_ = recv_params;
 
   // TODO(pthatcher): Move local streams into AudioSendParameters, and
   // only give it to the media channel once we have a remote
   // description too (without a remote description, we won't be able
   // to send them anyway).
   if (!UpdateLocalStreams_w(audio->streams(), action, error_desc)) {
     SafeSetError("Failed to set local audio description streams.", error_desc);
     return false;
   }
 
   set_local_content_direction(content->direction());
-  ChangeState();
+  ChangeState_w();
   return true;
 }
 
 bool VoiceChannel::SetRemoteContent_w(const MediaContentDescription* content,
                                       ContentAction action,
                                       std::string* error_desc) {
   TRACE_EVENT0("webrtc", "VoiceChannel::SetRemoteContent_w");
   ASSERT(worker_thread() == rtc::Thread::Current());
   LOG(LS_INFO) << "Setting remote voice description";
 
   const AudioContentDescription* audio =
       static_cast<const AudioContentDescription*>(content);
   ASSERT(audio != NULL);
   if (!audio) {
     SafeSetError("Can't find audio content in remote description.", error_desc);
     return false;
   }
 
-  if (!SetRtpTransportParameters_w(content, action, CS_REMOTE, error_desc)) {
+  if (!SetRtpTransportParameters(content, action, CS_REMOTE, error_desc)) {
     return false;
   }
 
   AudioSendParameters send_params = last_send_params_;
   RtpSendParametersFromMediaDescription(audio, &send_params);
   if (audio->agc_minus_10db()) {
     send_params.options.adjust_agc_delta = rtc::Optional<int>(kAgcMinus10db);
   }
 
   bool parameters_applied = media_channel()->SetSendParameters(send_params);
@@ skipping 11 matching lines @@
   if (!UpdateRemoteStreams_w(audio->streams(), action, error_desc)) {
     SafeSetError("Failed to set remote audio description streams.", error_desc);
     return false;
   }
 
   if (audio->rtp_header_extensions_set()) {
     MaybeCacheRtpAbsSendTimeHeaderExtension(audio->rtp_header_extensions());
   }
 
   set_remote_content_direction(content->direction());
-  ChangeState();
+  ChangeState_w();
   return true;
 }
 
 void VoiceChannel::HandleEarlyMediaTimeout() {
   // This occurs on the main thread, not the worker thread.
   if (!received_media_) {
     LOG(LS_INFO) << "No early media received before timeout";
     SignalEarlyMediaTimeout(this);
   }
 }
@@ skipping 37 matching lines @@
 
 void VoiceChannel::OnAudioMonitorUpdate(AudioMonitor* monitor,
                                         const AudioInfo& info) {
   SignalAudioMonitor(this, info);
 }
 
 void VoiceChannel::GetSrtpCryptoSuites(std::vector<int>* crypto_suites) const {
   GetSupportedAudioCryptoSuites(crypto_suites);
 }
 
-VideoChannel::VideoChannel(rtc::Thread* thread,
+VideoChannel::VideoChannel(rtc::Thread* worker_thread,
+                           rtc::Thread* network_thread,
                            VideoMediaChannel* media_channel,
                            TransportController* transport_controller,
                            const std::string& content_name,
                            bool rtcp)
-    : BaseChannel(thread,
+    : BaseChannel(worker_thread,
+                  network_thread,
                   media_channel,
                   transport_controller,
                   content_name,
                   rtcp) {}
 
 bool VideoChannel::Init() {
   if (!BaseChannel::Init()) {
     return false;
   }
   return true;
@@ skipping 41 matching lines @@
 bool VideoChannel::SetRtpParameters(uint32_t ssrc,
                                     const webrtc::RtpParameters& parameters) {
   return InvokeOnWorker(
       Bind(&VideoChannel::SetRtpParameters_w, this, ssrc, parameters));
 }
 
 bool VideoChannel::SetRtpParameters_w(uint32_t ssrc,
                                       webrtc::RtpParameters parameters) {
   return media_channel()->SetRtpParameters(ssrc, parameters);
 }
-void VideoChannel::ChangeState() {
+
+void VideoChannel::ChangeState_w() {
   // Send outgoing data if we're the active call, we have the remote content,
   // and we have had some form of connectivity.
   bool send = IsReadyToSend();
   if (!media_channel()->SetSend(send)) {
     LOG(LS_ERROR) << "Failed to SetSend on video channel";
     // TODO(gangji): Report error back to server.
   }
 
   LOG(LS_INFO) << "Changing video state, send=" << send;
 }
@@ skipping 31 matching lines @@
   LOG(LS_INFO) << "Setting local video description";
 
   const VideoContentDescription* video =
       static_cast<const VideoContentDescription*>(content);
   ASSERT(video != NULL);
   if (!video) {
     SafeSetError("Can't find video content in local description.", error_desc);
     return false;
   }
 
-  if (!SetRtpTransportParameters_w(content, action, CS_LOCAL, error_desc)) {
+  if (!SetRtpTransportParameters(content, action, CS_LOCAL, error_desc)) {
     return false;
   }
 
   VideoRecvParameters recv_params = last_recv_params_;
   RtpParametersFromMediaDescription(video, &recv_params);
   if (!media_channel()->SetRecvParameters(recv_params)) {
     SafeSetError("Failed to set local video description recv parameters.",
                  error_desc);
     return false;
   }
   for (const VideoCodec& codec : video->codecs()) {
     bundle_filter()->AddPayloadType(codec.id);
   }
   last_recv_params_ = recv_params;
 
   // TODO(pthatcher): Move local streams into VideoSendParameters, and
   // only give it to the media channel once we have a remote
   // description too (without a remote description, we won't be able
   // to send them anyway).
   if (!UpdateLocalStreams_w(video->streams(), action, error_desc)) {
     SafeSetError("Failed to set local video description streams.", error_desc);
     return false;
   }
 
   set_local_content_direction(content->direction());
-  ChangeState();
+  ChangeState_w();
   return true;
 }
 
 bool VideoChannel::SetRemoteContent_w(const MediaContentDescription* content,
                                       ContentAction action,
                                       std::string* error_desc) {
   TRACE_EVENT0("webrtc", "VideoChannel::SetRemoteContent_w");
   ASSERT(worker_thread() == rtc::Thread::Current());
   LOG(LS_INFO) << "Setting remote video description";
 
   const VideoContentDescription* video =
       static_cast<const VideoContentDescription*>(content);
   ASSERT(video != NULL);
   if (!video) {
     SafeSetError("Can't find video content in remote description.", error_desc);
     return false;
   }
 
-
-  if (!SetRtpTransportParameters_w(content, action, CS_REMOTE, error_desc)) {
+  if (!SetRtpTransportParameters(content, action, CS_REMOTE, error_desc)) {
     return false;
   }
 
   VideoSendParameters send_params = last_send_params_;
   RtpSendParametersFromMediaDescription(video, &send_params);
   if (video->conference_mode()) {
     send_params.conference_mode = true;
   }
 
   bool parameters_applied = media_channel()->SetSendParameters(send_params);
@@ skipping 12 matching lines @@
   if (!UpdateRemoteStreams_w(video->streams(), action, error_desc)) {
     SafeSetError("Failed to set remote video description streams.", error_desc);
     return false;
   }
 
   if (video->rtp_header_extensions_set()) {
     MaybeCacheRtpAbsSendTimeHeaderExtension(video->rtp_header_extensions());
   }
 
   set_remote_content_direction(content->direction());
-  ChangeState();
+  ChangeState_w();
   return true;
 }
 
 void VideoChannel::OnMessage(rtc::Message *pmsg) {
   switch (pmsg->message_id) {
     case MSG_CHANNEL_ERROR: {
       const VideoChannelErrorMessageData* data =
           static_cast<VideoChannelErrorMessageData*>(pmsg->pdata);
       delete data;
       break;
@@ skipping 14 matching lines @@
 void VideoChannel::OnMediaMonitorUpdate(
     VideoMediaChannel* media_channel, const VideoMediaInfo &info) {
   ASSERT(media_channel == this->media_channel());
   SignalMediaMonitor(this, info);
 }
 
 void VideoChannel::GetSrtpCryptoSuites(std::vector<int>* crypto_suites) const {
   GetSupportedVideoCryptoSuites(crypto_suites);
 }
 
-DataChannel::DataChannel(rtc::Thread* thread,
+DataChannel::DataChannel(rtc::Thread* worker_thread,
+                         rtc::Thread* network_thread,
                          DataMediaChannel* media_channel,
                          TransportController* transport_controller,
                          const std::string& content_name,
                          bool rtcp)
-    : BaseChannel(thread,
+    : BaseChannel(worker_thread,
+                  network_thread,
                   media_channel,
                   transport_controller,
                   content_name,
                   rtcp),
       data_channel_type_(cricket::DCT_NONE),
       ready_to_send_data_(false) {}
 
 DataChannel::~DataChannel() {
   TRACE_EVENT0("webrtc", "DataChannel::~DataChannel");
   StopMediaMonitor();
@@ skipping 83 matching lines @@
   if (!data) {
     SafeSetError("Can't find data content in local description.", error_desc);
     return false;
   }
 
   if (!SetDataChannelTypeFromContent(data, error_desc)) {
     return false;
   }
 
   if (data_channel_type_ == DCT_RTP) {
-    if (!SetRtpTransportParameters_w(content, action, CS_LOCAL, error_desc)) {
+    if (!SetRtpTransportParameters(content, action, CS_LOCAL, error_desc)) {
       return false;
     }
   }
 
   // FYI: We send the SCTP port number (not to be confused with the
   // underlying UDP port number) as a codec parameter.  So even SCTP
   // data channels need codecs.
   DataRecvParameters recv_params = last_recv_params_;
   RtpParametersFromMediaDescription(data, &recv_params);
   if (!media_channel()->SetRecvParameters(recv_params)) {
@@ skipping 11 matching lines @@
   // TODO(pthatcher): Move local streams into DataSendParameters, and
   // only give it to the media channel once we have a remote
   // description too (without a remote description, we won't be able
   // to send them anyway).
   if (!UpdateLocalStreams_w(data->streams(), action, error_desc)) {
     SafeSetError("Failed to set local data description streams.", error_desc);
     return false;
   }
 
   set_local_content_direction(content->direction());
-  ChangeState();
+  ChangeState_w();
   return true;
 }
 
 bool DataChannel::SetRemoteContent_w(const MediaContentDescription* content,
                                      ContentAction action,
                                      std::string* error_desc) {
   TRACE_EVENT0("webrtc", "DataChannel::SetRemoteContent_w");
   ASSERT(worker_thread() == rtc::Thread::Current());
 
   const DataContentDescription* data =
       static_cast<const DataContentDescription*>(content);
   ASSERT(data != NULL);
   if (!data) {
     SafeSetError("Can't find data content in remote description.", error_desc);
     return false;
   }
 
   // If the remote data doesn't have codecs and isn't an update, it
   // must be empty, so ignore it.
   if (!data->has_codecs() && action != CA_UPDATE) {
     return true;
   }
 
   if (!SetDataChannelTypeFromContent(data, error_desc)) {
     return false;
   }
 
   LOG(LS_INFO) << "Setting remote data description";
   if (data_channel_type_ == DCT_RTP &&
-      !SetRtpTransportParameters_w(content, action, CS_REMOTE, error_desc)) {
+      !SetRtpTransportParameters(content, action, CS_REMOTE, error_desc)) {
     return false;
   }
 
 
   DataSendParameters send_params = last_send_params_;
   RtpSendParametersFromMediaDescription<DataCodec>(data, &send_params);
   if (!media_channel()->SetSendParameters(send_params)) {
     SafeSetError("Failed to set remote data description send parameters.",
                  error_desc);
     return false;
   }
   last_send_params_ = send_params;
 
   // TODO(pthatcher): Move remote streams into DataRecvParameters,
   // and only give it to the media channel once we have a local
   // description too (without a local description, we won't be able to
   // recv them anyway).
   if (!UpdateRemoteStreams_w(data->streams(), action, error_desc)) {
     SafeSetError("Failed to set remote data description streams.",
                  error_desc);
     return false;
   }
 
   set_remote_content_direction(content->direction());
-  ChangeState();
+  ChangeState_w();
   return true;
 }
 
-void DataChannel::ChangeState() {
+void DataChannel::ChangeState_w() {
   // Render incoming data if we're the active call, and we have the local
   // content. We receive data on the default channel and multiplexed streams.
   bool recv = IsReadyToReceive();
   if (!media_channel()->SetReceive(recv)) {
     LOG(LS_ERROR) << "Failed to SetReceive on data channel";
   }
 
   // Send outgoing data if we're the active call, we have the remote content,
   // and we have had some form of connectivity.
   bool send = IsReadyToSend();
@@ skipping 100 matching lines @@
   return (data_channel_type_ == DCT_RTP) && BaseChannel::ShouldSetupDtlsSrtp();
 }
 
 void DataChannel::OnStreamClosedRemotely(uint32_t sid) {
   rtc::TypedMessageData<uint32_t>* message =
       new rtc::TypedMessageData<uint32_t>(sid);
   signaling_thread()->Post(this, MSG_STREAMCLOSEDREMOTELY, message);
 }
 
 }  // namespace cricket