diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-19 00:47:55 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-19 00:47:55 +0000 |
commit | 26a029d407be480d791972afb5975cf62c9360a6 (patch) | |
tree | f435a8308119effd964b339f76abb83a57c29483 /third_party/libwebrtc/net/dcsctp/rx/reassembly_queue.cc | |
parent | Initial commit. (diff) | |
download | firefox-26a029d407be480d791972afb5975cf62c9360a6.tar.xz firefox-26a029d407be480d791972afb5975cf62c9360a6.zip |
Adding upstream version 124.0.1.upstream/124.0.1
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'third_party/libwebrtc/net/dcsctp/rx/reassembly_queue.cc')
-rw-r--r-- | third_party/libwebrtc/net/dcsctp/rx/reassembly_queue.cc | 281 |
1 files changed, 281 insertions, 0 deletions
diff --git a/third_party/libwebrtc/net/dcsctp/rx/reassembly_queue.cc b/third_party/libwebrtc/net/dcsctp/rx/reassembly_queue.cc new file mode 100644 index 0000000000..573443635c --- /dev/null +++ b/third_party/libwebrtc/net/dcsctp/rx/reassembly_queue.cc @@ -0,0 +1,281 @@ +/* + * 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/rx/reassembly_queue.h" + +#include <stddef.h> + +#include <algorithm> +#include <cstdint> +#include <memory> +#include <set> +#include <string> +#include <utility> +#include <vector> + +#include "absl/strings/string_view.h" +#include "absl/types/optional.h" +#include "api/array_view.h" +#include "net/dcsctp/common/sequence_numbers.h" +#include "net/dcsctp/common/str_join.h" +#include "net/dcsctp/packet/chunk/forward_tsn_common.h" +#include "net/dcsctp/packet/data.h" +#include "net/dcsctp/packet/parameter/outgoing_ssn_reset_request_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/interleaved_reassembly_streams.h" +#include "net/dcsctp/rx/reassembly_streams.h" +#include "net/dcsctp/rx/traditional_reassembly_streams.h" +#include "rtc_base/logging.h" + +namespace dcsctp { +namespace { +std::unique_ptr<ReassemblyStreams> CreateStreams( + absl::string_view log_prefix, + ReassemblyStreams::OnAssembledMessage on_assembled_message, + bool use_message_interleaving) { + if (use_message_interleaving) { + return std::make_unique<InterleavedReassemblyStreams>( + log_prefix, std::move(on_assembled_message)); + } + return std::make_unique<TraditionalReassemblyStreams>( + log_prefix, std::move(on_assembled_message)); +} +} // namespace + +ReassemblyQueue::ReassemblyQueue(absl::string_view log_prefix, + TSN peer_initial_tsn, + size_t max_size_bytes, + bool use_message_interleaving) + : log_prefix_(log_prefix), + max_size_bytes_(max_size_bytes), + watermark_bytes_(max_size_bytes * kHighWatermarkLimit), + last_assembled_tsn_watermark_( + tsn_unwrapper_.Unwrap(TSN(*peer_initial_tsn - 1))), + last_completed_reset_req_seq_nbr_(ReconfigRequestSN(0)), + streams_(CreateStreams( + log_prefix_, + [this](rtc::ArrayView<const UnwrappedTSN> tsns, + DcSctpMessage message) { + AddReassembledMessage(tsns, std::move(message)); + }, + use_message_interleaving)) {} + +void ReassemblyQueue::Add(TSN tsn, Data data) { + RTC_DLOG(LS_VERBOSE) << log_prefix_ << "added tsn=" << *tsn + << ", stream=" << *data.stream_id << ":" << *data.mid + << ":" << *data.fsn << ", type=" + << (data.is_beginning && data.is_end ? "complete" + : data.is_beginning ? "first" + : data.is_end ? "last" + : "middle"); + + UnwrappedTSN unwrapped_tsn = tsn_unwrapper_.Unwrap(tsn); + + // If a stream reset has been received with a "sender's last assigned tsn" in + // the future, the socket is in "deferred reset processing" mode and must + // buffer chunks until it's exited. + if (deferred_reset_streams_.has_value() && + unwrapped_tsn > deferred_reset_streams_->sender_last_assigned_tsn && + deferred_reset_streams_->streams.contains(data.stream_id)) { + RTC_DLOG(LS_VERBOSE) + << log_prefix_ << "Deferring chunk with tsn=" << *tsn + << ", sid=" << *data.stream_id << " until tsn=" + << *deferred_reset_streams_->sender_last_assigned_tsn.Wrap(); + // https://tools.ietf.org/html/rfc6525#section-5.2.2 + // "In this mode, any data arriving with a TSN larger than the + // Sender's Last Assigned TSN for the affected stream(s) MUST be queued + // locally and held until the cumulative acknowledgment point reaches the + // Sender's Last Assigned TSN." + queued_bytes_ += data.size(); + deferred_reset_streams_->deferred_actions.push_back( + [this, tsn, data = std::move(data)]() mutable { + queued_bytes_ -= data.size(); + Add(tsn, std::move(data)); + }); + } else { + queued_bytes_ += streams_->Add(unwrapped_tsn, std::move(data)); + } + + // https://tools.ietf.org/html/rfc4960#section-6.9 + // "Note: If the data receiver runs out of buffer space while still + // waiting for more fragments to complete the reassembly of the message, it + // should dispatch part of its inbound message through a partial delivery + // API (see Section 10), freeing some of its receive buffer space so that + // the rest of the message may be received." + + // TODO(boivie): Support EOR flag and partial delivery? + RTC_DCHECK(IsConsistent()); +} + +void ReassemblyQueue::ResetStreamsAndLeaveDeferredReset( + rtc::ArrayView<const StreamID> stream_ids) { + RTC_DLOG(LS_VERBOSE) << log_prefix_ << "Resetting streams: [" + << StrJoin(stream_ids, ",", + [](rtc::StringBuilder& sb, StreamID sid) { + sb << *sid; + }) + << "]"; + + // https://tools.ietf.org/html/rfc6525#section-5.2.2 + // "... streams MUST be reset to 0 as the next expected SSN." + streams_->ResetStreams(stream_ids); + + if (deferred_reset_streams_.has_value()) { + RTC_DLOG(LS_VERBOSE) << log_prefix_ + << "Leaving deferred reset processing, feeding back " + << deferred_reset_streams_->deferred_actions.size() + << " actions"; + // https://tools.ietf.org/html/rfc6525#section-5.2.2 + // "Any queued TSNs (queued at step E2) MUST now be released and processed + // normally." + auto deferred_actions = + std::move(deferred_reset_streams_->deferred_actions); + deferred_reset_streams_ = absl::nullopt; + + for (auto& action : deferred_actions) { + action(); + } + } + + RTC_DCHECK(IsConsistent()); +} + +void ReassemblyQueue::EnterDeferredReset( + TSN sender_last_assigned_tsn, + rtc::ArrayView<const StreamID> streams) { + if (!deferred_reset_streams_.has_value()) { + RTC_DLOG(LS_VERBOSE) << log_prefix_ + << "Entering deferred reset; sender_last_assigned_tsn=" + << *sender_last_assigned_tsn; + deferred_reset_streams_ = absl::make_optional<DeferredResetStreams>( + tsn_unwrapper_.Unwrap(sender_last_assigned_tsn), + webrtc::flat_set<StreamID>(streams.begin(), streams.end())); + } + RTC_DCHECK(IsConsistent()); +} + +std::vector<DcSctpMessage> ReassemblyQueue::FlushMessages() { + std::vector<DcSctpMessage> ret; + reassembled_messages_.swap(ret); + return ret; +} + +void ReassemblyQueue::AddReassembledMessage( + rtc::ArrayView<const UnwrappedTSN> tsns, + DcSctpMessage message) { + RTC_DLOG(LS_VERBOSE) << log_prefix_ << "Assembled message from TSN=[" + << StrJoin(tsns, ",", + [](rtc::StringBuilder& sb, UnwrappedTSN tsn) { + sb << *tsn.Wrap(); + }) + << "], message; stream_id=" << *message.stream_id() + << ", ppid=" << *message.ppid() + << ", payload=" << message.payload().size() << " bytes"; + + for (const UnwrappedTSN tsn : tsns) { + if (tsn == last_assembled_tsn_watermark_.next_value()) { + // Update watermark, or insert into delivered_tsns_ + last_assembled_tsn_watermark_.Increment(); + } else { + delivered_tsns_.insert(tsn); + } + } + + // With new TSNs in delivered_tsns, gaps might be filled. + MaybeMoveLastAssembledWatermarkFurther(); + + reassembled_messages_.emplace_back(std::move(message)); +} + +void ReassemblyQueue::MaybeMoveLastAssembledWatermarkFurther() { + // `delivered_tsns_` contain TSNS when there is a gap between ranges of + // assembled TSNs. `last_assembled_tsn_watermark_` should not be adjacent to + // that list, because if so, it can be moved. + while (!delivered_tsns_.empty() && + *delivered_tsns_.begin() == + last_assembled_tsn_watermark_.next_value()) { + last_assembled_tsn_watermark_.Increment(); + delivered_tsns_.erase(delivered_tsns_.begin()); + } +} + +void ReassemblyQueue::HandleForwardTsn( + TSN new_cumulative_tsn, + rtc::ArrayView<const AnyForwardTsnChunk::SkippedStream> skipped_streams) { + UnwrappedTSN tsn = tsn_unwrapper_.Unwrap(new_cumulative_tsn); + + if (deferred_reset_streams_.has_value() && + tsn > deferred_reset_streams_->sender_last_assigned_tsn) { + RTC_DLOG(LS_VERBOSE) << log_prefix_ << "ForwardTSN to " << *tsn.Wrap() + << "- deferring."; + deferred_reset_streams_->deferred_actions.emplace_back( + [this, new_cumulative_tsn, + streams = std::vector<AnyForwardTsnChunk::SkippedStream>( + skipped_streams.begin(), skipped_streams.end())] { + HandleForwardTsn(new_cumulative_tsn, streams); + }); + RTC_DCHECK(IsConsistent()); + return; + } + + RTC_DLOG(LS_VERBOSE) << log_prefix_ << "ForwardTSN to " << *tsn.Wrap() + << " - performing."; + last_assembled_tsn_watermark_ = std::max(last_assembled_tsn_watermark_, tsn); + delivered_tsns_.erase(delivered_tsns_.begin(), + delivered_tsns_.upper_bound(tsn)); + MaybeMoveLastAssembledWatermarkFurther(); + queued_bytes_ -= streams_->HandleForwardTsn(tsn, skipped_streams); + RTC_DCHECK(IsConsistent()); +} + +bool ReassemblyQueue::IsConsistent() const { + // `delivered_tsns_` and `last_assembled_tsn_watermark_` mustn't overlap or be + // adjacent. + if (!delivered_tsns_.empty() && + last_assembled_tsn_watermark_.next_value() >= *delivered_tsns_.begin()) { + return false; + } + + // Allow queued_bytes_ to be larger than max_size_bytes, as it's not actively + // enforced in this class. This comparison will still trigger if queued_bytes_ + // became "negative". + return (queued_bytes_ >= 0 && queued_bytes_ <= 2 * max_size_bytes_); +} + +HandoverReadinessStatus ReassemblyQueue::GetHandoverReadiness() const { + HandoverReadinessStatus status = streams_->GetHandoverReadiness(); + if (!delivered_tsns_.empty()) { + status.Add(HandoverUnreadinessReason::kReassemblyQueueDeliveredTSNsGap); + } + if (deferred_reset_streams_.has_value()) { + status.Add(HandoverUnreadinessReason::kStreamResetDeferred); + } + return status; +} + +void ReassemblyQueue::AddHandoverState(DcSctpSocketHandoverState& state) { + state.rx.last_assembled_tsn = last_assembled_tsn_watermark_.Wrap().value(); + state.rx.last_completed_deferred_reset_req_sn = + last_completed_reset_req_seq_nbr_.value(); + streams_->AddHandoverState(state); +} + +void ReassemblyQueue::RestoreFromState(const DcSctpSocketHandoverState& state) { + // Validate that the component is in pristine state. + RTC_DCHECK(last_completed_reset_req_seq_nbr_ == ReconfigRequestSN(0)); + + last_assembled_tsn_watermark_ = + tsn_unwrapper_.Unwrap(TSN(state.rx.last_assembled_tsn)); + last_completed_reset_req_seq_nbr_ = + ReconfigRequestSN(state.rx.last_completed_deferred_reset_req_sn); + streams_->RestoreFromState(state); +} +} // namespace dcsctp |