diff --git a/call/degraded_call.cc b/call/degraded_call.cc index 61102a6abe..5de50fd337 100644 --- a/call/degraded_call.cc +++ b/call/degraded_call.cc @@ -20,28 +20,39 @@ namespace webrtc { DegradedCall::FakeNetworkPipeOnTaskQueue::FakeNetworkPipeOnTaskQueue( TaskQueueFactory* task_queue_factory, Clock* clock, - std::unique_ptr network_behavior, - Transport* transport) + std::unique_ptr network_behavior) : clock_(clock), task_queue_(task_queue_factory->CreateTaskQueue( "DegradedSendQueue", TaskQueueFactory::Priority::NORMAL)), - pipe_(clock, std::move(network_behavior), transport) {} + pipe_(clock, std::move(network_behavior)) {} void DegradedCall::FakeNetworkPipeOnTaskQueue::SendRtp( const uint8_t* packet, size_t length, - const PacketOptions& options) { - pipe_.SendRtp(packet, length, options); + const PacketOptions& options, + Transport* transport) { + pipe_.SendRtp(packet, length, options, transport); Process(); } void DegradedCall::FakeNetworkPipeOnTaskQueue::SendRtcp(const uint8_t* packet, - size_t length) { - pipe_.SendRtcp(packet, length); + size_t length, + Transport* transport) { + pipe_.SendRtcp(packet, length, transport); Process(); } +void DegradedCall::FakeNetworkPipeOnTaskQueue::AddActiveTransport( + Transport* transport) { + pipe_.AddActiveTransport(transport); +} + +void DegradedCall::FakeNetworkPipeOnTaskQueue::RemoveActiveTransport( + Transport* transport) { + pipe_.RemoveActiveTransport(transport); +} + bool DegradedCall::FakeNetworkPipeOnTaskQueue::Process() { pipe_.Process(); auto time_to_next = pipe_.TimeUntilNextProcess(); @@ -69,6 +80,51 @@ bool DegradedCall::FakeNetworkPipeOnTaskQueue::Process() { return true; } +DegradedCall::FakeNetworkPipeTransportAdapter::FakeNetworkPipeTransportAdapter( + FakeNetworkPipeOnTaskQueue* fake_network, + Call* call, + Clock* clock, + Transport* real_transport) + : network_pipe_(fake_network), + call_(call), + clock_(clock), + real_transport_(real_transport) { + network_pipe_->AddActiveTransport(real_transport); +} + +DegradedCall::FakeNetworkPipeTransportAdapter:: + ~FakeNetworkPipeTransportAdapter() { + network_pipe_->RemoveActiveTransport(real_transport_); +} + +bool DegradedCall::FakeNetworkPipeTransportAdapter::SendRtp( + const uint8_t* packet, + size_t length, + const PacketOptions& options) { + // A call here comes from the RTP stack (probably pacer). We intercept it and + // put it in the fake network pipe instead, but report to Call that is has + // been sent, so that the bandwidth estimator sees the delay we add. + network_pipe_->SendRtp(packet, length, options, real_transport_); + if (options.packet_id != -1) { + rtc::SentPacket sent_packet; + sent_packet.packet_id = options.packet_id; + sent_packet.send_time_ms = clock_->TimeInMilliseconds(); + sent_packet.info.included_in_feedback = options.included_in_feedback; + sent_packet.info.included_in_allocation = options.included_in_allocation; + sent_packet.info.packet_size_bytes = length; + sent_packet.info.packet_type = rtc::PacketType::kData; + call_->OnSentPacket(sent_packet); + } + return true; +} + +bool DegradedCall::FakeNetworkPipeTransportAdapter::SendRtcp( + const uint8_t* packet, + size_t length) { + network_pipe_->SendRtcp(packet, length, real_transport_); + return true; +} + DegradedCall::DegradedCall( std::unique_ptr call, absl::optional send_config, @@ -87,17 +143,36 @@ DegradedCall::DegradedCall( absl::make_unique(clock_, std::move(network)); receive_pipe_->SetReceiver(call_->Receiver()); } + if (send_config_) { + auto network = absl::make_unique(*send_config_); + send_simulated_network_ = network.get(); + send_pipe_ = absl::make_unique( + task_queue_factory_, clock_, std::move(network)); + } } DegradedCall::~DegradedCall() = default; AudioSendStream* DegradedCall::CreateAudioSendStream( const AudioSendStream::Config& config) { + if (send_config_) { + auto transport_adapter = absl::make_unique( + send_pipe_.get(), call_.get(), clock_, config.send_transport); + AudioSendStream::Config degrade_config = config; + degrade_config.send_transport = transport_adapter.get(); + AudioSendStream* send_stream = call_->CreateAudioSendStream(degrade_config); + if (send_stream) { + audio_send_transport_adapters_[send_stream] = + std::move(transport_adapter); + } + return send_stream; + } return call_->CreateAudioSendStream(config); } void DegradedCall::DestroyAudioSendStream(AudioSendStream* send_stream) { call_->DestroyAudioSendStream(send_stream); + audio_send_transport_adapters_.erase(send_stream); } AudioReceiveStream* DegradedCall::CreateAudioReceiveStream( @@ -113,34 +188,41 @@ void DegradedCall::DestroyAudioReceiveStream( VideoSendStream* DegradedCall::CreateVideoSendStream( VideoSendStream::Config config, VideoEncoderConfig encoder_config) { - if (send_config_ && !send_pipe_) { - auto network = absl::make_unique(*send_config_); - send_simulated_network_ = network.get(); - send_pipe_ = absl::make_unique( - task_queue_factory_, clock_, std::move(network), config.send_transport); - config.send_transport = this; + std::unique_ptr transport_adapter; + if (send_config_) { + transport_adapter = absl::make_unique( + send_pipe_.get(), call_.get(), clock_, config.send_transport); + config.send_transport = transport_adapter.get(); } - return call_->CreateVideoSendStream(std::move(config), - std::move(encoder_config)); + VideoSendStream* send_stream = call_->CreateVideoSendStream( + std::move(config), std::move(encoder_config)); + if (send_stream && transport_adapter) { + video_send_transport_adapters_[send_stream] = std::move(transport_adapter); + } + return send_stream; } VideoSendStream* DegradedCall::CreateVideoSendStream( VideoSendStream::Config config, VideoEncoderConfig encoder_config, std::unique_ptr fec_controller) { - if (send_config_ && !send_pipe_) { - auto network = absl::make_unique(*send_config_); - send_simulated_network_ = network.get(); - send_pipe_ = absl::make_unique( - task_queue_factory_, clock_, std::move(network), config.send_transport); - config.send_transport = this; + std::unique_ptr transport_adapter; + if (send_config_) { + transport_adapter = absl::make_unique( + send_pipe_.get(), call_.get(), clock_, config.send_transport); + config.send_transport = transport_adapter.get(); } - return call_->CreateVideoSendStream( + VideoSendStream* send_stream = call_->CreateVideoSendStream( std::move(config), std::move(encoder_config), std::move(fec_controller)); + if (send_stream && transport_adapter) { + video_send_transport_adapters_[send_stream] = std::move(transport_adapter); + } + return send_stream; } void DegradedCall::DestroyVideoSendStream(VideoSendStream* send_stream) { call_->DestroyVideoSendStream(send_stream); + video_send_transport_adapters_.erase(send_stream); } VideoReceiveStream* DegradedCall::CreateVideoReceiveStream( @@ -199,31 +281,6 @@ void DegradedCall::OnSentPacket(const rtc::SentPacket& sent_packet) { call_->OnSentPacket(sent_packet); } -bool DegradedCall::SendRtp(const uint8_t* packet, - size_t length, - const PacketOptions& options) { - // A call here comes from the RTP stack (probably pacer). We intercept it and - // put it in the fake network pipe instead, but report to Call that is has - // been sent, so that the bandwidth estimator sees the delay we add. - send_pipe_->SendRtp(packet, length, options); - if (options.packet_id != -1) { - rtc::SentPacket sent_packet; - sent_packet.packet_id = options.packet_id; - sent_packet.send_time_ms = clock_->TimeInMilliseconds(); - sent_packet.info.included_in_feedback = options.included_in_feedback; - sent_packet.info.included_in_allocation = options.included_in_allocation; - sent_packet.info.packet_size_bytes = length; - sent_packet.info.packet_type = rtc::PacketType::kData; - call_->OnSentPacket(sent_packet); - } - return true; -} - -bool DegradedCall::SendRtcp(const uint8_t* packet, size_t length) { - send_pipe_->SendRtcp(packet, length); - return true; -} - PacketReceiver::DeliveryStatus DegradedCall::DeliverPacket( MediaType media_type, rtc::CopyOnWriteBuffer packet, @@ -233,7 +290,7 @@ PacketReceiver::DeliveryStatus DegradedCall::DeliverPacket( // This is not optimal, but there are many places where there are thread // checks that fail if we're not using the worker thread call into this // method. If we want to fix this we probably need a task queue to do handover - // of all overriden methods, which feels like overikill for the current use + // of all overriden methods, which feels like overkill for the current use // case. // By just having this thread call out via the Process() method we work around // that, with the tradeoff that a non-zero delay may become a little larger @@ -241,5 +298,4 @@ PacketReceiver::DeliveryStatus DegradedCall::DeliverPacket( receive_pipe_->Process(); return status; } - } // namespace webrtc diff --git a/call/degraded_call.h b/call/degraded_call.h index 609049333f..49230ca1ed 100644 --- a/call/degraded_call.h +++ b/call/degraded_call.h @@ -14,6 +14,7 @@ #include #include +#include #include #include "absl/types/optional.h" @@ -40,7 +41,7 @@ #include "system_wrappers/include/clock.h" namespace webrtc { -class DegradedCall : public Call, private Transport, private PacketReceiver { +class DegradedCall : public Call, private PacketReceiver { public: explicit DegradedCall( std::unique_ptr call, @@ -88,13 +89,6 @@ class DegradedCall : public Call, private Transport, private PacketReceiver { void OnSentPacket(const rtc::SentPacket& sent_packet) override; protected: - // Implements Transport. - bool SendRtp(const uint8_t* packet, - size_t length, - const PacketOptions& options) override; - - bool SendRtcp(const uint8_t* packet, size_t length) override; - // Implements PacketReceiver. DeliveryStatus DeliverPacket(MediaType media_type, rtc::CopyOnWriteBuffer packet, @@ -106,13 +100,16 @@ class DegradedCall : public Call, private Transport, private PacketReceiver { FakeNetworkPipeOnTaskQueue( TaskQueueFactory* task_queue_factory, Clock* clock, - std::unique_ptr network_behavior, - Transport* transport); + std::unique_ptr network_behavior); void SendRtp(const uint8_t* packet, size_t length, - const PacketOptions& options); - void SendRtcp(const uint8_t* packet, size_t length); + const PacketOptions& options, + Transport* transport); + void SendRtcp(const uint8_t* packet, size_t length, Transport* transport); + + void AddActiveTransport(Transport* transport); + void RemoveActiveTransport(Transport* transport); private: // Try to process packets on the fake network queue. @@ -125,15 +122,44 @@ class DegradedCall : public Call, private Transport, private PacketReceiver { absl::optional next_process_ms_ RTC_GUARDED_BY(&task_queue_); }; + // For audio/video send stream, a TransportAdapter instance is used to + // intercept packets to be sent, and put them into a common FakeNetworkPipe + // in such as way that they will eventually (unless dropped) be forwarded to + // the correct Transport for that stream. + class FakeNetworkPipeTransportAdapter : public Transport { + public: + FakeNetworkPipeTransportAdapter(FakeNetworkPipeOnTaskQueue* fake_network, + Call* call, + Clock* clock, + Transport* real_transport); + ~FakeNetworkPipeTransportAdapter(); + + bool SendRtp(const uint8_t* packet, + size_t length, + const PacketOptions& options) override; + bool SendRtcp(const uint8_t* packet, size_t length) override; + + private: + FakeNetworkPipeOnTaskQueue* const network_pipe_; + Call* const call_; + Clock* const clock_; + Transport* const real_transport_; + }; + Clock* const clock_; const std::unique_ptr call_; TaskQueueFactory* const task_queue_factory_; void SetClientBitratePreferences( const webrtc::BitrateSettings& preferences) override {} + const absl::optional send_config_; SimulatedNetwork* send_simulated_network_; std::unique_ptr send_pipe_; + std::map> + audio_send_transport_adapters_; + std::map> + video_send_transport_adapters_; const absl::optional receive_config_; SimulatedNetwork* receive_simulated_network_; diff --git a/call/fake_network_pipe.cc b/call/fake_network_pipe.cc index ea0415ce8e..8844700e67 100644 --- a/call/fake_network_pipe.cc +++ b/call/fake_network_pipe.cc @@ -35,14 +35,16 @@ NetworkPacket::NetworkPacket(rtc::CopyOnWriteBuffer packet, absl::optional packet_options, bool is_rtcp, MediaType media_type, - absl::optional packet_time_us) + absl::optional packet_time_us, + Transport* transport) : packet_(std::move(packet)), send_time_(send_time), arrival_time_(arrival_time), packet_options_(packet_options), is_rtcp_(is_rtcp), media_type_(media_type), - packet_time_us_(packet_time_us) {} + packet_time_us_(packet_time_us), + transport_(transport) {} NetworkPacket::NetworkPacket(NetworkPacket&& o) : packet_(std::move(o.packet_)), @@ -51,7 +53,8 @@ NetworkPacket::NetworkPacket(NetworkPacket&& o) packet_options_(o.packet_options_), is_rtcp_(o.is_rtcp_), media_type_(o.media_type_), - packet_time_us_(o.packet_time_us_) {} + packet_time_us_(o.packet_time_us_), + transport_(o.transport_) {} NetworkPacket::~NetworkPacket() = default; @@ -63,6 +66,7 @@ NetworkPacket& NetworkPacket::operator=(NetworkPacket&& o) { is_rtcp_ = o.is_rtcp_; media_type_ = o.media_type_; packet_time_us_ = o.packet_time_us_; + transport_ = o.transport_; return *this; } @@ -86,7 +90,7 @@ FakeNetworkPipe::FakeNetworkPipe( : clock_(clock), network_behavior_(std::move(network_behavior)), receiver_(receiver), - transport_(nullptr), + global_transport_(nullptr), clock_offset_ms_(0), dropped_packets_(0), sent_packets_(0), @@ -100,33 +104,74 @@ FakeNetworkPipe::FakeNetworkPipe( : clock_(clock), network_behavior_(std::move(network_behavior)), receiver_(nullptr), - transport_(transport), + global_transport_(transport), clock_offset_ms_(0), dropped_packets_(0), sent_packets_(0), total_packet_delay_us_(0), - last_log_time_us_(clock_->TimeInMicroseconds()) {} + last_log_time_us_(clock_->TimeInMicroseconds()) { + RTC_DCHECK(global_transport_); + AddActiveTransport(global_transport_); +} -FakeNetworkPipe::~FakeNetworkPipe() = default; +FakeNetworkPipe::~FakeNetworkPipe() { + if (global_transport_) { + RemoveActiveTransport(global_transport_); + } + RTC_DCHECK(active_transports_.empty()); +} void FakeNetworkPipe::SetReceiver(PacketReceiver* receiver) { rtc::CritScope crit(&config_lock_); receiver_ = receiver; } +void FakeNetworkPipe::AddActiveTransport(Transport* transport) { + rtc::CritScope crit(&config_lock_); + active_transports_[transport]++; +} + +void FakeNetworkPipe::RemoveActiveTransport(Transport* transport) { + rtc::CritScope crit(&config_lock_); + auto it = active_transports_.find(transport); + RTC_CHECK(it != active_transports_.end()); + if (--(it->second) == 0) { + active_transports_.erase(it); + } +} + bool FakeNetworkPipe::SendRtp(const uint8_t* packet, size_t length, const PacketOptions& options) { - RTC_DCHECK(HasTransport()); + RTC_DCHECK(global_transport_); EnqueuePacket(rtc::CopyOnWriteBuffer(packet, length), options, false, - MediaType::ANY); + global_transport_); return true; } bool FakeNetworkPipe::SendRtcp(const uint8_t* packet, size_t length) { - RTC_DCHECK(HasTransport()); + RTC_DCHECK(global_transport_); EnqueuePacket(rtc::CopyOnWriteBuffer(packet, length), absl::nullopt, true, - MediaType::ANY); + global_transport_); + return true; +} + +bool FakeNetworkPipe::SendRtp(const uint8_t* packet, + size_t length, + const PacketOptions& options, + Transport* transport) { + RTC_DCHECK(transport); + EnqueuePacket(rtc::CopyOnWriteBuffer(packet, length), options, false, + transport); + return true; +} + +bool FakeNetworkPipe::SendRtcp(const uint8_t* packet, + size_t length, + Transport* transport) { + RTC_DCHECK(transport); + EnqueuePacket(rtc::CopyOnWriteBuffer(packet, length), absl::nullopt, true, + transport); return true; } @@ -155,14 +200,30 @@ bool FakeNetworkPipe::EnqueuePacket(rtc::CopyOnWriteBuffer packet, absl::optional packet_time_us) { rtc::CritScope crit(&process_lock_); int64_t time_now_us = clock_->TimeInMicroseconds(); - size_t packet_size = packet.size(); - NetworkPacket net_packet(std::move(packet), time_now_us, time_now_us, options, - is_rtcp, media_type, packet_time_us); + return EnqueuePacket(NetworkPacket(std::move(packet), time_now_us, + time_now_us, options, is_rtcp, media_type, + packet_time_us, nullptr)); +} + +bool FakeNetworkPipe::EnqueuePacket(rtc::CopyOnWriteBuffer packet, + absl::optional options, + bool is_rtcp, + Transport* transport) { + rtc::CritScope crit(&process_lock_); + int64_t time_now_us = clock_->TimeInMicroseconds(); + return EnqueuePacket(NetworkPacket(std::move(packet), time_now_us, + time_now_us, options, is_rtcp, + MediaType::ANY, absl::nullopt, transport)); +} + +bool FakeNetworkPipe::EnqueuePacket(NetworkPacket&& net_packet) { + int64_t send_time_us = net_packet.send_time(); + size_t packet_size = net_packet.data_length(); packets_in_flight_.emplace_back(StoredPacket(std::move(net_packet))); int64_t packet_id = reinterpret_cast(&packets_in_flight_.back()); bool sent = network_behavior_->EnqueuePacket( - PacketInFlightInfo(packet_size, time_now_us, packet_id)); + PacketInFlightInfo(packet_size, send_time_us, packet_id)); if (!sent) { packets_in_flight_.pop_back(); @@ -266,13 +327,18 @@ void FakeNetworkPipe::Process() { } void FakeNetworkPipe::DeliverNetworkPacket(NetworkPacket* packet) { - if (transport_) { + Transport* transport = packet->transport(); + if (transport) { RTC_DCHECK(!receiver_); + if (active_transports_.find(transport) == active_transports_.end()) { + // Transport has been destroyed, ignore this packet. + return; + } if (packet->is_rtcp()) { - transport_->SendRtcp(packet->data(), packet->data_length()); + transport->SendRtcp(packet->data(), packet->data_length()); } else { - transport_->SendRtp(packet->data(), packet->data_length(), - packet->packet_options()); + transport->SendRtp(packet->data(), packet->data_length(), + packet->packet_options()); } } else if (receiver_) { int64_t packet_time_us = packet->packet_time_us().value_or(-1); @@ -297,10 +363,6 @@ absl::optional FakeNetworkPipe::TimeUntilNextProcess() { return absl::nullopt; } -bool FakeNetworkPipe::HasTransport() const { - rtc::CritScope crit(&config_lock_); - return transport_ != nullptr; -} bool FakeNetworkPipe::HasReceiver() const { rtc::CritScope crit(&config_lock_); return receiver_ != nullptr; diff --git a/call/fake_network_pipe.h b/call/fake_network_pipe.h index 661815bf44..24340a2f29 100644 --- a/call/fake_network_pipe.h +++ b/call/fake_network_pipe.h @@ -41,7 +41,8 @@ class NetworkPacket { absl::optional packet_options, bool is_rtcp, MediaType media_type, - absl::optional packet_time_us); + absl::optional packet_time_us, + Transport* transport); // Disallow copy constructor and copy assignment (no deep copies of |data_|). NetworkPacket(const NetworkPacket&) = delete; @@ -65,6 +66,7 @@ class NetworkPacket { bool is_rtcp() const { return is_rtcp_; } MediaType media_type() const { return media_type_; } absl::optional packet_time_us() const { return packet_time_us_; } + Transport* transport() const { return transport_; } private: rtc::CopyOnWriteBuffer packet_; @@ -82,6 +84,7 @@ class NetworkPacket { // network pipe. MediaType media_type_; absl::optional packet_time_us_; + Transport* transport_; }; // Class faking a network link, internally is uses an implementation of a @@ -89,7 +92,6 @@ class NetworkPacket { class FakeNetworkPipe : public SimulatedPacketReceiverInterface { public: // Will keep |network_behavior| alive while pipe is alive itself. - // Use these constructors if you plan to insert packets using DeliverPacket(). FakeNetworkPipe(Clock* clock, std::unique_ptr network_behavior); FakeNetworkPipe(Clock* clock, @@ -112,6 +114,12 @@ class FakeNetworkPipe : public SimulatedPacketReceiverInterface { // Must not be called in parallel with DeliverPacket or Process. void SetReceiver(PacketReceiver* receiver) override; + // Adds/subtracts references to Transport instances. If a Transport is + // destroyed we cannot use to forward a potential delayed packet, these + // methods are used to maintain a map of which instances are live. + void AddActiveTransport(Transport* transport); + void RemoveActiveTransport(Transport* transport); + // Implements Transport interface. When/if packets are delivered, they will // be passed to the transport instance given in SetReceiverTransport(). These // methods should only be called if a Transport instance was provided in the @@ -121,6 +129,15 @@ class FakeNetworkPipe : public SimulatedPacketReceiverInterface { const PacketOptions& options); bool SendRtcp(const uint8_t* packet, size_t length); + // Methods for use with Transport interface. When/if packets are delivered, + // they will be passed to the instance specified by the |transport| parameter. + // Note that that instance must be in the map of active transports. + bool SendRtp(const uint8_t* packet, + size_t length, + const PacketOptions& options, + Transport* transport); + bool SendRtcp(const uint8_t* packet, size_t length, Transport* transport); + // Implements the PacketReceiver interface. When/if packets are delivered, // they will be passed directly to the receiver instance given in // SetReceiver(), without passing through a Demuxer. The receive time @@ -163,22 +180,26 @@ class FakeNetworkPipe : public SimulatedPacketReceiverInterface { StoredPacket() = delete; }; - // Returns true if enqueued, or false if packet was dropped. - virtual bool EnqueuePacket(rtc::CopyOnWriteBuffer packet, - absl::optional options, - bool is_rtcp, - MediaType media_type, - absl::optional packet_time_us); - + // Returns true if enqueued, or false if packet was dropped. Use this method + // when enqueueing packets that should be received by PacketReceiver instance. bool EnqueuePacket(rtc::CopyOnWriteBuffer packet, absl::optional options, bool is_rtcp, - MediaType media_type) { - return EnqueuePacket(packet, options, is_rtcp, media_type, absl::nullopt); - } + MediaType media_type, + absl::optional packet_time_us); + + // Returns true if enqueued, or false if packet was dropped. Use this method + // when enqueueing packets that should be received by Transport instance. + bool EnqueuePacket(rtc::CopyOnWriteBuffer packet, + absl::optional options, + bool is_rtcp, + Transport* transport); + + bool EnqueuePacket(NetworkPacket&& net_packet) + RTC_EXCLUSIVE_LOCKS_REQUIRED(process_lock_); + void DeliverNetworkPacket(NetworkPacket* packet) RTC_EXCLUSIVE_LOCKS_REQUIRED(config_lock_); - bool HasTransport() const; bool HasReceiver() const; Clock* const clock_; @@ -186,7 +207,7 @@ class FakeNetworkPipe : public SimulatedPacketReceiverInterface { rtc::CriticalSection config_lock_; const std::unique_ptr network_behavior_; PacketReceiver* receiver_ RTC_GUARDED_BY(config_lock_); - Transport* const transport_ RTC_GUARDED_BY(config_lock_); + Transport* const global_transport_; // |process_lock| guards the data structures involved in delay and loss // processes, such as the packet queues. @@ -206,6 +227,8 @@ class FakeNetworkPipe : public SimulatedPacketReceiverInterface { int64_t total_packet_delay_us_ RTC_GUARDED_BY(process_lock_); int64_t last_log_time_us_; + std::map active_transports_ RTC_GUARDED_BY(config_lock_); + RTC_DISALLOW_COPY_AND_ASSIGN(FakeNetworkPipe); };