| /* |
| * Copyright (c) 2021 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. |
| */ |
| #include <cstdint> |
| #include <deque> |
| #include <memory> |
| #include <string> |
| #include <utility> |
| #include <vector> |
| |
| #include "absl/memory/memory.h" |
| #include "absl/strings/string_view.h" |
| #include "absl/types/optional.h" |
| #include "api/array_view.h" |
| #include "api/task_queue/pending_task_safety_flag.h" |
| #include "api/task_queue/task_queue_base.h" |
| #include "api/test/create_network_emulation_manager.h" |
| #include "api/test/network_emulation_manager.h" |
| #include "api/units/data_rate.h" |
| #include "api/units/time_delta.h" |
| #include "net/dcsctp/public/dcsctp_options.h" |
| #include "net/dcsctp/public/dcsctp_socket.h" |
| #include "net/dcsctp/public/types.h" |
| #include "net/dcsctp/socket/dcsctp_socket.h" |
| #include "net/dcsctp/testing/testing_macros.h" |
| #include "net/dcsctp/timer/task_queue_timeout.h" |
| #include "rtc_base/copy_on_write_buffer.h" |
| #include "rtc_base/gunit.h" |
| #include "rtc_base/logging.h" |
| #include "rtc_base/random.h" |
| #include "rtc_base/socket_address.h" |
| #include "rtc_base/strings/string_format.h" |
| #include "rtc_base/time_utils.h" |
| #include "test/gmock.h" |
| |
| #if !defined(WEBRTC_ANDROID) && defined(NDEBUG) && \ |
| !defined(THREAD_SANITIZER) && !defined(MEMORY_SANITIZER) |
| #define DCSCTP_NDEBUG_TEST(t) t |
| #else |
| // In debug mode, and when MSAN or TSAN sanitizers are enabled, these tests are |
| // too expensive to run due to extensive consistency checks that iterate on all |
| // outstanding chunks. Same with low-end Android devices, which have |
| // difficulties with these tests. |
| #define DCSCTP_NDEBUG_TEST(t) DISABLED_##t |
| #endif |
| |
| namespace dcsctp { |
| namespace { |
| using ::testing::AllOf; |
| using ::testing::Ge; |
| using ::testing::Le; |
| using ::testing::SizeIs; |
| using ::webrtc::DataRate; |
| using ::webrtc::TimeDelta; |
| using ::webrtc::Timestamp; |
| |
| constexpr StreamID kStreamId(1); |
| constexpr PPID kPpid(53); |
| constexpr size_t kSmallPayloadSize = 10; |
| constexpr size_t kLargePayloadSize = 10000; |
| constexpr size_t kHugePayloadSize = 262144; |
| constexpr size_t kBufferedAmountLowThreshold = kLargePayloadSize * 2; |
| constexpr webrtc::TimeDelta kPrintBandwidthDuration = |
| webrtc::TimeDelta::Seconds(1); |
| constexpr webrtc::TimeDelta kBenchmarkRuntime(webrtc::TimeDelta::Seconds(10)); |
| constexpr webrtc::TimeDelta kAWhile(webrtc::TimeDelta::Seconds(1)); |
| |
| inline int GetUniqueSeed() { |
| static int seed = 0; |
| return ++seed; |
| } |
| |
| DcSctpOptions MakeOptionsForTest() { |
| DcSctpOptions options; |
| |
| // Throughput numbers are affected by the MTU. Ensure it's constant. |
| options.mtu = 1200; |
| |
| // By disabling the heartbeat interval, there will no timers at all running |
| // when the socket is idle, which makes it easy to just continue the test |
| // until there are no more scheduled tasks. Note that it _will_ run for longer |
| // than necessary as timers aren't cancelled when they are stopped (as that's |
| // not supported), but it's still simulated time and passes quickly. |
| options.heartbeat_interval = DurationMs(0); |
| return options; |
| } |
| |
| // When doing throughput tests, knowing what each actor should do. |
| enum class ActorMode { |
| kAtRest, |
| kThroughputSender, |
| kThroughputReceiver, |
| kLimitedRetransmissionSender, |
| }; |
| |
| // An abstraction around EmulatedEndpoint, representing a bound socket that |
| // will send its packet to a given destination. |
| class BoundSocket : public webrtc::EmulatedNetworkReceiverInterface { |
| public: |
| void Bind(webrtc::EmulatedEndpoint* endpoint) { |
| endpoint_ = endpoint; |
| uint16_t port = endpoint->BindReceiver(0, this).value(); |
| source_address_ = |
| rtc::SocketAddress(endpoint_->GetPeerLocalAddress(), port); |
| } |
| |
| void SetDestination(const BoundSocket& socket) { |
| dest_address_ = socket.source_address_; |
| } |
| |
| void SetReceiver(std::function<void(rtc::CopyOnWriteBuffer)> receiver) { |
| receiver_ = std::move(receiver); |
| } |
| |
| void SendPacket(rtc::ArrayView<const uint8_t> data) { |
| endpoint_->SendPacket(source_address_, dest_address_, |
| rtc::CopyOnWriteBuffer(data.data(), data.size())); |
| } |
| |
| private: |
| // Implementation of `webrtc::EmulatedNetworkReceiverInterface`. |
| void OnPacketReceived(webrtc::EmulatedIpPacket packet) override { |
| receiver_(std::move(packet.data)); |
| } |
| |
| std::function<void(rtc::CopyOnWriteBuffer)> receiver_; |
| webrtc::EmulatedEndpoint* endpoint_ = nullptr; |
| rtc::SocketAddress source_address_; |
| rtc::SocketAddress dest_address_; |
| }; |
| |
| // Sends at a constant rate but with random packet sizes. |
| class SctpActor : public DcSctpSocketCallbacks { |
| public: |
| SctpActor(absl::string_view name, |
| BoundSocket& emulated_socket, |
| const DcSctpOptions& sctp_options) |
| : log_prefix_(std::string(name) + ": "), |
| thread_(rtc::Thread::Current()), |
| emulated_socket_(emulated_socket), |
| timeout_factory_( |
| *thread_, |
| [this]() { return TimeMs(Now().ms()); }, |
| [this](dcsctp::TimeoutID timeout_id) { |
| sctp_socket_.HandleTimeout(timeout_id); |
| }), |
| random_(GetUniqueSeed()), |
| sctp_socket_(name, *this, nullptr, sctp_options), |
| last_bandwidth_printout_(Now()) { |
| emulated_socket.SetReceiver([this](rtc::CopyOnWriteBuffer buf) { |
| // The receiver will be executed on the NetworkEmulation task queue, but |
| // the dcSCTP socket is owned by `thread_` and is not thread-safe. |
| thread_->PostTask([this, buf] { this->sctp_socket_.ReceivePacket(buf); }); |
| }); |
| } |
| |
| void PrintBandwidth() { |
| Timestamp now = Now(); |
| TimeDelta duration = now - last_bandwidth_printout_; |
| |
| double bitrate_mbps = |
| static_cast<double>(received_bytes_ * 8) / duration.ms() / 1000; |
| RTC_LOG(LS_INFO) << log_prefix() |
| << rtc::StringFormat("Received %0.2f Mbps", bitrate_mbps); |
| |
| received_bitrate_mbps_.push_back(bitrate_mbps); |
| received_bytes_ = 0; |
| last_bandwidth_printout_ = now; |
| // Print again in a second. |
| if (mode_ == ActorMode::kThroughputReceiver) { |
| thread_->PostDelayedTask( |
| SafeTask(safety_.flag(), [this] { PrintBandwidth(); }), |
| kPrintBandwidthDuration); |
| } |
| } |
| |
| void SendPacket(rtc::ArrayView<const uint8_t> data) override { |
| emulated_socket_.SendPacket(data); |
| } |
| |
| std::unique_ptr<Timeout> CreateTimeout( |
| webrtc::TaskQueueBase::DelayPrecision precision) override { |
| return timeout_factory_.CreateTimeout(precision); |
| } |
| |
| Timestamp Now() override { return Timestamp::Millis(rtc::TimeMillis()); } |
| |
| uint32_t GetRandomInt(uint32_t low, uint32_t high) override { |
| return random_.Rand(low, high); |
| } |
| |
| void OnMessageReceived(DcSctpMessage message) override { |
| received_bytes_ += message.payload().size(); |
| last_received_message_ = std::move(message); |
| } |
| |
| void OnError(ErrorKind error, absl::string_view message) override { |
| RTC_LOG(LS_WARNING) << log_prefix() << "Socket error: " << ToString(error) |
| << "; " << message; |
| } |
| |
| void OnAborted(ErrorKind error, absl::string_view message) override { |
| RTC_LOG(LS_ERROR) << log_prefix() << "Socket abort: " << ToString(error) |
| << "; " << message; |
| } |
| |
| void OnConnected() override {} |
| |
| void OnClosed() override {} |
| |
| void OnConnectionRestarted() override {} |
| |
| void OnStreamsResetFailed(rtc::ArrayView<const StreamID> outgoing_streams, |
| absl::string_view reason) override {} |
| |
| void OnStreamsResetPerformed( |
| rtc::ArrayView<const StreamID> outgoing_streams) override {} |
| |
| void OnIncomingStreamsReset( |
| rtc::ArrayView<const StreamID> incoming_streams) override {} |
| |
| void NotifyOutgoingMessageBufferEmpty() override {} |
| |
| void OnBufferedAmountLow(StreamID stream_id) override { |
| if (mode_ == ActorMode::kThroughputSender) { |
| std::vector<uint8_t> payload(kHugePayloadSize); |
| sctp_socket_.Send(DcSctpMessage(kStreamId, kPpid, std::move(payload)), |
| SendOptions()); |
| |
| } else if (mode_ == ActorMode::kLimitedRetransmissionSender) { |
| while (sctp_socket_.buffered_amount(kStreamId) < |
| kBufferedAmountLowThreshold * 2) { |
| SendOptions send_options; |
| send_options.max_retransmissions = 0; |
| sctp_socket_.Send( |
| DcSctpMessage(kStreamId, kPpid, |
| std::vector<uint8_t>(kLargePayloadSize)), |
| send_options); |
| |
| send_options.max_retransmissions = absl::nullopt; |
| sctp_socket_.Send( |
| DcSctpMessage(kStreamId, kPpid, |
| std::vector<uint8_t>(kSmallPayloadSize)), |
| send_options); |
| } |
| } |
| } |
| |
| absl::optional<DcSctpMessage> ConsumeReceivedMessage() { |
| if (!last_received_message_.has_value()) { |
| return absl::nullopt; |
| } |
| DcSctpMessage ret = *std::move(last_received_message_); |
| last_received_message_ = absl::nullopt; |
| return ret; |
| } |
| |
| DcSctpSocket& sctp_socket() { return sctp_socket_; } |
| |
| void SetActorMode(ActorMode mode) { |
| mode_ = mode; |
| if (mode_ == ActorMode::kThroughputSender) { |
| sctp_socket_.SetBufferedAmountLowThreshold(kStreamId, |
| kBufferedAmountLowThreshold); |
| std::vector<uint8_t> payload(kHugePayloadSize); |
| sctp_socket_.Send(DcSctpMessage(kStreamId, kPpid, std::move(payload)), |
| SendOptions()); |
| |
| } else if (mode_ == ActorMode::kLimitedRetransmissionSender) { |
| sctp_socket_.SetBufferedAmountLowThreshold(kStreamId, |
| kBufferedAmountLowThreshold); |
| std::vector<uint8_t> payload(kHugePayloadSize); |
| sctp_socket_.Send(DcSctpMessage(kStreamId, kPpid, std::move(payload)), |
| SendOptions()); |
| |
| } else if (mode == ActorMode::kThroughputReceiver) { |
| thread_->PostDelayedTask( |
| SafeTask(safety_.flag(), [this] { PrintBandwidth(); }), |
| kPrintBandwidthDuration); |
| } |
| } |
| |
| // Returns the average bitrate, stripping the first `remove_first_n` that |
| // represent the time it took to ramp up the congestion control algorithm. |
| double avg_received_bitrate_mbps(size_t remove_first_n = 3) const { |
| std::vector<double> bitrates = received_bitrate_mbps_; |
| bitrates.erase(bitrates.begin(), bitrates.begin() + remove_first_n); |
| |
| double sum = 0; |
| for (double bitrate : bitrates) { |
| sum += bitrate; |
| } |
| |
| return sum / bitrates.size(); |
| } |
| |
| private: |
| std::string log_prefix() const { |
| rtc::StringBuilder sb; |
| sb << log_prefix_; |
| sb << rtc::TimeMillis(); |
| sb << ": "; |
| return sb.Release(); |
| } |
| |
| ActorMode mode_ = ActorMode::kAtRest; |
| const std::string log_prefix_; |
| rtc::Thread* thread_; |
| BoundSocket& emulated_socket_; |
| TaskQueueTimeoutFactory timeout_factory_; |
| webrtc::Random random_; |
| DcSctpSocket sctp_socket_; |
| size_t received_bytes_ = 0; |
| absl::optional<DcSctpMessage> last_received_message_; |
| Timestamp last_bandwidth_printout_; |
| // Per-second received bitrates, in Mbps |
| std::vector<double> received_bitrate_mbps_; |
| webrtc::ScopedTaskSafety safety_; |
| }; |
| |
| class DcSctpSocketNetworkTest : public testing::Test { |
| protected: |
| DcSctpSocketNetworkTest() |
| : options_(MakeOptionsForTest()), |
| emulation_(webrtc::CreateNetworkEmulationManager( |
| {.time_mode = webrtc::TimeMode::kSimulated})) {} |
| |
| void MakeNetwork(const webrtc::BuiltInNetworkBehaviorConfig& config) { |
| webrtc::EmulatedEndpoint* endpoint_a = |
| emulation_->CreateEndpoint(webrtc::EmulatedEndpointConfig()); |
| webrtc::EmulatedEndpoint* endpoint_z = |
| emulation_->CreateEndpoint(webrtc::EmulatedEndpointConfig()); |
| |
| webrtc::EmulatedNetworkNode* node1 = emulation_->CreateEmulatedNode(config); |
| webrtc::EmulatedNetworkNode* node2 = emulation_->CreateEmulatedNode(config); |
| |
| emulation_->CreateRoute(endpoint_a, {node1}, endpoint_z); |
| emulation_->CreateRoute(endpoint_z, {node2}, endpoint_a); |
| |
| emulated_socket_a_.Bind(endpoint_a); |
| emulated_socket_z_.Bind(endpoint_z); |
| |
| emulated_socket_a_.SetDestination(emulated_socket_z_); |
| emulated_socket_z_.SetDestination(emulated_socket_a_); |
| } |
| |
| void Sleep(webrtc::TimeDelta duration) { |
| // Sleep in one-millisecond increments, to let timers expire when expected. |
| for (int i = 0; i < duration.ms(); ++i) { |
| emulation_->time_controller()->AdvanceTime(webrtc::TimeDelta::Millis(1)); |
| } |
| } |
| |
| DcSctpOptions options_; |
| std::unique_ptr<webrtc::NetworkEmulationManager> emulation_; |
| BoundSocket emulated_socket_a_; |
| BoundSocket emulated_socket_z_; |
| }; |
| |
| TEST_F(DcSctpSocketNetworkTest, CanConnectAndShutdown) { |
| webrtc::BuiltInNetworkBehaviorConfig pipe_config; |
| MakeNetwork(pipe_config); |
| |
| SctpActor sender("A", emulated_socket_a_, options_); |
| SctpActor receiver("Z", emulated_socket_z_, options_); |
| EXPECT_THAT(sender.sctp_socket().state(), SocketState::kClosed); |
| |
| sender.sctp_socket().Connect(); |
| Sleep(kAWhile); |
| EXPECT_THAT(sender.sctp_socket().state(), SocketState::kConnected); |
| |
| sender.sctp_socket().Shutdown(); |
| Sleep(kAWhile); |
| EXPECT_THAT(sender.sctp_socket().state(), SocketState::kClosed); |
| } |
| |
| TEST_F(DcSctpSocketNetworkTest, CanSendLargeMessage) { |
| webrtc::BuiltInNetworkBehaviorConfig pipe_config; |
| pipe_config.queue_delay_ms = 30; |
| MakeNetwork(pipe_config); |
| |
| SctpActor sender("A", emulated_socket_a_, options_); |
| SctpActor receiver("Z", emulated_socket_z_, options_); |
| sender.sctp_socket().Connect(); |
| |
| constexpr size_t kPayloadSize = 100 * 1024; |
| |
| std::vector<uint8_t> payload(kPayloadSize); |
| sender.sctp_socket().Send(DcSctpMessage(kStreamId, kPpid, payload), |
| SendOptions()); |
| |
| Sleep(kAWhile); |
| |
| ASSERT_HAS_VALUE_AND_ASSIGN(DcSctpMessage message, |
| receiver.ConsumeReceivedMessage()); |
| |
| EXPECT_THAT(message.payload(), SizeIs(kPayloadSize)); |
| |
| sender.sctp_socket().Shutdown(); |
| Sleep(kAWhile); |
| } |
| |
| TEST_F(DcSctpSocketNetworkTest, CanSendMessagesReliablyWithLowBandwidth) { |
| webrtc::BuiltInNetworkBehaviorConfig pipe_config; |
| pipe_config.queue_delay_ms = 30; |
| pipe_config.link_capacity = DataRate::KilobitsPerSec(1000); |
| MakeNetwork(pipe_config); |
| |
| SctpActor sender("A", emulated_socket_a_, options_); |
| SctpActor receiver("Z", emulated_socket_z_, options_); |
| sender.sctp_socket().Connect(); |
| |
| sender.SetActorMode(ActorMode::kThroughputSender); |
| receiver.SetActorMode(ActorMode::kThroughputReceiver); |
| |
| Sleep(kBenchmarkRuntime); |
| sender.SetActorMode(ActorMode::kAtRest); |
| receiver.SetActorMode(ActorMode::kAtRest); |
| |
| Sleep(kAWhile); |
| |
| sender.sctp_socket().Shutdown(); |
| |
| Sleep(kAWhile); |
| |
| // Verify that the bitrates are in the range of 0.5-1.0 Mbps. |
| double bitrate = receiver.avg_received_bitrate_mbps(); |
| EXPECT_THAT(bitrate, AllOf(Ge(0.5), Le(1.0))); |
| } |
| |
| TEST_F(DcSctpSocketNetworkTest, |
| DCSCTP_NDEBUG_TEST(CanSendMessagesReliablyWithMediumBandwidth)) { |
| webrtc::BuiltInNetworkBehaviorConfig pipe_config; |
| pipe_config.queue_delay_ms = 30; |
| pipe_config.link_capacity = DataRate::KilobitsPerSec(18000); |
| MakeNetwork(pipe_config); |
| |
| SctpActor sender("A", emulated_socket_a_, options_); |
| SctpActor receiver("Z", emulated_socket_z_, options_); |
| sender.sctp_socket().Connect(); |
| |
| sender.SetActorMode(ActorMode::kThroughputSender); |
| receiver.SetActorMode(ActorMode::kThroughputReceiver); |
| |
| Sleep(kBenchmarkRuntime); |
| sender.SetActorMode(ActorMode::kAtRest); |
| receiver.SetActorMode(ActorMode::kAtRest); |
| |
| Sleep(kAWhile); |
| |
| sender.sctp_socket().Shutdown(); |
| |
| Sleep(kAWhile); |
| |
| // Verify that the bitrates are in the range of 16-18 Mbps. |
| double bitrate = receiver.avg_received_bitrate_mbps(); |
| EXPECT_THAT(bitrate, AllOf(Ge(16), Le(18))); |
| } |
| |
| TEST_F(DcSctpSocketNetworkTest, CanSendMessagesReliablyWithMuchPacketLoss) { |
| webrtc::BuiltInNetworkBehaviorConfig config; |
| config.queue_delay_ms = 30; |
| config.loss_percent = 1; |
| MakeNetwork(config); |
| |
| SctpActor sender("A", emulated_socket_a_, options_); |
| SctpActor receiver("Z", emulated_socket_z_, options_); |
| sender.sctp_socket().Connect(); |
| |
| sender.SetActorMode(ActorMode::kThroughputSender); |
| receiver.SetActorMode(ActorMode::kThroughputReceiver); |
| |
| Sleep(kBenchmarkRuntime); |
| sender.SetActorMode(ActorMode::kAtRest); |
| receiver.SetActorMode(ActorMode::kAtRest); |
| |
| Sleep(kAWhile); |
| |
| sender.sctp_socket().Shutdown(); |
| |
| Sleep(kAWhile); |
| |
| // TCP calculator gives: 1200 MTU, 60ms RTT and 1% packet loss -> 1.6Mbps. |
| // This test is doing slightly better (doesn't have any additional header |
| // overhead etc). Verify that the bitrates are in the range of 1.5-2.5 Mbps. |
| double bitrate = receiver.avg_received_bitrate_mbps(); |
| EXPECT_THAT(bitrate, AllOf(Ge(1.5), Le(2.5))); |
| } |
| |
| TEST_F(DcSctpSocketNetworkTest, DCSCTP_NDEBUG_TEST(HasHighBandwidth)) { |
| webrtc::BuiltInNetworkBehaviorConfig pipe_config; |
| pipe_config.queue_delay_ms = 30; |
| MakeNetwork(pipe_config); |
| |
| SctpActor sender("A", emulated_socket_a_, options_); |
| SctpActor receiver("Z", emulated_socket_z_, options_); |
| sender.sctp_socket().Connect(); |
| |
| sender.SetActorMode(ActorMode::kThroughputSender); |
| receiver.SetActorMode(ActorMode::kThroughputReceiver); |
| |
| Sleep(kBenchmarkRuntime); |
| |
| sender.SetActorMode(ActorMode::kAtRest); |
| receiver.SetActorMode(ActorMode::kAtRest); |
| Sleep(kAWhile); |
| |
| sender.sctp_socket().Shutdown(); |
| Sleep(kAWhile); |
| |
| // Verify that the bitrate is in the range of 540-640 Mbps |
| double bitrate = receiver.avg_received_bitrate_mbps(); |
| EXPECT_THAT(bitrate, AllOf(Ge(520), Le(640))); |
| } |
| } // namespace |
| } // namespace dcsctp |