Network thread

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 PacketMessageData : public rtc::MessageData {
+  rtc::CopyOnWriteBuffer packet;
+  rtc::PacketOptions options;
+};
+
+struct IncomingPacketMessageData : public rtc::MessageData {
+  rtc::CopyOnWriteBuffer packet;
+  rtc::PacketTime packet_time;
+};
+
+struct ChangeState : public rtc::MessageData {
+  bool send;
+  bool recv;
+};
+
 }  // namespace
 
 enum {
   MSG_EARLYMEDIATIMEOUT = 1,
   MSG_RTPPACKET,
   MSG_RTCPPACKET,
   MSG_CHANNEL_ERROR,
   MSG_READYTOSENDDATA,
   MSG_DATARECEIVED,
   MSG_FIRSTPACKETRECEIVED,
   MSG_STREAMCLOSEDREMOTELY,
+  MSG_INCOMING_RTP_PACKET,
+  MSG_INCOMING_RTCP_PACKET,
+  MSG_CHANGE_STATE,
+  MSG_NOT_READY_TO_SEND,
+  MSG_READY_TO_SEND,
 };
 
 // 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());
+  RTC_DCHECK(worker_thread_->IsCurrent());
+  RTC_DCHECK_EQ(network_thread, transport_controller->network_thread());
   LOG(LS_INFO) << "Created channel for " << content_name;
 }
 
 BaseChannel::~BaseChannel() {
   TRACE_EVENT0("webrtc", "BaseChannel::~BaseChannel");
-  ASSERT(worker_thread_ == rtc::Thread::Current());
+  RTC_DCHECK(worker_thread_->IsCurrent());
   Deinit();
   StopConnectionMonitor();
   FlushRtcpMessages();  // Send any outstanding RTCP packets.
   worker_thread_->Clear(this);  // eats any outstanding messages or packets
+  network_thread_->Clear(this);
   // 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] {
+    if (transport_channel_) {
+      DisconnectFromTransportChannel(transport_channel_);
+      transport_controller_->DestroyTransportChannel_n(
+          transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTP);
+    }
+    if (rtcp_transport_channel_) {
+      DisconnectFromTransportChannel(rtcp_transport_channel_);
+      transport_controller_->DestroyTransportChannel_n(
+          transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTCP);
+    }
+  });
   LOG(LS_INFO) << "Destroyed channel";
 }
 
 bool BaseChannel::Init() {
   if (!SetTransport(content_name())) {
     return false;
   }
 
   if (!SetDtlsSrtpCryptoSuites(transport_channel(), false)) {
     return false;
   }
   if (rtcp_transport_enabled() &&
       !SetDtlsSrtpCryptoSuites(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::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()) {
     // Set |writable_| to false such that UpdateWritableState_w can set up
     // DTLS-SRTP when the writable_ becomes true again.
     writable_ = false;
     srtp_filter_.ResetParams();
   }
 
   // TODO(guoweis): Remove this grossness when we remove non-muxed 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),
-        false /* update_writablity */);
+    set_rtcp_transport_channel(&transport_name, false /* update_writablity */);
     if (!rtcp_transport_channel()) {
       return false;
     }
   }
 
   // We're not updating the writablity during the transition state.
-  set_transport_channel(transport_controller_->CreateTransportChannel_w(
-      transport_name, cricket::ICE_CANDIDATE_COMPONENT_RTP));
+  set_transport_channel(transport_name);
   if (!transport_channel()) {
     return false;
   }
 
   // TODO(guoweis): Remove this grossness when we remove non-muxed RTCP.
   if (rtcp_transport_enabled()) {
     // We can only update the RTCP ready to send after set_transport_channel has
     // handled channel writability.
-    SetReadyToSend(
+    SetReadyToSend_n(
         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());
+void BaseChannel::set_transport_channel(const std::string& transport_name) {
+  RTC_DCHECK(network_thread_->IsCurrent());
 
+  TransportChannel* new_tc = transport_controller_->CreateTransportChannel_n(
+      transport_name, cricket::ICE_CANDIDATE_COMPONENT_RTP);
   TransportChannel* old_tc = transport_channel_;
   if (!old_tc && !new_tc) {
     // Nothing to do
     return;
   }
-  ASSERT(old_tc != new_tc);
+  RTC_DCHECK_NE(old_tc, new_tc);
 
   if (old_tc) {
     DisconnectFromTransportChannel(old_tc);
-    transport_controller_->DestroyTransportChannel_w(
+    transport_controller_->DestroyTransportChannel_n(
         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());
+  UpdateWritableState_n();
+  SetReadyToSend_n(false, new_tc && new_tc->writable());
 }
 
-void BaseChannel::set_rtcp_transport_channel(TransportChannel* new_tc,
+void BaseChannel::set_rtcp_transport_channel(const std::string* transport_name,
                                              bool update_writablity) {
-  ASSERT(worker_thread_ == rtc::Thread::Current());
+  RTC_DCHECK(network_thread_->IsCurrent());
+
+  TransportChannel* new_tc =
+      transport_name
+          ? transport_controller_->CreateTransportChannel_n(
+                *transport_name, cricket::ICE_CANDIDATE_COMPONENT_RTCP)
+          : nullptr;
 
   TransportChannel* old_tc = rtcp_transport_channel_;
   if (!old_tc && !new_tc) {
     // Nothing to do
     return;
   }
-  ASSERT(old_tc != new_tc);
+  RTC_DCHECK_NE(old_tc, new_tc);
 
   if (old_tc) {
     DisconnectFromTransportChannel(old_tc);
-    transport_controller_->DestroyTransportChannel_w(
+    transport_controller_->DestroyTransportChannel_n(
         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();
-    SetReadyToSend(true, new_tc && new_tc->writable());
+    UpdateWritableState_n();
+    SetReadyToSend_n(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()));
   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;
+  });
 }
 
 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_);
-  SetReadyToSend(channel == rtcp_transport_channel_, true);
+  SetReadyToSend_n(channel == rtcp_transport_channel_, true);
 }
 
 void BaseChannel::OnDtlsState(TransportChannel* channel,
                               DtlsTransportState state) {
   if (!ShouldSetupDtlsSrtp()) {
     return;
   }
 
   // Reset the srtp filter if it's not the CONNECTED state. For the CONNECTED
   // state, setting up DTLS-SRTP context is deferred to ChannelWritable_w to
@@ skipping 10 matching lines @@
     CandidatePairInterface* selected_candidate_pair,
     int last_sent_packet_id) {
   ASSERT(channel == transport_channel_ || channel == rtcp_transport_channel_);
   rtc::NetworkRoute network_route;
   if (selected_candidate_pair) {
     network_route = rtc::NetworkRoute(
         selected_candidate_pair->local_candidate().network_id(),
         selected_candidate_pair->remote_candidate().network_id(),
         last_sent_packet_id);
   }
+  // TODO(danilchap): reroute this call to worker thread when it will start to
+  // do something.
   media_channel()->OnNetworkRouteChanged(channel->transport_name(),
                                          network_route);
 }
 
-void BaseChannel::SetReadyToSend(bool rtcp, bool ready) {
+void BaseChannel::SetReadyToSend_n(bool rtcp, bool ready) {
   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);
+  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_));
+
+  if (worker_thread_->IsCurrent()) {
+    media_channel_->OnReadyToSend(ready_to_send);
   } else {
-    // Notify the MediaChannel when either rtp or rtcp channel can't send.
-    media_channel_->OnReadyToSend(false);
+    worker_thread_->Post(
+        this, ready_to_send ? MSG_READY_TO_SEND : MSG_NOT_READY_TO_SEND);
   }
 }
 
 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_) {
+  // needed. Since UDP is unreliable anyway, this should be a non-issue. ///
+  // except for tests....
+  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;
     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");
     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 41 matching lines @@
         return false;
       }
     }
 
     // Update the length of the packet now that we've added the auth tag.
     packet->SetSize(len);
   } 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);
+      SetReadyToSend_n(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,
+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");
     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);
+  if (worker_thread_->IsCurrent()) {
+    // Push it down to the media channel.
+    if (!rtcp) {
+      media_channel_->OnPacketReceived(packet, packet_time);
+    } else {
+      media_channel_->OnRtcpReceived(packet, packet_time);
+    }
   } else {
-    media_channel_->OnRtcpReceived(packet, packet_time);
+    IncomingPacketMessageData* message_data = new IncomingPacketMessageData;
+    message_data->packet = std::move(*packet);
+    message_data->packet_time = packet_time;
+    int message_id = rtcp ? MSG_INCOMING_RTCP_PACKET : MSG_INCOMING_RTP_PACKET;
+    worker_thread_->Post(this, message_id, message_data);
   }
 }
 
 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 &&
@@ skipping 18 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());
+  RTC_DCHECK(worker_thread_->IsCurrent());
   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();
+  ChangeState_n();
 }
 
-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();
+  // 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());
+  RTC_DCHECK(signaling_thread()->IsCurrent());
   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()) {
     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();
+  ChangeState_n();
 }
 
-bool BaseChannel::SetRtpTransportParameters_w(
+bool BaseChannel::SetRtpTransportParameters_n(
     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);
   }
 
-  if (!SetSrtp_w(content->cryptos(), action, src, error_desc)) {
+  if (!SetSrtp_n(content->cryptos(), action, src, error_desc)) {
     return false;
   }
 
-  if (!SetRtcpMux_w(content->rtcp_mux(), action, src, error_desc)) {
+  if (!SetRtcpMux_n(content->rtcp_mux(), action, src, error_desc)) {
     return false;
   }
 
   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 156 matching lines @@
       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: {
+      RTC_DCHECK(network_thread_->IsCurrent());
       PacketMessageData* data = static_cast<PacketMessageData*>(pmsg->pdata);
       SendPacket(pmsg->message_id == MSG_RTCPPACKET, &data->packet,
                  data->options);
       delete data;  // because it is Posted
       break;
     }
     case MSG_FIRSTPACKETRECEIVED: {
       SignalFirstPacketReceived(this);
       break;
     }
+    case MSG_INCOMING_RTP_PACKET: {
+      RTC_DCHECK(worker_thread_->IsCurrent());
+      IncomingPacketMessageData* data =
+          static_cast<IncomingPacketMessageData*>(pmsg->pdata);
+      media_channel_->OnPacketReceived(&data->packet, data->packet_time);
+      delete data;
+      break;
+    }
+    case MSG_INCOMING_RTCP_PACKET: {
+      RTC_DCHECK(worker_thread_->IsCurrent());
+      IncomingPacketMessageData* data =
+          static_cast<IncomingPacketMessageData*>(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;
+    }
   }
 }
 
 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_RTCPPACKET, &rtcp_messages);
+    for (rtc::MessageList::iterator it = rtcp_messages.begin();
+         it != rtcp_messages.end(); ++it) {
+      network_thread_->Send(this, MSG_RTCPPACKET, it->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_n() {
+  RTC_DCHECK(network_thread_->IsCurrent());
+  if (worker_thread_->IsCurrent()) {
+    ChangeState_w();
+  } else {
+    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 (!InvokeOnNetwork(Bind(&BaseChannel::SetRtpTransportParameters_n,
+                            static_cast<BaseChannel*>(this), 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 (!InvokeOnNetwork(Bind(&BaseChannel::SetRtpTransportParameters_n,
+                            static_cast<BaseChannel*>(this), 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 (!InvokeOnNetwork(Bind(&BaseChannel::SetRtpTransportParameters_n,
+                            static_cast<BaseChannel*>(this), 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 (!InvokeOnNetwork(Bind(&BaseChannel::SetRtpTransportParameters_n,
+                            static_cast<BaseChannel*>(this), 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 (!InvokeOnNetwork(Bind(&BaseChannel::SetRtpTransportParameters_n,
+                              static_cast<BaseChannel*>(this), 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)) {
-    return false;
+  if (data_channel_type_ == DCT_RTP) {
+    if (!InvokeOnNetwork(Bind(&BaseChannel::SetRtpTransportParameters_n,
+                              static_cast<BaseChannel*>(this), 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