1 files changed, 177 insertions, 0 deletions
diff --git a/third_party/libwebrtc/net/dcsctp/rx/reassembly_queue.h b/third_party/libwebrtc/net/dcsctp/rx/reassembly_queue.h
new file mode 100644
index 0000000000..761ec3556c
--- /dev/null
+++ b/third_party/libwebrtc/net/dcsctp/rx/reassembly_queue.h
@@ -0,0 +1,177 @@
+/*
+ *  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 <stddef.h>
+
+#include <cstdint>
+#include <memory>
+#include <set>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "absl/functional/any_invocable.h"
+#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"
+#include "rtc_base/containers/flat_set.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,
+                  bool use_message_interleaving = false);
+
+  // 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<DcSctpMessage> 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 HandleForwardTsn(
+      TSN new_cumulative_tsn,
+      rtc::ArrayView<const AnyForwardTsnChunk::SkippedStream> skipped_streams);
+
+  // Resets the provided streams and leaves deferred reset processing, if
+  // enabled.
+  void ResetStreamsAndLeaveDeferredReset(
+      rtc::ArrayView<const StreamID> stream_ids);
+
+  // Enters deferred reset processing.
+  void EnterDeferredReset(TSN sender_last_assigned_tsn,
+                          rtc::ArrayView<const StreamID> streams);
+
+  // 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);
+  void RestoreFromState(const DcSctpSocketHandoverState& state);
+
+ private:
+  struct DeferredResetStreams {
+    DeferredResetStreams(UnwrappedTSN sender_last_assigned_tsn,
+                         webrtc::flat_set<StreamID> streams)
+        : sender_last_assigned_tsn(sender_last_assigned_tsn),
+          streams(std::move(streams)) {}
+
+    UnwrappedTSN sender_last_assigned_tsn;
+    webrtc::flat_set<StreamID> streams;
+    std::vector<absl::AnyInvocable<void(void)>> deferred_actions;
+  };
+
+  bool IsConsistent() const;
+  void AddReassembledMessage(rtc::ArrayView<const UnwrappedTSN> tsns,
+                             DcSctpMessage message);
+  void MaybeMoveLastAssembledWatermarkFurther();
+
+  const absl::string_view 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<UnwrappedTSN> delivered_tsns_;
+  // Messages that have been reassembled, and will be returned by
+  // `FlushMessages`.
+  std::vector<DcSctpMessage> reassembled_messages_;
+
+  // If present, "deferred reset processing" mode is active.
+  absl::optional<DeferredResetStreams> 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<ReassemblyStreams> streams_;
+};
+}  // namespace dcsctp
+
+#endif  // NET_DCSCTP_RX_REASSEMBLY_QUEUE_H_