Move RTP/RTCP demuxing logic from BaseChannel to RtpTransport.

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 87 matching lines @@
 };
 
 struct DataChannelErrorMessageData : public rtc::MessageData {
   DataChannelErrorMessageData(uint32_t in_ssrc,
                               DataMediaChannel::Error in_error)
       : ssrc(in_ssrc), error(in_error) {}
   uint32_t ssrc;
   DataMediaChannel::Error error;
 };
 
-static const char* PacketType(bool rtcp) {
-  return (!rtcp) ? "RTP" : "RTCP";
-}
-
 static bool ValidPacket(bool rtcp, const rtc::CopyOnWriteBuffer* packet) {
   // Check the packet size. We could check the header too if needed.
-  return (packet &&
-          packet->size() >= (!rtcp ? kMinRtpPacketLen : kMinRtcpPacketLen) &&
-          packet->size() <= kMaxRtpPacketLen);
+  return packet && IsValidRtpRtcpPacketSize(rtcp, packet->size());
 }
 
 static bool IsReceiveContentDirection(MediaContentDirection direction) {
   return direction == MD_SENDRECV || direction == MD_RECVONLY;
 }
 
 static bool IsSendContentDirection(MediaContentDirection direction) {
   return direction == MD_SENDRECV || direction == MD_SENDONLY;
 }
 
@@ skipping 45 matching lines @@
       rtp_transport_(rtcp_mux_required),
       srtp_required_(srtp_required),
       media_channel_(media_channel),
       selected_candidate_pair_(nullptr) {
   RTC_DCHECK(worker_thread_ == rtc::Thread::Current());
 #if defined(ENABLE_EXTERNAL_AUTH)
   srtp_filter_.EnableExternalAuth();
 #endif
   rtp_transport_.SignalReadyToSend.connect(
       this, &BaseChannel::OnTransportReadyToSend);
+  // TODO(zstein):  RtpTransport::SignalPacketReceived will probably be replaced
+  // with a callback interface later so that the demuxer can select which
+  // channel to signal.
+  rtp_transport_.SignalPacketReceived.connect(this,
+                                              &BaseChannel::OnPacketReceived);
   LOG(LS_INFO) << "Created channel for " << content_name;
 }
 
 BaseChannel::~BaseChannel() {
   TRACE_EVENT0("webrtc", "BaseChannel::~BaseChannel");
   RTC_DCHECK(worker_thread_ == rtc::Thread::Current());
   Deinit();
   StopConnectionMonitor();
   // Eats any outstanding messages or packets.
   worker_thread_->Clear(&invoker_);
@@ skipping 13 matching lines @@
   // media_channel may use them from a different thread.
   if (rtp_dtls_transport_) {
     DisconnectFromDtlsTransport(rtp_dtls_transport_);
   } else if (rtp_transport_.rtp_packet_transport()) {
     DisconnectFromPacketTransport(rtp_transport_.rtp_packet_transport());
   }
   if (rtcp_dtls_transport_) {
     DisconnectFromDtlsTransport(rtcp_dtls_transport_);
   } else if (rtp_transport_.rtcp_packet_transport()) {
     DisconnectFromPacketTransport(rtp_transport_.rtcp_packet_transport());
   }

[pthatcher1, 2017/06/01 18:49:17]

I think you need rtp_transport_->SetRtpPacketTransport(null) here.

[Zach Stein, 2017/06/01 19:56:54]

Done. Thanks for helping me track this down.
This should fix the race tsan caught.

 
   // Clear pending read packets/messages.
   network_thread_->Clear(&invoker_);
   network_thread_->Clear(this);
 }
 
 bool BaseChannel::Init_w(DtlsTransportInternal* rtp_dtls_transport,
                          DtlsTransportInternal* rtcp_dtls_transport,
                          rtc::PacketTransportInternal* rtp_packet_transport,
                          rtc::PacketTransportInternal* rtcp_packet_transport) {
@@ skipping 164 matching lines @@
   for (const auto& pair : socket_options) {
     new_packet_transport->SetOption(pair.first, pair.second);
   }
 }
 
 void BaseChannel::ConnectToDtlsTransport(DtlsTransportInternal* transport) {
   RTC_DCHECK(network_thread_->IsCurrent());
 
   // TODO(zstein): de-dup with ConnectToPacketTransport
   transport->SignalWritableState.connect(this, &BaseChannel::OnWritableState);
-  transport->SignalReadPacket.connect(this, &BaseChannel::OnPacketRead);
   transport->SignalDtlsState.connect(this, &BaseChannel::OnDtlsState);
   transport->SignalSentPacket.connect(this, &BaseChannel::SignalSentPacket_n);
   transport->ice_transport()->SignalSelectedCandidatePairChanged.connect(
       this, &BaseChannel::OnSelectedCandidatePairChanged);
 }
 
 void BaseChannel::DisconnectFromDtlsTransport(
     DtlsTransportInternal* transport) {
   RTC_DCHECK(network_thread_->IsCurrent());
   OnSelectedCandidatePairChanged(transport->ice_transport(), nullptr, -1,
                                  false);
 
   transport->SignalWritableState.disconnect(this);
-  transport->SignalReadPacket.disconnect(this);
   transport->SignalDtlsState.disconnect(this);
   transport->SignalSentPacket.disconnect(this);
   transport->ice_transport()->SignalSelectedCandidatePairChanged.disconnect(
       this);
 }
 
 void BaseChannel::ConnectToPacketTransport(
     rtc::PacketTransportInternal* transport) {
   RTC_DCHECK_RUN_ON(network_thread_);
   transport->SignalWritableState.connect(this, &BaseChannel::OnWritableState);
-  transport->SignalReadPacket.connect(this, &BaseChannel::OnPacketRead);
   transport->SignalSentPacket.connect(this, &BaseChannel::SignalSentPacket_n);
 }
 
 void BaseChannel::DisconnectFromPacketTransport(
     rtc::PacketTransportInternal* transport) {
   RTC_DCHECK_RUN_ON(network_thread_);
   transport->SignalWritableState.disconnect(this);
-  transport->SignalReadPacket.disconnect(this);
   transport->SignalSentPacket.disconnect(this);
 }
 
 bool BaseChannel::Enable(bool enable) {
   worker_thread_->Invoke<void>(
       RTC_FROM_HERE,
       Bind(enable ? &BaseChannel::EnableMedia_w : &BaseChannel::DisableMedia_w,
            this));
   return true;
 }
@@ skipping 125 matching lines @@
   return transport ? transport->SetOption(opt, value) : -1;
 }
 
 void BaseChannel::OnWritableState(rtc::PacketTransportInternal* transport) {
   RTC_DCHECK(transport == rtp_transport_.rtp_packet_transport() ||
              transport == rtp_transport_.rtcp_packet_transport());
   RTC_DCHECK(network_thread_->IsCurrent());
   UpdateWritableState_n();
 }
 
-void BaseChannel::OnPacketRead(rtc::PacketTransportInternal* transport,
-                               const char* data,
-                               size_t len,
-                               const rtc::PacketTime& packet_time,
-                               int flags) {
-  TRACE_EVENT0("webrtc", "BaseChannel::OnPacketRead");
-  // OnPacketRead gets called from P2PSocket; now pass data to MediaEngine
-  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(transport, data, len);
-  rtc::CopyOnWriteBuffer packet(data, len);
-  HandlePacket(rtcp, &packet, packet_time);
-}
-
 void BaseChannel::OnDtlsState(DtlsTransportInternal* transport,
                               DtlsTransportState state) {
   if (!ShouldSetupDtlsSrtp_n()) {
     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
   // cover other scenarios like the whole transport is writable (not just this
   // TransportChannel) or when TransportChannel is attached after DTLS is
@@ skipping 29 matching lines @@
       Bind(&MediaChannel::OnNetworkRouteChanged, media_channel_, transport_name,
            network_route));
 }
 
 void BaseChannel::OnTransportReadyToSend(bool ready) {
   invoker_.AsyncInvoke<void>(
       RTC_FROM_HERE, worker_thread_,
       Bind(&MediaChannel::OnReadyToSend, media_channel_, ready));
 }
 
-bool BaseChannel::PacketIsRtcp(const rtc::PacketTransportInternal* transport,
-                               const char* data,
-                               size_t len) {
-  return (transport == rtp_transport_.rtcp_packet_transport() ||
-          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, 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.
@@ skipping 12 matching lines @@
   // 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.
   if (!rtp_transport_.IsWritable(rtcp)) {
     return false;
   }
 
   // Protect ourselves against crazy data.
   if (!ValidPacket(rtcp, packet)) {
     LOG(LS_ERROR) << "Dropping outgoing " << content_name_ << " "
-                  << PacketType(rtcp)
+                  << RtpRtcpStringLiteral(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;
@@ skipping 71 matching lines @@
                   << " and crypto is required";
     RTC_NOTREACHED();
     return false;
   }
 
   // Bon voyage.
   int flags = (secure() && secure_dtls()) ? PF_SRTP_BYPASS : PF_NORMAL;
   return rtp_transport_.SendPacket(rtcp, packet, updated_options, flags);
 }
 
-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());
+bool BaseChannel::HandlesPayloadType(int packet_type) const {
+  return rtp_transport_.HandlesPayloadType(packet_type);
 }
 
-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.
+void BaseChannel::OnPacketReceived(bool rtcp,
+                                   rtc::CopyOnWriteBuffer& packet,
+                                   const rtc::PacketTime& packet_time) {
   if (!has_received_packet_ && !rtcp) {
     has_received_packet_ = true;
     signaling_thread()->Post(RTC_FROM_HERE, 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());
+    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);
         LOG(LS_ERROR) << "Failed to unprotect " << content_name_
-                      << " RTP packet: size=" << len
-                      << ", seqnum=" << seq_num << ", SSRC=" << ssrc;
+                      << " RTP packet: size=" << len << ", seqnum=" << seq_num
+                      << ", SSRC=" << ssrc;
         return;
       }
     } else {
       res = srtp_filter_.UnprotectRtcp(data, len, &len);
       if (!res) {
         int type = -1;
         GetRtcpType(data, len, &type);
         LOG(LS_ERROR) << "Failed to unprotect " << content_name_
                       << " RTCP packet: size=" << len << ", type=" << type;
         return;
       }
     }
 
-    packet->SetSize(len);
+    packet.SetSize(len);
   } else if (srtp_required_) {
     // Our session description indicates that SRTP is required, but we got a
     // packet before our SRTP filter is active. This means either that
     // a) we got SRTP packets before we received the SDES keys, in which case
     //    we can't decrypt it anyway, or
     // b) we got SRTP packets before DTLS completed on both the RTP and RTCP
     //    transports, so we haven't yet extracted keys, even if DTLS did
     //    complete on the transport 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)
+    LOG(LS_WARNING) << "Can't process incoming " << RtpRtcpStringLiteral(rtcp)
                     << " packet when SRTP is inactive and crypto is required";
     return;
   }
 
   invoker_.AsyncInvoke<void>(
       RTC_FROM_HERE, worker_thread_,
-      Bind(&BaseChannel::OnPacketReceived, this, rtcp, *packet, packet_time));
+      Bind(&BaseChannel::ProcessPacket, this, rtcp, packet, packet_time));
 }
 
-void BaseChannel::OnPacketReceived(bool rtcp,
-                                   const rtc::CopyOnWriteBuffer& packet,
-                                   const rtc::PacketTime& packet_time) {
+void BaseChannel::ProcessPacket(bool rtcp,
+                                const rtc::CopyOnWriteBuffer& packet,
+                                const rtc::PacketTime& packet_time) {
   RTC_DCHECK(worker_thread_->IsCurrent());
+
   // Need to copy variable because OnRtcpReceived/OnPacketReceived
   // requires non-const pointer to buffer. This doesn't memcpy the actual data.
   rtc::CopyOnWriteBuffer data(packet);
   if (rtcp) {
     media_channel_->OnRtcpReceived(&data, packet_time);
   } else {
     media_channel_->OnPacketReceived(&data, packet_time);
   }
 }
 
@@ skipping 109 matching lines @@
   RTC_DCHECK(transport->IsDtlsActive());
 
   int selected_crypto_suite;
 
   if (!transport->GetSrtpCryptoSuite(&selected_crypto_suite)) {
     LOG(LS_ERROR) << "No DTLS-SRTP selected crypto suite";
     return false;
   }
 
   LOG(LS_INFO) << "Installing keys from DTLS-SRTP on " << content_name() << " "
-               << PacketType(rtcp);
+               << RtpRtcpStringLiteral(rtcp);
 
   int key_len;
   int salt_len;
   if (!rtc::GetSrtpKeyAndSaltLengths(selected_crypto_suite, &key_len,
       &salt_len)) {
     LOG(LS_ERROR) << "Unknown DTLS-SRTP crypto suite" << selected_crypto_suite;
     return false;
   }
 
   // OK, we're now doing DTLS (RFC 5764)
@@ skipping 440 matching lines @@
       delete data;
       break;
     }
     case MSG_FIRSTPACKETRECEIVED: {
       SignalFirstPacketReceived(this);
       break;
     }
   }
 }
 
+void BaseChannel::AddHandledPayloadType(int payload_type) {
+  rtp_transport_.AddHandledPayloadType(payload_type);
+}
+
 void BaseChannel::FlushRtcpMessages_n() {
   // Flush all remaining RTCP messages. This should only be called in
   // destructor.
   RTC_DCHECK(network_thread_->IsCurrent());
   rtc::MessageList rtcp_messages;
   network_thread_->Clear(this, MSG_SEND_RTCP_PACKET, &rtcp_messages);
   for (const auto& message : rtcp_messages) {
     network_thread_->Send(RTC_FROM_HERE, this, MSG_SEND_RTCP_PACKET,
                           message.pdata);
   }
@@ skipping 191 matching lines @@
 }
 
 int VoiceChannel::GetOutputLevel_w() {
   return media_channel()->GetOutputLevel();
 }
 
 void VoiceChannel::GetActiveStreams_w(AudioInfo::StreamList* actives) {
   media_channel()->GetActiveStreams(actives);
 }
 
-void VoiceChannel::OnPacketRead(rtc::PacketTransportInternal* transport,
-                                const char* data,
-                                size_t len,
-                                const rtc::PacketTime& packet_time,
-                                int flags) {
-  BaseChannel::OnPacketRead(transport, data, len, packet_time, flags);
+void VoiceChannel::OnPacketReceived(bool rtcp,
+                                    rtc::CopyOnWriteBuffer& packet,
+                                    const rtc::PacketTime& packet_time) {
+  BaseChannel::OnPacketReceived(rtcp, packet, packet_time);
   // 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(transport, data, len)) {
+  if (!received_media_ && !rtcp) {
     received_media_ = true;
   }
 }
 
 void BaseChannel::UpdateMediaSendRecvState() {
   RTC_DCHECK(network_thread_->IsCurrent());
   invoker_.AsyncInvoke<void>(
       RTC_FROM_HERE, worker_thread_,
       Bind(&BaseChannel::UpdateMediaSendRecvState_w, this));
 }
@@ skipping 78 matching lines @@
   }
 
   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);
+    AddHandledPayloadType(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;
@@ skipping 252 matching lines @@
   }
 
   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);
+    AddHandledPayloadType(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;
@@ skipping 174 matching lines @@
   }
 
   DataRecvParameters recv_params = last_recv_params_;
   RtpParametersFromMediaDescription(data, &recv_params);
   if (!media_channel()->SetRecvParameters(recv_params)) {
     SafeSetError("Failed to set remote data description recv parameters.",
                  error_desc);
     return false;
   }
   for (const DataCodec& codec : data->codecs()) {
-    bundle_filter()->AddPayloadType(codec.id);
+    AddHandledPayloadType(codec.id);
   }
   last_recv_params_ = recv_params;
 
   // 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;
@@ skipping 152 matching lines @@
 
 void RtpDataChannel::OnDataChannelReadyToSend(bool writable) {
   // This is usded for congestion control to indicate that the stream is ready
   // to send by the MediaChannel, as opposed to OnReadyToSend, which indicates
   // that the transport channel is ready.
   signaling_thread()->Post(RTC_FROM_HERE, this, MSG_READYTOSENDDATA,
                            new DataChannelReadyToSendMessageData(writable));
 }
 
 }  // namespace cricket