mollyim/webrtc
1
0
Fork 0
mirror of https://github.com/mollyim/webrtc.git synced 2025-05-15 23:01:21 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions
webrtc/net/dcsctp/socket/stream_reset_handler_test.cc
Victor Boivie de276cf049 dcsctp: Remove initial TSN from reassembly queue 
With a previous refactoring, which made the data tracker responsible for
ensuring that the reassembly queue doesn't see any duplicate received
chunks, it no longer needs to know the initial peer's TSN. Removing.

Bug: None
Change-Id: I0e2aef1de0293f1860b46dee0089757c9c300aea
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/345701
Commit-Queue: Victor Boivie <boivie@webrtc.org>
Reviewed-by: Florent Castelli <orphis@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#41997}
2024-04-04 19:19:47 +00:00

914 lines
38 KiB
C++
Raw Blame History

/*
* 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 "net/dcsctp/socket/stream_reset_handler.h"
#include <array>
#include <cstdint>
#include <memory>
#include <type_traits>
#include <vector>
#include "absl/types/optional.h"
#include "api/array_view.h"
#include "api/task_queue/task_queue_base.h"
#include "net/dcsctp/common/handover_testing.h"
#include "net/dcsctp/common/internal_types.h"
#include "net/dcsctp/packet/chunk/forward_tsn_common.h"
#include "net/dcsctp/packet/chunk/reconfig_chunk.h"
#include "net/dcsctp/packet/parameter/incoming_ssn_reset_request_parameter.h"
#include "net/dcsctp/packet/parameter/outgoing_ssn_reset_request_parameter.h"
#include "net/dcsctp/packet/parameter/parameter.h"
#include "net/dcsctp/packet/parameter/reconfiguration_response_parameter.h"
#include "net/dcsctp/public/dcsctp_message.h"
#include "net/dcsctp/public/types.h"
#include "net/dcsctp/rx/data_tracker.h"
#include "net/dcsctp/rx/reassembly_queue.h"
#include "net/dcsctp/socket/mock_context.h"
#include "net/dcsctp/socket/mock_dcsctp_socket_callbacks.h"
#include "net/dcsctp/testing/data_generator.h"
#include "net/dcsctp/testing/testing_macros.h"
#include "net/dcsctp/timer/timer.h"
#include "net/dcsctp/tx/mock_send_queue.h"
#include "net/dcsctp/tx/retransmission_queue.h"
#include "rtc_base/gunit.h"
#include "test/gmock.h"
namespace dcsctp {
namespace {
using ::testing::IsEmpty;
using ::testing::NiceMock;
using ::testing::Property;
using ::testing::Return;
using ::testing::SizeIs;
using ::testing::UnorderedElementsAre;
using ::webrtc::TimeDelta;
using ResponseResult = ReconfigurationResponseParameter::Result;
using SkippedStream = AnyForwardTsnChunk::SkippedStream;
constexpr TSN kMyInitialTsn = MockContext::MyInitialTsn();
constexpr ReconfigRequestSN kMyInitialReqSn = ReconfigRequestSN(*kMyInitialTsn);
constexpr TSN kPeerInitialTsn = MockContext::PeerInitialTsn();
constexpr ReconfigRequestSN kPeerInitialReqSn =
ReconfigRequestSN(*kPeerInitialTsn);
constexpr uint32_t kArwnd = 131072;
constexpr TimeDelta kRto = TimeDelta::Millis(250);
constexpr std::array<uint8_t, 4> kShortPayload = {1, 2, 3, 4};
MATCHER_P3(SctpMessageIs, stream_id, ppid, expected_payload, "") {
if (arg.stream_id() != stream_id) {
*result_listener << "the stream_id is " << *arg.stream_id();
return false;
}
if (arg.ppid() != ppid) {
*result_listener << "the ppid is " << *arg.ppid();
return false;
}
if (std::vector<uint8_t>(arg.payload().begin(), arg.payload().end()) !=
std::vector<uint8_t>(expected_payload.begin(), expected_payload.end())) {
*result_listener << "the payload is wrong";
return false;
}
return true;
}
TSN AddTo(TSN tsn, int delta) {
return TSN(*tsn + delta);
}
ReconfigRequestSN AddTo(ReconfigRequestSN req_sn, int delta) {
return ReconfigRequestSN(*req_sn + delta);
}
class StreamResetHandlerTest : public testing::Test {
protected:
StreamResetHandlerTest()
: ctx_(&callbacks_),
timer_manager_([this](webrtc::TaskQueueBase::DelayPrecision precision) {
return callbacks_.CreateTimeout(precision);
}),
delayed_ack_timer_(timer_manager_.CreateTimer(
"test/delayed_ack",
[]() { return TimeDelta::Zero(); },
TimerOptions(TimeDelta::Zero()))),
t3_rtx_timer_(timer_manager_.CreateTimer(
"test/t3_rtx",
[]() { return TimeDelta::Zero(); },
TimerOptions(TimeDelta::Zero()))),
data_tracker_(std::make_unique<DataTracker>("log: ",
delayed_ack_timer_.get(),
kPeerInitialTsn)),
reasm_(std::make_unique<ReassemblyQueue>("log: ", kArwnd)),
retransmission_queue_(std::make_unique<RetransmissionQueue>(
"",
&callbacks_,
kMyInitialTsn,
kArwnd,
producer_,
[](TimeDelta rtt) {},
[]() {},
*t3_rtx_timer_,
DcSctpOptions())),
handler_(
std::make_unique<StreamResetHandler>("log: ",
&ctx_,
&timer_manager_,
data_tracker_.get(),
reasm_.get(),
retransmission_queue_.get())) {
EXPECT_CALL(ctx_, current_rto).WillRepeatedly(Return(kRto));
}
void AdvanceTime(TimeDelta duration) {
callbacks_.AdvanceTime(duration);
for (;;) {
absl::optional<TimeoutID> timeout_id = callbacks_.GetNextExpiredTimeout();
if (!timeout_id.has_value()) {
break;
}
timer_manager_.HandleTimeout(*timeout_id);
}
}
// Handles the passed in RE-CONFIG `chunk` and returns the responses
// that are sent in the response RE-CONFIG.
std::vector<ReconfigurationResponseParameter> HandleAndCatchResponse(
ReConfigChunk chunk) {
handler_->HandleReConfig(std::move(chunk));
std::vector<uint8_t> payload = callbacks_.ConsumeSentPacket();
if (payload.empty()) {
EXPECT_TRUE(false);
return {};
}
std::vector<ReconfigurationResponseParameter> responses;
absl::optional<SctpPacket> p = SctpPacket::Parse(payload, DcSctpOptions());
if (!p.has_value()) {
EXPECT_TRUE(false);
return {};
}
if (p->descriptors().size() != 1) {
EXPECT_TRUE(false);
return {};
}
absl::optional<ReConfigChunk> response_chunk =
ReConfigChunk::Parse(p->descriptors()[0].data);
if (!response_chunk.has_value()) {
EXPECT_TRUE(false);
return {};
}
for (const auto& desc : response_chunk->parameters().descriptors()) {
if (desc.type == ReconfigurationResponseParameter::kType) {
absl::optional<ReconfigurationResponseParameter> response =
ReconfigurationResponseParameter::Parse(desc.data);
if (!response.has_value()) {
EXPECT_TRUE(false);
return {};
}
responses.emplace_back(*std::move(response));
}
}
return responses;
}
void PerformHandover() {
EXPECT_TRUE(handler_->GetHandoverReadiness().IsReady());
EXPECT_TRUE(data_tracker_->GetHandoverReadiness().IsReady());
EXPECT_TRUE(reasm_->GetHandoverReadiness().IsReady());
EXPECT_TRUE(retransmission_queue_->GetHandoverReadiness().IsReady());
DcSctpSocketHandoverState state;
handler_->AddHandoverState(state);
data_tracker_->AddHandoverState(state);
reasm_->AddHandoverState(state);
retransmission_queue_->AddHandoverState(state);
g_handover_state_transformer_for_test(&state);
data_tracker_ = std::make_unique<DataTracker>(
"log: ", delayed_ack_timer_.get(), kPeerInitialTsn);
data_tracker_->RestoreFromState(state);
reasm_ = std::make_unique<ReassemblyQueue>("log: ", kArwnd);
reasm_->RestoreFromState(state);
retransmission_queue_ = std::make_unique<RetransmissionQueue>(
"", &callbacks_, kMyInitialTsn, kArwnd, producer_, [](TimeDelta rtt) {},
[]() {}, *t3_rtx_timer_, DcSctpOptions(),
/*supports_partial_reliability=*/true,
/*use_message_interleaving=*/false);
retransmission_queue_->RestoreFromState(state);
handler_ = std::make_unique<StreamResetHandler>(
"log: ", &ctx_, &timer_manager_, data_tracker_.get(), reasm_.get(),
retransmission_queue_.get(), &state);
}
DataGenerator gen_;
NiceMock<MockDcSctpSocketCallbacks> callbacks_;
NiceMock<MockContext> ctx_;
NiceMock<MockSendQueue> producer_;
TimerManager timer_manager_;
std::unique_ptr<Timer> delayed_ack_timer_;
std::unique_ptr<Timer> t3_rtx_timer_;
std::unique_ptr<DataTracker> data_tracker_;
std::unique_ptr<ReassemblyQueue> reasm_;
std::unique_ptr<RetransmissionQueue> retransmission_queue_;
std::unique_ptr<StreamResetHandler> handler_;
};
TEST_F(StreamResetHandlerTest, ChunkWithNoParametersReturnsError) {
EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
EXPECT_CALL(callbacks_, OnError).Times(1);
handler_->HandleReConfig(ReConfigChunk(Parameters()));
}
TEST_F(StreamResetHandlerTest, ChunkWithInvalidParametersReturnsError) {
Parameters::Builder builder;
// Two OutgoingSSNResetRequestParameter in a RE-CONFIG is not valid.
builder.Add(OutgoingSSNResetRequestParameter(ReconfigRequestSN(1),
ReconfigRequestSN(10),
kPeerInitialTsn, {StreamID(1)}));
builder.Add(OutgoingSSNResetRequestParameter(ReconfigRequestSN(2),
ReconfigRequestSN(10),
kPeerInitialTsn, {StreamID(2)}));
EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
EXPECT_CALL(callbacks_, OnError).Times(1);
handler_->HandleReConfig(ReConfigChunk(builder.Build()));
}
TEST_F(StreamResetHandlerTest, FailToDeliverWithoutResettingStream) {
reasm_->Add(kPeerInitialTsn, gen_.Ordered({1, 2, 3, 4}, "BE"));
reasm_->Add(AddTo(kPeerInitialTsn, 1), gen_.Ordered({1, 2, 3, 4}, "BE"));
data_tracker_->Observe(kPeerInitialTsn);
data_tracker_->Observe(AddTo(kPeerInitialTsn, 1));
EXPECT_THAT(reasm_->FlushMessages(),
UnorderedElementsAre(
SctpMessageIs(StreamID(1), PPID(53), kShortPayload),
SctpMessageIs(StreamID(1), PPID(53), kShortPayload)));
gen_.ResetStream();
reasm_->Add(AddTo(kPeerInitialTsn, 2), gen_.Ordered({1, 2, 3, 4}, "BE"));
EXPECT_THAT(reasm_->FlushMessages(), IsEmpty());
}
TEST_F(StreamResetHandlerTest, ResetStreamsNotDeferred) {
reasm_->Add(kPeerInitialTsn, gen_.Ordered({1, 2, 3, 4}, "BE"));
reasm_->Add(AddTo(kPeerInitialTsn, 1), gen_.Ordered({1, 2, 3, 4}, "BE"));
data_tracker_->Observe(kPeerInitialTsn);
data_tracker_->Observe(AddTo(kPeerInitialTsn, 1));
EXPECT_THAT(reasm_->FlushMessages(),
UnorderedElementsAre(
SctpMessageIs(StreamID(1), PPID(53), kShortPayload),
SctpMessageIs(StreamID(1), PPID(53), kShortPayload)));
Parameters::Builder builder;
builder.Add(OutgoingSSNResetRequestParameter(
kPeerInitialReqSn, ReconfigRequestSN(3), AddTo(kPeerInitialTsn, 1),
{StreamID(1)}));
std::vector<ReconfigurationResponseParameter> responses =
HandleAndCatchResponse(ReConfigChunk(builder.Build()));
EXPECT_THAT(responses, SizeIs(1));
EXPECT_EQ(responses[0].result(), ResponseResult::kSuccessPerformed);
gen_.ResetStream();
reasm_->Add(AddTo(kPeerInitialTsn, 2), gen_.Ordered({1, 2, 3, 4}, "BE"));
EXPECT_THAT(reasm_->FlushMessages(),
UnorderedElementsAre(
SctpMessageIs(StreamID(1), PPID(53), kShortPayload)));
}
TEST_F(StreamResetHandlerTest, ResetStreamsDeferred) {
constexpr StreamID kStreamId = StreamID(1);
data_tracker_->Observe(TSN(10));
reasm_->Add(TSN(10), gen_.Ordered({1, 2, 3, 4}, "BE", {.mid = MID(0)}));
data_tracker_->Observe(TSN(11));
reasm_->Add(TSN(11), gen_.Ordered({1, 2, 3, 4}, "BE", {.mid = MID(1)}));
EXPECT_THAT(
reasm_->FlushMessages(),
UnorderedElementsAre(SctpMessageIs(kStreamId, PPID(53), kShortPayload),
SctpMessageIs(kStreamId, PPID(53), kShortPayload)));
Parameters::Builder builder;
builder.Add(OutgoingSSNResetRequestParameter(
ReconfigRequestSN(10), ReconfigRequestSN(3), TSN(13), {kStreamId}));
EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())),
ElementsAre(Property(&ReconfigurationResponseParameter::result,
ResponseResult::kInProgress)));
data_tracker_->Observe(TSN(15));
reasm_->Add(TSN(15), gen_.Ordered({1, 2, 3, 4}, "BE",
{.mid = MID(1), .ppid = PPID(5)}));
data_tracker_->Observe(TSN(14));
reasm_->Add(TSN(14), gen_.Ordered({1, 2, 3, 4}, "BE",
{.mid = MID(0), .ppid = PPID(4)}));
data_tracker_->Observe(TSN(13));
reasm_->Add(TSN(13), gen_.Ordered({1, 2, 3, 4}, "BE",
{.mid = MID(3), .ppid = PPID(3)}));
data_tracker_->Observe(TSN(12));
reasm_->Add(TSN(12), gen_.Ordered({1, 2, 3, 4}, "BE",
{.mid = MID(2), .ppid = PPID(2)}));
builder.Add(OutgoingSSNResetRequestParameter(
ReconfigRequestSN(11), ReconfigRequestSN(4), TSN(13), {kStreamId}));
EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())),
ElementsAre(Property(&ReconfigurationResponseParameter::result,
ResponseResult::kSuccessPerformed)));
EXPECT_THAT(
reasm_->FlushMessages(),
UnorderedElementsAre(SctpMessageIs(kStreamId, PPID(2), kShortPayload),
SctpMessageIs(kStreamId, PPID(3), kShortPayload),
SctpMessageIs(kStreamId, PPID(4), kShortPayload),
SctpMessageIs(kStreamId, PPID(5), kShortPayload)));
}
TEST_F(StreamResetHandlerTest, ResetStreamsDeferredOnlySelectedStreams) {
// This test verifies the receiving behavior of receiving messages on
// streams 1, 2 and 3, and receiving a reset request on stream 1, 2, causing
// deferred reset processing.
// Reset stream 1,2 with "last assigned TSN=12"
Parameters::Builder builder;
builder.Add(OutgoingSSNResetRequestParameter(ReconfigRequestSN(10),
ReconfigRequestSN(3), TSN(12),
{StreamID(1), StreamID(2)}));
EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())),
ElementsAre(Property(&ReconfigurationResponseParameter::result,
ResponseResult::kInProgress)));
// TSN 10, SID 1 - before TSN 12 -> deliver
data_tracker_->Observe(TSN(10));
reasm_->Add(TSN(10), gen_.Ordered({1, 2, 3, 4}, "BE",
{.stream_id = StreamID(1),
.mid = MID(0),
.ppid = PPID(1001)}));
// TSN 11, SID 2 - before TSN 12 -> deliver
data_tracker_->Observe(TSN(11));
reasm_->Add(TSN(11), gen_.Ordered({1, 2, 3, 4}, "BE",
{.stream_id = StreamID(2),
.mid = MID(0),
.ppid = PPID(1002)}));
// TSN 12, SID 3 - at TSN 12 -> deliver
data_tracker_->Observe(TSN(12));
reasm_->Add(TSN(12), gen_.Ordered({1, 2, 3, 4}, "BE",
{.stream_id = StreamID(3),
.mid = MID(0),
.ppid = PPID(1003)}));
// TSN 13, SID 1 - after TSN 12 and SID=1 -> defer
data_tracker_->Observe(TSN(13));
reasm_->Add(TSN(13), gen_.Ordered({1, 2, 3, 4}, "BE",
{.stream_id = StreamID(1),
.mid = MID(0),
.ppid = PPID(1004)}));
// TSN 14, SID 2 - after TSN 12 and SID=2 -> defer
data_tracker_->Observe(TSN(14));
reasm_->Add(TSN(14), gen_.Ordered({1, 2, 3, 4}, "BE",
{.stream_id = StreamID(2),
.mid = MID(0),
.ppid = PPID(1005)}));
// TSN 15, SID 3 - after TSN 12, but SID 3 is not reset -> deliver
data_tracker_->Observe(TSN(15));
reasm_->Add(TSN(15), gen_.Ordered({1, 2, 3, 4}, "BE",
{.stream_id = StreamID(3),
.mid = MID(1),
.ppid = PPID(1006)}));
EXPECT_THAT(reasm_->FlushMessages(),
UnorderedElementsAre(
SctpMessageIs(StreamID(1), PPID(1001), kShortPayload),
SctpMessageIs(StreamID(2), PPID(1002), kShortPayload),
SctpMessageIs(StreamID(3), PPID(1003), kShortPayload),
SctpMessageIs(StreamID(3), PPID(1006), kShortPayload)));
builder.Add(OutgoingSSNResetRequestParameter(ReconfigRequestSN(11),
ReconfigRequestSN(3), TSN(13),
{StreamID(1), StreamID(2)}));
EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())),
ElementsAre(Property(&ReconfigurationResponseParameter::result,
ResponseResult::kSuccessPerformed)));
EXPECT_THAT(reasm_->FlushMessages(),
UnorderedElementsAre(
SctpMessageIs(StreamID(1), PPID(1004), kShortPayload),
SctpMessageIs(StreamID(2), PPID(1005), kShortPayload)));
}
TEST_F(StreamResetHandlerTest, ResetStreamsDefersForwardTsn) {
// This test verifies that FORWARD-TSNs are deferred if they want to move
// the cumulative ack TSN point past sender's last assigned TSN.
static constexpr StreamID kStreamId = StreamID(42);
// Simulate sender sends:
// * TSN 10 (SSN=0, BE, lost),
// * TSN 11 (SSN=1, BE, lost),
// * TSN 12 (SSN=2, BE, lost)
// * RESET THE STREAM
// * TSN 13 (SSN=0, B, received)
// * TSN 14 (SSN=0, E, lost),
// * TSN 15 (SSN=1, BE, received)
Parameters::Builder builder;
builder.Add(OutgoingSSNResetRequestParameter(
ReconfigRequestSN(10), ReconfigRequestSN(3), TSN(12), {kStreamId}));
EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())),
ElementsAre(Property(&ReconfigurationResponseParameter::result,
ResponseResult::kInProgress)));
// TSN 13, B, after TSN=12 -> defer
data_tracker_->Observe(TSN(13));
reasm_->Add(TSN(13),
gen_.Ordered(
{1, 2, 3, 4}, "B",
{.stream_id = kStreamId, .mid = MID(0), .ppid = PPID(1004)}));
// TSN 15, BE, after TSN=12 -> defer
data_tracker_->Observe(TSN(15));
reasm_->Add(TSN(15),
gen_.Ordered(
{1, 2, 3, 4}, "BE",
{.stream_id = kStreamId, .mid = MID(1), .ppid = PPID(1005)}));
// Time passes, sender decides to send FORWARD-TSN up to the RESET.
data_tracker_->HandleForwardTsn(TSN(12));
reasm_->HandleForwardTsn(
TSN(12), std::vector<SkippedStream>({SkippedStream(kStreamId, SSN(2))}));
// The receiver sends a SACK in response to that. The stream hasn't been
// reset yet, but the sender now decides that TSN=13-14 is to be skipped.
// As this has a TSN 14, after TSN=12 -> defer it.
data_tracker_->HandleForwardTsn(TSN(14));
reasm_->HandleForwardTsn(
TSN(14), std::vector<SkippedStream>({SkippedStream(kStreamId, SSN(0))}));
// Reset the stream -> deferred TSNs should be re-added.
builder.Add(OutgoingSSNResetRequestParameter(
ReconfigRequestSN(11), ReconfigRequestSN(3), TSN(12), {kStreamId}));
EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())),
ElementsAre(Property(&ReconfigurationResponseParameter::result,
ResponseResult::kSuccessPerformed)));
EXPECT_THAT(reasm_->FlushMessages(),
UnorderedElementsAre(
SctpMessageIs(kStreamId, PPID(1005), kShortPayload)));
}
TEST_F(StreamResetHandlerTest, SendOutgoingRequestDirectly) {
EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest();
ASSERT_TRUE(reconfig.has_value());
ASSERT_HAS_VALUE_AND_ASSIGN(
OutgoingSSNResetRequestParameter req,
reconfig->parameters().get<OutgoingSSNResetRequestParameter>());
EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn);
EXPECT_EQ(req.sender_last_assigned_tsn(),
TSN(*retransmission_queue_->next_tsn() - 1));
EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(42)));
}
TEST_F(StreamResetHandlerTest, ResetMultipleStreamsInOneRequest) {
EXPECT_CALL(producer_, PrepareResetStream(StreamID(40)));
EXPECT_CALL(producer_, PrepareResetStream(StreamID(41)));
EXPECT_CALL(producer_, PrepareResetStream(StreamID(42))).Times(2);
EXPECT_CALL(producer_, PrepareResetStream(StreamID(43)));
EXPECT_CALL(producer_, PrepareResetStream(StreamID(44)));
handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
handler_->ResetStreams(
std::vector<StreamID>({StreamID(43), StreamID(44), StreamID(41)}));
handler_->ResetStreams(std::vector<StreamID>({StreamID(42), StreamID(40)}));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(
std::vector<StreamID>({StreamID(40), StreamID(41), StreamID(42),
StreamID(43), StreamID(44)})));
absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest();
ASSERT_TRUE(reconfig.has_value());
ASSERT_HAS_VALUE_AND_ASSIGN(
OutgoingSSNResetRequestParameter req,
reconfig->parameters().get<OutgoingSSNResetRequestParameter>());
EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn);
EXPECT_EQ(req.sender_last_assigned_tsn(),
TSN(*retransmission_queue_->next_tsn() - 1));
EXPECT_THAT(req.stream_ids(),
UnorderedElementsAre(StreamID(40), StreamID(41), StreamID(42),
StreamID(43), StreamID(44)));
}
TEST_F(StreamResetHandlerTest, SendOutgoingRequestDeferred) {
EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset())
.WillOnce(Return(false))
.WillOnce(Return(false))
.WillOnce(Return(true));
EXPECT_FALSE(handler_->MakeStreamResetRequest().has_value());
EXPECT_FALSE(handler_->MakeStreamResetRequest().has_value());
EXPECT_TRUE(handler_->MakeStreamResetRequest().has_value());
}
TEST_F(StreamResetHandlerTest, SendOutgoingResettingOnPositiveResponse) {
EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest();
ASSERT_TRUE(reconfig.has_value());
ASSERT_HAS_VALUE_AND_ASSIGN(
OutgoingSSNResetRequestParameter req,
reconfig->parameters().get<OutgoingSSNResetRequestParameter>());
Parameters::Builder builder;
builder.Add(ReconfigurationResponseParameter(
req.request_sequence_number(), ResponseResult::kSuccessPerformed));
ReConfigChunk response_reconfig(builder.Build());
EXPECT_CALL(producer_, CommitResetStreams);
EXPECT_CALL(producer_, RollbackResetStreams).Times(0);
// Processing a response shouldn't result in sending anything.
EXPECT_CALL(callbacks_, OnError).Times(0);
EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
handler_->HandleReConfig(std::move(response_reconfig));
}
TEST_F(StreamResetHandlerTest, SendOutgoingResetRollbackOnError) {
EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest();
ASSERT_TRUE(reconfig.has_value());
ASSERT_HAS_VALUE_AND_ASSIGN(
OutgoingSSNResetRequestParameter req,
reconfig->parameters().get<OutgoingSSNResetRequestParameter>());
Parameters::Builder builder;
builder.Add(ReconfigurationResponseParameter(
req.request_sequence_number(), ResponseResult::kErrorBadSequenceNumber));
ReConfigChunk response_reconfig(builder.Build());
EXPECT_CALL(producer_, CommitResetStreams).Times(0);
EXPECT_CALL(producer_, RollbackResetStreams);
// Only requests should result in sending responses.
EXPECT_CALL(callbacks_, OnError).Times(0);
EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
handler_->HandleReConfig(std::move(response_reconfig));
}
TEST_F(StreamResetHandlerTest, SendOutgoingResetRetransmitOnInProgress) {
static constexpr StreamID kStreamToReset = StreamID(42);
EXPECT_CALL(producer_, PrepareResetStream(kStreamToReset));
handler_->ResetStreams(std::vector<StreamID>({kStreamToReset}));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(std::vector<StreamID>({kStreamToReset})));
absl::optional<ReConfigChunk> reconfig1 = handler_->MakeStreamResetRequest();
ASSERT_TRUE(reconfig1.has_value());
ASSERT_HAS_VALUE_AND_ASSIGN(
OutgoingSSNResetRequestParameter req1,
reconfig1->parameters().get<OutgoingSSNResetRequestParameter>());
// Simulate that the peer responded "In Progress".
Parameters::Builder builder;
builder.Add(ReconfigurationResponseParameter(req1.request_sequence_number(),
ResponseResult::kInProgress));
ReConfigChunk response_reconfig(builder.Build());
EXPECT_CALL(producer_, CommitResetStreams()).Times(0);
EXPECT_CALL(producer_, RollbackResetStreams()).Times(0);
// Processing a response shouldn't result in sending anything.
EXPECT_CALL(callbacks_, OnError).Times(0);
EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
handler_->HandleReConfig(std::move(response_reconfig));
// Let some time pass, so that the reconfig timer expires, and retries the
// same request.
EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(1);
AdvanceTime(kRto);
std::vector<uint8_t> payload = callbacks_.ConsumeSentPacket();
ASSERT_FALSE(payload.empty());
ASSERT_HAS_VALUE_AND_ASSIGN(SctpPacket packet,
SctpPacket::Parse(payload, DcSctpOptions()));
ASSERT_THAT(packet.descriptors(), SizeIs(1));
ASSERT_HAS_VALUE_AND_ASSIGN(
ReConfigChunk reconfig2,
ReConfigChunk::Parse(packet.descriptors()[0].data));
ASSERT_HAS_VALUE_AND_ASSIGN(
OutgoingSSNResetRequestParameter req2,
reconfig2.parameters().get<OutgoingSSNResetRequestParameter>());
EXPECT_EQ(req2.request_sequence_number(),
AddTo(req1.request_sequence_number(), 1));
EXPECT_THAT(req2.stream_ids(), UnorderedElementsAre(kStreamToReset));
}
TEST_F(StreamResetHandlerTest, ResetWhileRequestIsSentWillQueue) {
EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
absl::optional<ReConfigChunk> reconfig1 = handler_->MakeStreamResetRequest();
ASSERT_TRUE(reconfig1.has_value());
ASSERT_HAS_VALUE_AND_ASSIGN(
OutgoingSSNResetRequestParameter req1,
reconfig1->parameters().get<OutgoingSSNResetRequestParameter>());
EXPECT_EQ(req1.request_sequence_number(), kMyInitialReqSn);
EXPECT_EQ(req1.sender_last_assigned_tsn(),
AddTo(retransmission_queue_->next_tsn(), -1));
EXPECT_THAT(req1.stream_ids(), UnorderedElementsAre(StreamID(42)));
// Streams reset while the request is in-flight will be queued.
EXPECT_CALL(producer_, PrepareResetStream(StreamID(41)));
EXPECT_CALL(producer_, PrepareResetStream(StreamID(43)));
StreamID stream_ids[] = {StreamID(41), StreamID(43)};
handler_->ResetStreams(stream_ids);
EXPECT_EQ(handler_->MakeStreamResetRequest(), absl::nullopt);
Parameters::Builder builder;
builder.Add(ReconfigurationResponseParameter(
req1.request_sequence_number(), ResponseResult::kSuccessPerformed));
ReConfigChunk response_reconfig(builder.Build());
EXPECT_CALL(producer_, CommitResetStreams()).Times(1);
EXPECT_CALL(producer_, RollbackResetStreams()).Times(0);
// Processing a response shouldn't result in sending anything.
EXPECT_CALL(callbacks_, OnError).Times(0);
EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
handler_->HandleReConfig(std::move(response_reconfig));
// Response has been processed. A new request can be sent.
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(std::vector<StreamID>({StreamID(41), StreamID(43)})));
absl::optional<ReConfigChunk> reconfig2 = handler_->MakeStreamResetRequest();
ASSERT_TRUE(reconfig2.has_value());
ASSERT_HAS_VALUE_AND_ASSIGN(
OutgoingSSNResetRequestParameter req2,
reconfig2->parameters().get<OutgoingSSNResetRequestParameter>());
EXPECT_EQ(req2.request_sequence_number(), AddTo(kMyInitialReqSn, 1));
EXPECT_EQ(req2.sender_last_assigned_tsn(),
TSN(*retransmission_queue_->next_tsn() - 1));
EXPECT_THAT(req2.stream_ids(),
UnorderedElementsAre(StreamID(41), StreamID(43)));
}
TEST_F(StreamResetHandlerTest, SendIncomingResetJustReturnsNothingPerformed) {
Parameters::Builder builder;
builder.Add(
IncomingSSNResetRequestParameter(kPeerInitialReqSn, {StreamID(1)}));
std::vector<ReconfigurationResponseParameter> responses =
HandleAndCatchResponse(ReConfigChunk(builder.Build()));
ASSERT_THAT(responses, SizeIs(1));
EXPECT_THAT(responses[0].response_sequence_number(), kPeerInitialReqSn);
EXPECT_THAT(responses[0].result(), ResponseResult::kSuccessNothingToDo);
}
TEST_F(StreamResetHandlerTest, SendSameRequestTwiceIsIdempotent) {
// Simulate that receiving the same chunk twice (due to network issues,
// or retransmissions, causing a RECONFIG to be re-received) is idempotent.
for (int i = 0; i < 2; ++i) {
Parameters::Builder builder;
builder.Add(OutgoingSSNResetRequestParameter(
kPeerInitialReqSn, ReconfigRequestSN(3), AddTo(kPeerInitialTsn, 1),
{StreamID(1)}));
std::vector<ReconfigurationResponseParameter> responses1 =
HandleAndCatchResponse(ReConfigChunk(builder.Build()));
EXPECT_THAT(responses1, SizeIs(1));
EXPECT_EQ(responses1[0].result(), ResponseResult::kInProgress);
}
}
TEST_F(StreamResetHandlerTest,
HandoverIsAllowedOnlyWhenNoStreamIsBeingOrWillBeReset) {
EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_EQ(
handler_->GetHandoverReadiness(),
HandoverReadinessStatus(HandoverUnreadinessReason::kPendingStreamReset));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset())
.WillOnce(Return(true))
.WillOnce(Return(false));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
ASSERT_TRUE(handler_->MakeStreamResetRequest().has_value());
EXPECT_EQ(handler_->GetHandoverReadiness(),
HandoverReadinessStatus(
HandoverUnreadinessReason::kPendingStreamResetRequest));
// Reset more streams while the request is in-flight.
EXPECT_CALL(producer_, PrepareResetStream(StreamID(41)));
EXPECT_CALL(producer_, PrepareResetStream(StreamID(43)));
StreamID stream_ids[] = {StreamID(41), StreamID(43)};
handler_->ResetStreams(stream_ids);
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_EQ(handler_->GetHandoverReadiness(),
HandoverReadinessStatus()
.Add(HandoverUnreadinessReason::kPendingStreamResetRequest)
.Add(HandoverUnreadinessReason::kPendingStreamReset));
// Processing a response to first request.
EXPECT_CALL(producer_, CommitResetStreams()).Times(1);
handler_->HandleReConfig(
ReConfigChunk(Parameters::Builder()
.Add(ReconfigurationResponseParameter(
kMyInitialReqSn, ResponseResult::kSuccessPerformed))
.Build()));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_EQ(
handler_->GetHandoverReadiness(),
HandoverReadinessStatus(HandoverUnreadinessReason::kPendingStreamReset));
// Second request can be sent.
EXPECT_CALL(producer_, HasStreamsReadyToBeReset())
.WillOnce(Return(true))
.WillOnce(Return(false));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(std::vector<StreamID>({StreamID(41), StreamID(43)})));
ASSERT_TRUE(handler_->MakeStreamResetRequest().has_value());
EXPECT_EQ(handler_->GetHandoverReadiness(),
HandoverReadinessStatus(
HandoverUnreadinessReason::kPendingStreamResetRequest));
// Processing a response to second request.
EXPECT_CALL(producer_, CommitResetStreams()).Times(1);
handler_->HandleReConfig(ReConfigChunk(
Parameters::Builder()
.Add(ReconfigurationResponseParameter(
AddTo(kMyInitialReqSn, 1), ResponseResult::kSuccessPerformed))
.Build()));
// Seconds response has been processed. No pending resets.
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(false));
EXPECT_TRUE(handler_->GetHandoverReadiness().IsReady());
}
TEST_F(StreamResetHandlerTest, HandoverInInitialState) {
PerformHandover();
EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest();
ASSERT_TRUE(reconfig.has_value());
ASSERT_HAS_VALUE_AND_ASSIGN(
OutgoingSSNResetRequestParameter req,
reconfig->parameters().get<OutgoingSSNResetRequestParameter>());
EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn);
EXPECT_EQ(req.sender_last_assigned_tsn(),
TSN(*retransmission_queue_->next_tsn() - 1));
EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(42)));
}
TEST_F(StreamResetHandlerTest, HandoverAfterHavingResetOneStream) {
// Reset one stream
{
EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset())
.WillOnce(Return(true))
.WillOnce(Return(false));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
ASSERT_HAS_VALUE_AND_ASSIGN(ReConfigChunk reconfig,
handler_->MakeStreamResetRequest());
ASSERT_HAS_VALUE_AND_ASSIGN(
OutgoingSSNResetRequestParameter req,
reconfig.parameters().get<OutgoingSSNResetRequestParameter>());
EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn);
EXPECT_EQ(req.sender_last_assigned_tsn(),
TSN(*retransmission_queue_->next_tsn() - 1));
EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(42)));
EXPECT_CALL(producer_, CommitResetStreams()).Times(1);
handler_->HandleReConfig(
ReConfigChunk(Parameters::Builder()
.Add(ReconfigurationResponseParameter(
req.request_sequence_number(),
ResponseResult::kSuccessPerformed))
.Build()));
}
PerformHandover();
// Reset another stream after handover
{
EXPECT_CALL(producer_, PrepareResetStream(StreamID(43)));
handler_->ResetStreams(std::vector<StreamID>({StreamID(43)}));
EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
.WillOnce(Return(std::vector<StreamID>({StreamID(43)})));
ASSERT_HAS_VALUE_AND_ASSIGN(ReConfigChunk reconfig,
handler_->MakeStreamResetRequest());
ASSERT_HAS_VALUE_AND_ASSIGN(
OutgoingSSNResetRequestParameter req,
reconfig.parameters().get<OutgoingSSNResetRequestParameter>());
EXPECT_EQ(req.request_sequence_number(),
ReconfigRequestSN(kMyInitialReqSn.value() + 1));
EXPECT_EQ(req.sender_last_assigned_tsn(),
TSN(*retransmission_queue_->next_tsn() - 1));
EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(43)));
}
}
TEST_F(StreamResetHandlerTest, PerformCloseAfterOneFirstFailing) {
// Inject a stream reset on the first expected TSN (which hasn't been seen).
Parameters::Builder builder;
builder.Add(OutgoingSSNResetRequestParameter(
kPeerInitialReqSn, ReconfigRequestSN(3), kPeerInitialTsn, {StreamID(1)}));
// The socket is expected to say "in progress" as that TSN hasn't been seen.
std::vector<ReconfigurationResponseParameter> responses =
HandleAndCatchResponse(ReConfigChunk(builder.Build()));
EXPECT_THAT(responses, SizeIs(1));
EXPECT_EQ(responses[0].result(), ResponseResult::kInProgress);
// Let the socket receive the TSN.
DataGeneratorOptions opts;
opts.mid = MID(0);
reasm_->Add(kPeerInitialTsn, gen_.Ordered({1, 2, 3, 4}, "BE", opts));
data_tracker_->Observe(kPeerInitialTsn);
// And emulate that time has passed, and the peer retries the stream reset,
// but now with an incremented request sequence number.
Parameters::Builder builder2;
builder2.Add(OutgoingSSNResetRequestParameter(
ReconfigRequestSN(*kPeerInitialReqSn + 1), ReconfigRequestSN(3),
kPeerInitialTsn, {StreamID(1)}));
// This is supposed to be handled well.
std::vector<ReconfigurationResponseParameter> responses2 =
HandleAndCatchResponse(ReConfigChunk(builder2.Build()));
EXPECT_THAT(responses2, SizeIs(1));
EXPECT_EQ(responses2[0].result(), ResponseResult::kSuccessPerformed);
}
} // namespace
} // namespace dcsctp
Reference in a new issue View git blame Copy permalink