/* * 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. */ #ifndef NET_DCSCTP_RX_REASSEMBLY_QUEUE_H_ #define NET_DCSCTP_RX_REASSEMBLY_QUEUE_H_ #include #include #include #include #include #include #include #include "absl/strings/string_view.h" #include "api/array_view.h" #include "net/dcsctp/common/internal_types.h" #include "net/dcsctp/common/sequence_numbers.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_handover_state.h" #include "net/dcsctp/public/dcsctp_message.h" #include "net/dcsctp/rx/reassembly_streams.h" namespace dcsctp { // Contains the received DATA chunks that haven't yet been reassembled, and // reassembles chunks when possible. // // The actual assembly is handled by an implementation of the // `ReassemblyStreams` interface. // // Except for reassembling fragmented messages, this class will also handle two // less common operations; To handle the receiver-side of partial reliability // (limited number of retransmissions or limited message lifetime) as well as // stream resetting, which is used when a sender wishes to close a data channel. // // Partial reliability is handled when a FORWARD-TSN or I-FORWARD-TSN chunk is // received, and it will simply delete any chunks matching the parameters in // that chunk. This is mainly implemented in ReassemblyStreams. // // Resetting streams is handled when a RECONFIG chunks is received, with an // "Outgoing SSN Reset Request" parameter. That parameter will contain a list of // streams to reset, and a `sender_last_assigned_tsn`. If this TSN is not yet // seen, the stream cannot be directly reset, and this class will respond that // the reset is "deferred". But if this TSN provided is known, the stream can be // immediately be reset. // // The ReassemblyQueue has a maximum size, as it would otherwise be an DoS // attack vector where a peer could consume all memory of the other peer by // sending a lot of ordered chunks, but carefully withholding an early one. It // also has a watermark limit, which the caller can query is the number of bytes // is above that limit. This is used by the caller to be selective in what to // add to the reassembly queue, so that it's not exhausted. The caller is // expected to call `is_full` prior to adding data to the queue and to act // accordingly if the queue is full. class ReassemblyQueue { public: // When the queue is filled over this fraction (of its maximum size), the // socket should restrict incoming data to avoid filling up the queue. static constexpr float kHighWatermarkLimit = 0.9; ReassemblyQueue(absl::string_view log_prefix, TSN peer_initial_tsn, size_t max_size_bytes, const DcSctpSocketHandoverState* handover_state = nullptr); // Adds a data chunk to the queue, with a `tsn` and other parameters in // `data`. void Add(TSN tsn, Data data); // Indicates if the reassembly queue has any reassembled messages that can be // retrieved by calling `FlushMessages`. bool HasMessages() const { return !reassembled_messages_.empty(); } // Returns any reassembled messages. std::vector FlushMessages(); // Handle a ForwardTSN chunk, when the sender has indicated that the received // (this class) should forget about some chunks. This is used to implement // partial reliability. void Handle(const AnyForwardTsnChunk& forward_tsn); // Given the reset stream request and the current cum_tsn_ack, might either // reset the streams directly (returns kSuccessPerformed), or at a later time, // by entering the "deferred reset processing" mode (returns kInProgress). ReconfigurationResponseParameter::Result ResetStreams( const OutgoingSSNResetRequestParameter& req, TSN cum_tsn_ack); // Given the current (updated) cum_tsn_ack, might leave "defererred reset // processing" mode and reset streams. Returns true if so. bool MaybeResetStreamsDeferred(TSN cum_ack_tsn); // The number of payload bytes that have been queued. Note that the actual // memory usage is higher due to additional overhead of tracking received // data. size_t queued_bytes() const { return queued_bytes_; } // The remaining bytes until the queue has reached the watermark limit. size_t remaining_bytes() const { return watermark_bytes_ - queued_bytes_; } // Indicates if the queue is full. Data should not be added to the queue when // it's full. bool is_full() const { return queued_bytes_ >= max_size_bytes_; } // Indicates if the queue is above the watermark limit, which is a certain // percentage of its size. bool is_above_watermark() const { return queued_bytes_ >= watermark_bytes_; } // Returns the watermark limit, in bytes. size_t watermark_bytes() const { return watermark_bytes_; } HandoverReadinessStatus GetHandoverReadiness() const; void AddHandoverState(DcSctpSocketHandoverState& state); private: bool IsConsistent() const; void AddReassembledMessage(rtc::ArrayView tsns, DcSctpMessage message); void MaybeMoveLastAssembledWatermarkFurther(); struct DeferredResetStreams { explicit DeferredResetStreams(OutgoingSSNResetRequestParameter req) : req(std::move(req)) {} OutgoingSSNResetRequestParameter req; std::vector> deferred_chunks; }; const std::string log_prefix_; const size_t max_size_bytes_; const size_t watermark_bytes_; UnwrappedTSN::Unwrapper tsn_unwrapper_; // Whenever a message has been assembled, either increase // `last_assembled_tsn_watermark_` or - if there are gaps - add the message's // TSNs into delivered_tsns_ so that messages are not re-delivered on // duplicate chunks. UnwrappedTSN last_assembled_tsn_watermark_; std::set delivered_tsns_; // Messages that have been reassembled, and will be returned by // `FlushMessages`. std::vector reassembled_messages_; // If present, "deferred reset processing" mode is active. absl::optional deferred_reset_streams_; // Contains the last request sequence number of the // OutgoingSSNResetRequestParameter that was performed. ReconfigRequestSN last_completed_reset_req_seq_nbr_; // The number of "payload bytes" that are in this queue, in total. size_t queued_bytes_ = 0; // The actual implementation of ReassemblyStreams. std::unique_ptr streams_; }; } // namespace dcsctp #endif // NET_DCSCTP_RX_REASSEMBLY_QUEUE_H_