diff --git a/net/dcsctp/tx/rr_send_queue.cc b/net/dcsctp/tx/rr_send_queue.cc index 9e45486f4a..b1812f0f8a 100644 --- a/net/dcsctp/tx/rr_send_queue.cc +++ b/net/dcsctp/tx/rr_send_queue.cc @@ -119,12 +119,11 @@ void RRSendQueue::ThresholdWatcher::SetLowThreshold(size_t low_threshold) { } void RRSendQueue::OutgoingStream::Add(DcSctpMessage message, - TimeMs expires_at, - const SendOptions& send_options) { + MessageAttributes attributes) { bool was_active = bytes_to_send_in_next_message() > 0; buffered_amount_.Increase(message.payload().size()); parent_.total_buffered_amount_.Increase(message.payload().size()); - items_.emplace_back(std::move(message), expires_at, send_options); + items_.emplace_back(std::move(message), std::move(attributes)); if (!was_active) { scheduler_stream_->MaybeMakeActive(); @@ -146,19 +145,18 @@ absl::optional RRSendQueue::OutgoingStream::Produce( // Allocate Message ID and SSN when the first fragment is sent. if (!item.message_id.has_value()) { // Oops, this entire message has already expired. Try the next one. - if (item.expires_at <= now) { - buffered_amount_.Decrease(item.remaining_size); - parent_.total_buffered_amount_.Decrease(item.remaining_size); + if (item.attributes.expires_at <= now) { + HandleMessageExpired(item); items_.pop_front(); continue; } MID& mid = - item.send_options.unordered ? next_unordered_mid_ : next_ordered_mid_; + item.attributes.unordered ? next_unordered_mid_ : next_ordered_mid_; item.message_id = mid; mid = MID(*mid + 1); } - if (!item.send_options.unordered && !item.ssn.has_value()) { + if (!item.attributes.unordered && !item.ssn.has_value()) { item.ssn = next_ssn_; next_ssn_ = SSN(*next_ssn_ + 1); } @@ -189,16 +187,11 @@ absl::optional RRSendQueue::OutgoingStream::Produce( SendQueue::DataToSend chunk(Data(stream_id, item.ssn.value_or(SSN(0)), item.message_id.value(), fsn, ppid, std::move(payload), is_beginning, is_end, - item.send_options.unordered)); - if (item.send_options.max_retransmissions.has_value() && - *item.send_options.max_retransmissions >= - std::numeric_limits::min() && - *item.send_options.max_retransmissions <= - std::numeric_limits::max()) { - chunk.max_retransmissions = - MaxRetransmits(*item.send_options.max_retransmissions); - } - chunk.expires_at = item.expires_at; + item.attributes.unordered)); + chunk.max_retransmissions = item.attributes.max_retransmissions; + chunk.expires_at = item.attributes.expires_at; + chunk.lifecycle_id = + is_end ? item.attributes.lifecycle_id : LifecycleId::NotSet(); if (is_end) { // The entire message has been sent, and its last data copied to `chunk`, @@ -224,15 +217,28 @@ absl::optional RRSendQueue::OutgoingStream::Produce( return absl::nullopt; } +void RRSendQueue::OutgoingStream::HandleMessageExpired( + OutgoingStream::Item& item) { + buffered_amount_.Decrease(item.remaining_size); + parent_.total_buffered_amount_.Decrease(item.remaining_size); + if (item.attributes.lifecycle_id.IsSet()) { + RTC_DLOG(LS_VERBOSE) << "Triggering OnLifecycleMessageExpired(" + << item.attributes.lifecycle_id.value() << ", false)"; + + parent_.callbacks_.OnLifecycleMessageExpired(item.attributes.lifecycle_id, + /*maybe_delivered=*/false); + parent_.callbacks_.OnLifecycleEnd(item.attributes.lifecycle_id); + } +} + bool RRSendQueue::OutgoingStream::Discard(IsUnordered unordered, MID message_id) { bool result = false; if (!items_.empty()) { Item& item = items_.front(); - if (item.send_options.unordered == unordered && - item.message_id.has_value() && *item.message_id == message_id) { - buffered_amount_.Decrease(item.remaining_size); - parent_.total_buffered_amount_.Decrease(item.remaining_size); + if (item.attributes.unordered == unordered && item.message_id.has_value() && + *item.message_id == message_id) { + HandleMessageExpired(item); items_.pop_front(); // Only partially sent messages are discarded, so if a message was @@ -277,8 +283,7 @@ void RRSendQueue::OutgoingStream::Pause() { // the fragments before actually resetting the stream. for (auto it = items_.begin(); it != items_.end();) { if (it->remaining_offset == 0) { - buffered_amount_.Decrease(it->remaining_size); - parent_.total_buffered_amount_.Decrease(it->remaining_size); + HandleMessageExpired(*it); it = items_.erase(it); } else { ++it; @@ -348,15 +353,23 @@ void RRSendQueue::Add(TimeMs now, RTC_DCHECK(!message.payload().empty()); // Any limited lifetime should start counting from now - when the message // has been added to the queue. - TimeMs expires_at = TimeMs::InfiniteFuture(); - if (send_options.lifetime.has_value()) { - // `expires_at` is the time when it expires. Which is slightly larger than - // the message's lifetime, as the message is alive during its entire - // lifetime (which may be zero). - expires_at = now + *send_options.lifetime + DurationMs(1); - } + + // `expires_at` is the time when it expires. Which is slightly larger than the + // message's lifetime, as the message is alive during its entire lifetime + // (which may be zero). + MessageAttributes attributes = { + .unordered = send_options.unordered, + .max_retransmissions = + send_options.max_retransmissions.has_value() + ? MaxRetransmits(send_options.max_retransmissions.value()) + : MaxRetransmits::NoLimit(), + .expires_at = send_options.lifetime.has_value() + ? now + *send_options.lifetime + DurationMs(1) + : TimeMs::InfiniteFuture(), + .lifecycle_id = send_options.lifecycle_id, + }; GetOrCreateStreamInfo(message.stream_id()) - .Add(std::move(message), expires_at, send_options); + .Add(std::move(message), std::move(attributes)); RTC_DCHECK(IsConsistent()); } diff --git a/net/dcsctp/tx/rr_send_queue.h b/net/dcsctp/tx/rr_send_queue.h index 9152f27bf7..e9b8cd2081 100644 --- a/net/dcsctp/tx/rr_send_queue.h +++ b/net/dcsctp/tx/rr_send_queue.h @@ -45,6 +45,12 @@ namespace dcsctp { // The send queue may trigger callbacks: // * `OnBufferedAmountLow`, `OnTotalBufferedAmountLow` // These will be triggered as defined in their documentation. +// * `OnLifecycleMessageExpired(/*maybe_delivered=*/false)`, `OnLifecycleEnd` +// These will be triggered when messages have been expired, abandoned or +// discarded from the send queue. If a message is fully produced, meaning +// that the last fragment has been produced, the responsibility to send +// lifecycle events is then transferred to the retransmission queue, which +// is the one asking to produce the message. class RRSendQueue : public SendQueue { public: RRSendQueue(absl::string_view log_prefix, @@ -96,6 +102,13 @@ class RRSendQueue : public SendQueue { void RestoreFromState(const DcSctpSocketHandoverState& state); private: + struct MessageAttributes { + IsUnordered unordered; + MaxRetransmits max_retransmissions; + TimeMs expires_at; + LifecycleId lifecycle_id; + }; + // Represents a value and a "low threshold" that when the value reaches or // goes under the "low threshold", will trigger `on_threshold_reached` // callback. @@ -139,9 +152,7 @@ class RRSendQueue : public SendQueue { StreamID stream_id() const { return scheduler_stream_->stream_id(); } // Enqueues a message to this stream. - void Add(DcSctpMessage message, - TimeMs expires_at, - const SendOptions& send_options); + void Add(DcSctpMessage message, MessageAttributes attributes); // Implementing `StreamScheduler::StreamProducer`. absl::optional Produce(TimeMs now, @@ -208,17 +219,13 @@ class RRSendQueue : public SendQueue { // An enqueued message and metadata. struct Item { - explicit Item(DcSctpMessage msg, - TimeMs expires_at, - const SendOptions& send_options) + explicit Item(DcSctpMessage msg, MessageAttributes attributes) : message(std::move(msg)), - expires_at(expires_at), - send_options(send_options), + attributes(std::move(attributes)), remaining_offset(0), remaining_size(message.payload().size()) {} DcSctpMessage message; - TimeMs expires_at; - SendOptions send_options; + MessageAttributes attributes; // The remaining payload (offset and size) to be sent, when it has been // fragmented. size_t remaining_offset; @@ -232,6 +239,7 @@ class RRSendQueue : public SendQueue { }; bool IsConsistent() const; + void HandleMessageExpired(OutgoingStream::Item& item); RRSendQueue& parent_; diff --git a/net/dcsctp/tx/rr_send_queue_test.cc b/net/dcsctp/tx/rr_send_queue_test.cc index 78b5ecd8db..95416b193a 100644 --- a/net/dcsctp/tx/rr_send_queue_test.cc +++ b/net/dcsctp/tx/rr_send_queue_test.cc @@ -812,5 +812,55 @@ TEST_F(RRSendQueueTest, WillSendMessagesByPrio) { EXPECT_FALSE(buf_.Produce(kNow, 1).has_value()); } +TEST_F(RRSendQueueTest, WillSendLifecycleExpireWhenExpiredInSendQueue) { + std::vector payload(kOneFragmentPacketSize); + buf_.Add(kNow, DcSctpMessage(StreamID(2), kPPID, payload), + SendOptions{.lifetime = DurationMs(1000), + .lifecycle_id = LifecycleId(1)}); + + EXPECT_CALL(callbacks_, OnLifecycleMessageExpired(LifecycleId(1), + /*maybe_delivered=*/false)); + EXPECT_CALL(callbacks_, OnLifecycleEnd(LifecycleId(1))); + EXPECT_FALSE(buf_.Produce(kNow + DurationMs(1001), kOneFragmentPacketSize) + .has_value()); +} + +TEST_F(RRSendQueueTest, WillSendLifecycleExpireWhenDiscardingDuringPause) { + std::vector payload(120); + + buf_.Add(kNow, DcSctpMessage(kStreamID, kPPID, payload), + SendOptions{.lifecycle_id = LifecycleId(1)}); + buf_.Add(kNow, DcSctpMessage(kStreamID, kPPID, payload), + SendOptions{.lifecycle_id = LifecycleId(2)}); + + absl::optional chunk_one = buf_.Produce(kNow, 50); + ASSERT_TRUE(chunk_one.has_value()); + EXPECT_EQ(chunk_one->data.stream_id, kStreamID); + EXPECT_EQ(buf_.total_buffered_amount(), 2 * payload.size() - 50); + + EXPECT_CALL(callbacks_, OnLifecycleMessageExpired(LifecycleId(2), + /*maybe_delivered=*/false)); + EXPECT_CALL(callbacks_, OnLifecycleEnd(LifecycleId(2))); + buf_.PrepareResetStream(StreamID(1)); + EXPECT_EQ(buf_.total_buffered_amount(), payload.size() - 50); +} + +TEST_F(RRSendQueueTest, WillSendLifecycleExpireWhenDiscardingExplicitly) { + std::vector payload(kOneFragmentPacketSize + 20); + + buf_.Add(kNow, DcSctpMessage(kStreamID, kPPID, payload), + SendOptions{.lifecycle_id = LifecycleId(1)}); + + absl::optional chunk_one = + buf_.Produce(kNow, kOneFragmentPacketSize); + ASSERT_TRUE(chunk_one.has_value()); + EXPECT_FALSE(chunk_one->data.is_end); + EXPECT_EQ(chunk_one->data.stream_id, kStreamID); + EXPECT_CALL(callbacks_, OnLifecycleMessageExpired(LifecycleId(1), + /*maybe_delivered=*/false)); + EXPECT_CALL(callbacks_, OnLifecycleEnd(LifecycleId(1))); + buf_.Discard(IsUnordered(false), chunk_one->data.stream_id, + chunk_one->data.message_id); +} } // namespace } // namespace dcsctp diff --git a/net/dcsctp/tx/send_queue.h b/net/dcsctp/tx/send_queue.h index a7e663530a..0b96e9041a 100644 --- a/net/dcsctp/tx/send_queue.h +++ b/net/dcsctp/tx/send_queue.h @@ -34,6 +34,10 @@ class SendQueue { // Partial reliability - RFC3758 MaxRetransmits max_retransmissions = MaxRetransmits::NoLimit(); TimeMs expires_at = TimeMs::InfiniteFuture(); + + // Lifecycle - set for the last fragment, and `LifecycleId::NotSet()` for + // all other fragments. + LifecycleId lifecycle_id = LifecycleId::NotSet(); }; virtual ~SendQueue() = default;