Split delivery_timeout into subgroup and object delivery timeouts in MoQT. Part of draft-18 update. PiperOrigin-RevId: 994669813
diff --git a/quiche/quic/moqt/moqt_framer.cc b/quiche/quic/moqt/moqt_framer.cc index 663835a..d52ac07 100644 --- a/quiche/quic/moqt/moqt_framer.cc +++ b/quiche/quic/moqt/moqt_framer.cc
@@ -305,22 +305,31 @@ return out; } +uint64_t TimeDeltaToMilliseconds(const quic::QuicTimeDelta& time_delta) { + if (time_delta == quic::QuicTimeDelta::Infinite()) { + return 0ULL; + } + return std::max(time_delta.ToMilliseconds(), int64_t{1}); +} + KeyValuePairList MessageParameters::ToKeyValuePairList() const { KeyValuePairList list; - if (delivery_timeout.has_value()) { - // Value cannot be zero. - int64_t milliseconds = - std::max(delivery_timeout->ToMilliseconds(), int64_t{1}); - list.insert(static_cast<uint64_t>(MessageParameter::kDeliveryTimeout), - static_cast<uint64_t>(milliseconds)); + if (object_delivery_timeout.has_value()) { + list.insert(static_cast<uint64_t>(MessageParameter::kObjectDeliveryTimeout), + TimeDeltaToMilliseconds(*object_delivery_timeout)); } for (const AuthToken& token : authorization_tokens) { list.insert(static_cast<uint64_t>(MessageParameter::kAuthorizationToken), SerializeAuthToken(token).AsStringView()); } + if (subgroup_delivery_timeout.has_value()) { + list.insert( + static_cast<uint64_t>(MessageParameter::kSubgroupDeliveryTimeout), + TimeDeltaToMilliseconds(*subgroup_delivery_timeout)); + } if (expires.has_value()) { list.insert(static_cast<uint64_t>(MessageParameter::kExpires), - static_cast<uint64_t>(expires->ToMilliseconds())); + TimeDeltaToMilliseconds(*expires)); } if (largest_object.has_value()) { list.insert(static_cast<uint64_t>(MessageParameter::kLargestObject),
diff --git a/quiche/quic/moqt/moqt_framer_test.cc b/quiche/quic/moqt/moqt_framer_test.cc index 54e53d1..09f355c 100644 --- a/quiche/quic/moqt/moqt_framer_test.cc +++ b/quiche/quic/moqt/moqt_framer_test.cc
@@ -472,4 +472,44 @@ EXPECT_EQ(buffer.size(), 0); } +TEST_F(MoqtFramerSimpleTest, MessageParametersTimeDeltaSerialization) { + MessageParameters params; + params.object_delivery_timeout = quic::QuicTimeDelta::Infinite(); + params.subgroup_delivery_timeout = quic::QuicTimeDelta::FromMicroseconds(500); + params.expires = quic::QuicTimeDelta::Infinite(); + + KeyValuePairList list = params.ToKeyValuePairList(); + std::optional<uint64_t> object_timeout; + std::optional<uint64_t> subgroup_timeout; + std::optional<uint64_t> expires; + list.ForEach( + [&](uint64_t key, std::variant<uint64_t, absl::string_view> value) { + if (key == + static_cast<uint64_t>(MessageParameter::kObjectDeliveryTimeout)) { + object_timeout = std::get<uint64_t>(value); + } else if (key == static_cast<uint64_t>( + MessageParameter::kSubgroupDeliveryTimeout)) { + subgroup_timeout = std::get<uint64_t>(value); + } else if (key == static_cast<uint64_t>(MessageParameter::kExpires)) { + expires = std::get<uint64_t>(value); + } + return true; + }); + EXPECT_EQ(object_timeout, 0ULL); + EXPECT_EQ(subgroup_timeout, 1ULL); + EXPECT_EQ(expires, 0ULL); + + params.expires = quic::QuicTimeDelta::FromMicroseconds(200); + list = params.ToKeyValuePairList(); + expires.reset(); + list.ForEach( + [&](uint64_t key, std::variant<uint64_t, absl::string_view> value) { + if (key == static_cast<uint64_t>(MessageParameter::kExpires)) { + expires = std::get<uint64_t>(value); + } + return true; + }); + EXPECT_EQ(expires, 1ULL); +} + } // namespace moqt::test
diff --git a/quiche/quic/moqt/moqt_integration_test.cc b/quiche/quic/moqt/moqt_integration_test.cc index f1b7918..d4f414b 100644 --- a/quiche/quic/moqt/moqt_integration_test.cc +++ b/quiche/quic/moqt/moqt_integration_test.cc
@@ -759,7 +759,7 @@ EXPECT_THAT(ack_deltas, ElementsAre(-123, 456)); } -TEST_F(MoqtIntegrationTest, DeliveryTimeout) { +TEST_F(MoqtIntegrationTest, SubgroupDeliveryTimeout) { EstablishSession(); // The loss is added after the handshake, to ensure that the connection has // enough congestion control window to work with for the further tests. @@ -785,7 +785,8 @@ }); MessageParameters parameters(MoqtFilterType::kLargestObject); // Set delivery timeout to ~ 1 RTT: any loss is fatal. - parameters.delivery_timeout = quic::QuicTimeDelta::FromMilliseconds(100); + parameters.subgroup_delivery_timeout = + quic::QuicTimeDelta::FromMilliseconds(100); client_->session()->Subscribe(full_track_name, &subscribe_visitor_, parameters); bool success = @@ -812,13 +813,14 @@ EXPECT_LT(bytes_received, 4000); } -TEST_F(MoqtIntegrationTest, AlternateDeliveryTimeout) { +// An object arrives, but the session doesn't grant stream credit until it's +// too late. +TEST_F(MoqtIntegrationTest, ObjectDeliveryTimeout) { EstablishSession(); FullTrackName full_track_name("foo", "bar"); MoqtKnownTrackPublisher publisher; server_->session()->set_publisher(&publisher); - server_->session()->UseAlternateDeliveryTimeout(); auto queue = std::make_shared<TestTrackPublisher>(full_track_name); auto track_publisher = std::make_shared<MockTrackPublisher>(full_track_name); publisher.Add(queue); @@ -835,8 +837,9 @@ stream_reset = true; }); MessageParameters parameters(MoqtFilterType::kLargestObject); - // Set delivery timeout to ~ 1 RTT: any loss is fatal. - parameters.delivery_timeout = quic::QuicTimeDelta::FromMilliseconds(100); + // Set delivery timeout to ~ 1 RTT. + parameters.object_delivery_timeout = + quic::QuicTimeDelta::FromMilliseconds(100); ON_CALL(*track_publisher, expiration) .WillByDefault(Return(quic::QuicTimeDelta::Zero())); client_->session()->Subscribe(full_track_name, &subscribe_visitor_, @@ -846,7 +849,7 @@ EXPECT_TRUE(success); success = false; - std::string data(1000, '\0'); + std::string data(65535, '\0'); size_t bytes_received = 0; EXPECT_CALL(subscribe_visitor_, OnObjectFragment) .WillRepeatedly( @@ -854,12 +857,21 @@ absl::string_view object, uint64_t offset) { bytes_received += object.size(); }); quic::QuicTime now = test_harness_.simulator().GetClock()->Now(); - queue->AddObject(Location{0, 0}, 0, data, false, now); - queue->AddObject(Location{1, 0}, 0, data, false, now); + // Send a small object that allows the subscriber to register the stream with + // this subgroup. + queue->AddObject(Location{0, 0}, 0, absl::string_view(data.data(), 100), + false, now); + // Send a huge object that will take more than 1 RTT to deliver. This clogs up + // the QUIC buffer and forces the next object to remain at the MOQT layer. + queue->AddObject(Location{0, 1}, 0, data, false, now); + // The third object will have to wait for an RTT, causing a timeout and reset. + queue->AddObject(Location{0, 2}, 0, absl::string_view(data.data(), 100), true, + now); success = test_harness_.RunUntilWithDefaultTimeout([&]() { return stream_reset; }); EXPECT_TRUE(success); - EXPECT_EQ(bytes_received, 2000); + // Partial object delivery is disabled, so only the first object is received. + EXPECT_EQ(bytes_received, 100); // On teardown, streams are destroyed in arbitrary order. If the uni stream // is destroyed before the bidi stream, there will be a second stream reset // notification to the visitor. If not, there won't be.
diff --git a/quiche/quic/moqt/moqt_key_value_pair.cc b/quiche/quic/moqt/moqt_key_value_pair.cc index 1ff4b4a..7422ee6 100644 --- a/quiche/quic/moqt/moqt_key_value_pair.cc +++ b/quiche/quic/moqt/moqt_key_value_pair.cc
@@ -74,12 +74,15 @@ } void MessageParameters::Update(const MessageParameters& other) { - if (other.delivery_timeout.has_value()) { - delivery_timeout = other.delivery_timeout; + if (other.object_delivery_timeout.has_value()) { + object_delivery_timeout = other.object_delivery_timeout; } if (!other.authorization_tokens.empty()) { authorization_tokens = other.authorization_tokens; } + if (other.subgroup_delivery_timeout.has_value()) { + subgroup_delivery_timeout = other.subgroup_delivery_timeout; + } if (other.expires.has_value()) { expires = other.expires; } @@ -105,22 +108,30 @@ } TrackProperties::TrackProperties( - std::optional<quic::QuicTimeDelta> delivery_timeout, + std::optional<quic::QuicTimeDelta> object_delivery_timeout, std::optional<quic::QuicTimeDelta> max_cache_duration, + std::optional<quic::QuicTimeDelta> subgroup_delivery_timeout, std::optional<MoqtPriority> publisher_priority, std::optional<MoqtDeliveryOrder> group_order, std::optional<bool> dynamic_groups, std::optional<absl::string_view> immutable_properties) { - if (delivery_timeout.has_value() && - *delivery_timeout != kDefaultDeliveryTimeout) { - insert(static_cast<uint64_t>(PropertyType::kDeliveryTimeout), - static_cast<uint64_t>(delivery_timeout->ToMilliseconds())); + if (object_delivery_timeout.has_value() && + *object_delivery_timeout != kDefaultDeliveryTimeout && + *object_delivery_timeout != quic::QuicTimeDelta::Zero()) { + insert(static_cast<uint64_t>(PropertyType::kObjectDeliveryTimeout), + static_cast<uint64_t>(object_delivery_timeout->ToMilliseconds())); } if (max_cache_duration.has_value() && *max_cache_duration != kDefaultMaxCacheDuration) { insert(static_cast<uint64_t>(PropertyType::kMaxCacheDuration), static_cast<uint64_t>(max_cache_duration->ToMilliseconds())); } + if (subgroup_delivery_timeout.has_value() && + *subgroup_delivery_timeout != kDefaultDeliveryTimeout && + *subgroup_delivery_timeout != quic::QuicTimeDelta::Zero()) { + insert(static_cast<uint64_t>(PropertyType::kSubgroupDeliveryTimeout), + static_cast<uint64_t>(subgroup_delivery_timeout->ToMilliseconds())); + } if (immutable_properties.has_value() && !immutable_properties->empty()) { insert(static_cast<uint64_t>(PropertyType::kImmutableProperties), *immutable_properties); @@ -140,19 +151,26 @@ } } -quic::QuicTimeDelta TrackProperties::delivery_timeout() const { +quic::QuicTimeDelta TrackProperties::object_delivery_timeout() const { std::optional<uint64_t> value = - GetValueIfExactlyOne(PropertyType::kDeliveryTimeout); - return value.has_value() ? quic::QuicTimeDelta::FromMilliseconds(*value) - : kDefaultDeliveryTimeout; + GetValueIfExactlyOne(PropertyType::kObjectDeliveryTimeout); + return (value.has_value() && *value > 0) + ? quic::QuicTimeDelta::FromMilliseconds(*value) + : kDefaultDeliveryTimeout; } - quic::QuicTimeDelta TrackProperties::max_cache_duration() const { std::optional<uint64_t> value = GetValueIfExactlyOne(PropertyType::kMaxCacheDuration); return value.has_value() ? quic::QuicTimeDelta::FromMilliseconds(*value) : kDefaultMaxCacheDuration; } +quic::QuicTimeDelta TrackProperties::subgroup_delivery_timeout() const { + std::optional<uint64_t> value = + GetValueIfExactlyOne(PropertyType::kSubgroupDeliveryTimeout); + return (value.has_value() && *value > 0) + ? quic::QuicTimeDelta::FromMilliseconds(*value) + : kDefaultDeliveryTimeout; +} absl::string_view TrackProperties::immutable_properties() const { ValueVector values = Get(static_cast<uint64_t>(PropertyType::kImmutableProperties)); @@ -186,10 +204,12 @@ bool TrackProperties::Validate() const { // TODO(martinduke): If immutable properties include an immutable properties // property, the track is malformed. - return (ValidateInner(PropertyType::kDeliveryTimeout, std::nullopt, + return (ValidateInner(PropertyType::kObjectDeliveryTimeout, std::nullopt, std::nullopt) && ValidateInner(PropertyType::kMaxCacheDuration, std::nullopt, std::nullopt) && + ValidateInner(PropertyType::kSubgroupDeliveryTimeout, std::nullopt, + std::nullopt) && ValidateInner(PropertyType::kDefaultPublisherPriority, std::nullopt, kMaxPriority) && ValidateInner(PropertyType::kDefaultPublisherGroupOrder,
diff --git a/quiche/quic/moqt/moqt_key_value_pair.h b/quiche/quic/moqt/moqt_key_value_pair.h index 5f158a3..10d7631 100644 --- a/quiche/quic/moqt/moqt_key_value_pair.h +++ b/quiche/quic/moqt/moqt_key_value_pair.h
@@ -191,8 +191,9 @@ }; enum class MessageParameter : uint64_t { - kDeliveryTimeout = 0x02, + kObjectDeliveryTimeout = 0x02, kAuthorizationToken = 0x03, + kSubgroupDeliveryTimeout = 0x06, kExpires = 0x08, kLargestObject = 0x09, kForward = 0x10, @@ -223,7 +224,8 @@ // with it. Otherwise, leave unchanged. void Update(const MessageParameters& other); - std::optional<quic::QuicTimeDelta> delivery_timeout; + std::optional<quic::QuicTimeDelta> subgroup_delivery_timeout; + std::optional<quic::QuicTimeDelta> object_delivery_timeout; std::vector<AuthToken> authorization_tokens; std::optional<quic::QuicTimeDelta> expires; std::optional<Location> largest_object; @@ -253,8 +255,9 @@ }; enum class PropertyType : uint64_t { - kDeliveryTimeout = 0x02, + kObjectDeliveryTimeout = 0x02, kMaxCacheDuration = 0x04, + kSubgroupDeliveryTimeout = 0x06, kImmutableProperties = 0x0b, kDefaultPublisherPriority = 0x0e, kDefaultPublisherGroupOrder = 0x22, @@ -275,8 +278,9 @@ TrackProperties() = default; TrackProperties(const TrackProperties&) = default; // Constructor for Original publishers to create their track properties. - TrackProperties(std::optional<quic::QuicTimeDelta> delivery_timeout, + TrackProperties(std::optional<quic::QuicTimeDelta> object_delivery_timeout, std::optional<quic::QuicTimeDelta> max_cache_duration, + std::optional<quic::QuicTimeDelta> subgroup_delivery_timeout, std::optional<MoqtPriority> publisher_priority, std::optional<MoqtDeliveryOrder> group_order, std::optional<bool> dynamic_groups, @@ -284,8 +288,9 @@ // If present and well-formed, returns the value of the property. Returns the // default value if missing or ill-formed. - quic::QuicTimeDelta delivery_timeout() const; + quic::QuicTimeDelta object_delivery_timeout() const; quic::QuicTimeDelta max_cache_duration() const; + quic::QuicTimeDelta subgroup_delivery_timeout() const; absl::string_view immutable_properties() const; MoqtPriority default_publisher_priority() const; MoqtDeliveryOrder default_publisher_group_order() const;
diff --git a/quiche/quic/moqt/moqt_key_value_pair_test.cc b/quiche/quic/moqt/moqt_key_value_pair_test.cc index 6400ef8..c04cb22 100644 --- a/quiche/quic/moqt/moqt_key_value_pair_test.cc +++ b/quiche/quic/moqt/moqt_key_value_pair_test.cc
@@ -136,13 +136,14 @@ TEST_F(MessageParametersTest, FromKeyValuePairList) { KeyValuePairList list; - list.insert(static_cast<uint64_t>(MessageParameter::kDeliveryTimeout), 1ULL); + list.insert(static_cast<uint64_t>(MessageParameter::kObjectDeliveryTimeout), + 1ULL); list.insert(static_cast<uint64_t>(MessageParameter::kForward), 0ULL); list.insert(static_cast<uint64_t>(MessageParameter::kOackWindowSize), 12345678ULL); MessageParameters parameters; QUICHE_EXPECT_OK(parameters.FromKeyValuePairList(list)); - EXPECT_EQ(parameters.delivery_timeout, + EXPECT_EQ(parameters.object_delivery_timeout, quic::QuicTimeDelta::FromMilliseconds(1)); EXPECT_FALSE(parameters.forward()); EXPECT_EQ(parameters.oack_window_size, @@ -152,10 +153,6 @@ TEST_F(MessageParametersTest, IllegalKeyValuePairs) { KeyValuePairList list; MessageParameters parameters; - list.insert(static_cast<uint64_t>(MessageParameter::kDeliveryTimeout), 0ULL); - EXPECT_THAT(parameters.FromKeyValuePairList(list), - StatusIs(absl::StatusCode::kInvalidArgument)); - list.clear(); list.insert(static_cast<uint64_t>(MessageParameter::kForward), 2ULL); EXPECT_THAT(parameters.FromKeyValuePairList(list), StatusIs(absl::StatusCode::kInvalidArgument)); @@ -181,7 +178,8 @@ TEST_F(MessageParametersTest, DuplicateParameters) { for (MessageParameter param : - {MessageParameter::kDeliveryTimeout, + {MessageParameter::kObjectDeliveryTimeout, + MessageParameter::kSubgroupDeliveryTimeout, // Auth token can be repeated. MessageParameter::kExpires, MessageParameter::kLargestObject, MessageParameter::kForward, MessageParameter::kSubscriberPriority, @@ -235,19 +233,20 @@ TEST_F(MessageParametersTest, Update) { MessageParameters p1; - p1.delivery_timeout = quic::QuicTimeDelta::FromMilliseconds(10); + p1.object_delivery_timeout = quic::QuicTimeDelta::FromMilliseconds(10); p1.expires = quic::QuicTimeDelta::FromMilliseconds(100); p1.set_forward(false); p1.subscriber_priority = 100; p1.new_group_request = 1; MessageParameters p2; - p2.delivery_timeout = quic::QuicTimeDelta::FromMilliseconds(20); + p2.object_delivery_timeout = quic::QuicTimeDelta::FromMilliseconds(20); p2.authorization_tokens.push_back( AuthToken(AuthTokenType::kOutOfBand, "token")); p2.set_forward(true); p2.group_order = MoqtDeliveryOrder::kDescending; p1.Update(p2); - EXPECT_EQ(p1.delivery_timeout, quic::QuicTimeDelta::FromMilliseconds(20)); + EXPECT_EQ(p1.object_delivery_timeout, + quic::QuicTimeDelta::FromMilliseconds(20)); EXPECT_EQ(p1.expires, quic::QuicTimeDelta::FromMilliseconds(100)); ASSERT_EQ(p1.authorization_tokens.size(), 1); EXPECT_EQ(p1.authorization_tokens[0], @@ -263,7 +262,7 @@ TEST_F(TrackPropertiesTest, DefaultConstructor) { TrackProperties properties; EXPECT_TRUE(properties.Validate()); - EXPECT_EQ(properties.delivery_timeout(), kDefaultDeliveryTimeout); + EXPECT_EQ(properties.object_delivery_timeout(), kDefaultDeliveryTimeout); EXPECT_EQ(properties.max_cache_duration(), kDefaultMaxCacheDuration); EXPECT_EQ(properties.default_publisher_priority(), kDefaultPublisherPriority); EXPECT_EQ(properties.default_publisher_group_order(), kDefaultGroupOrder); @@ -274,13 +273,16 @@ TEST_F(TrackPropertiesTest, AllProperties) { TrackProperties properties(quic::QuicTimeDelta::FromMilliseconds(1), quic::QuicTimeDelta::FromMilliseconds(2), + quic::QuicTimeDelta::FromMilliseconds(3), MoqtPriority(10), MoqtDeliveryOrder::kDescending, true, "properties"); EXPECT_TRUE(properties.Validate()); - EXPECT_EQ(properties.delivery_timeout(), + EXPECT_EQ(properties.object_delivery_timeout(), quic::QuicTimeDelta::FromMilliseconds(1)); EXPECT_EQ(properties.max_cache_duration(), quic::QuicTimeDelta::FromMilliseconds(2)); + EXPECT_EQ(properties.subgroup_delivery_timeout(), + quic::QuicTimeDelta::FromMilliseconds(3)); EXPECT_EQ(properties.default_publisher_priority(), MoqtPriority(10)); EXPECT_EQ(properties.default_publisher_group_order(), MoqtDeliveryOrder::kDescending); @@ -290,12 +292,13 @@ TEST_F(TrackPropertiesTest, ExplicitDefaults) { TrackProperties properties(kDefaultDeliveryTimeout, kDefaultMaxCacheDuration, - kDefaultPublisherPriority, kDefaultGroupOrder, - kDefaultDynamicGroups, ""); + kDefaultDeliveryTimeout, kDefaultPublisherPriority, + kDefaultGroupOrder, kDefaultDynamicGroups, ""); EXPECT_TRUE(properties.Validate()); EXPECT_EQ(properties.size(), 0); - EXPECT_EQ(properties.delivery_timeout(), kDefaultDeliveryTimeout); + EXPECT_EQ(properties.object_delivery_timeout(), kDefaultDeliveryTimeout); EXPECT_EQ(properties.max_cache_duration(), kDefaultMaxCacheDuration); + EXPECT_EQ(properties.subgroup_delivery_timeout(), kDefaultDeliveryTimeout); EXPECT_EQ(properties.default_publisher_priority(), kDefaultPublisherPriority); EXPECT_EQ(properties.default_publisher_group_order(), kDefaultGroupOrder); EXPECT_EQ(properties.dynamic_groups(), kDefaultDynamicGroups); @@ -309,9 +312,9 @@ properties.insert(0x42, 25ULL); EXPECT_TRUE(properties.Validate()); - properties.insert(static_cast<uint64_t>(PropertyType::kDeliveryTimeout), + properties.insert(static_cast<uint64_t>(PropertyType::kObjectDeliveryTimeout), 5ULL); - properties.insert(static_cast<uint64_t>(PropertyType::kDeliveryTimeout), + properties.insert(static_cast<uint64_t>(PropertyType::kObjectDeliveryTimeout), 6ULL); EXPECT_FALSE(properties.Validate()); @@ -324,6 +327,13 @@ properties.clear(); properties.insert( + static_cast<uint64_t>(PropertyType::kSubgroupDeliveryTimeout), 5ULL); + properties.insert( + static_cast<uint64_t>(PropertyType::kSubgroupDeliveryTimeout), 6ULL); + EXPECT_FALSE(properties.Validate()); + + properties.clear(); + properties.insert( static_cast<uint64_t>(PropertyType::kDefaultPublisherPriority), 256ULL); EXPECT_FALSE(properties.Validate()); properties.clear(); @@ -367,7 +377,7 @@ QUICHE_EXPECT_OK(properties.CheckForUnknownMandatoryProperty()); // Non-mandatory property types (< 0x4000). - properties.insert(static_cast<uint64_t>(PropertyType::kDeliveryTimeout), + properties.insert(static_cast<uint64_t>(PropertyType::kObjectDeliveryTimeout), 5ULL); properties.insert(0x3FFE, 1ULL); properties.insert(0x3FFF, "odd_optional");
diff --git a/quiche/quic/moqt/moqt_live_publisher.cc b/quiche/quic/moqt/moqt_live_publisher.cc index ee32d49..0b4820d 100644 --- a/quiche/quic/moqt/moqt_live_publisher.cc +++ b/quiche/quic/moqt/moqt_live_publisher.cc
@@ -17,7 +17,6 @@ #include "absl/container/flat_hash_set.h" #include "absl/status/status.h" #include "absl/strings/string_view.h" -#include "quiche/quic/core/quic_clock.h" #include "quiche/quic/moqt/moqt_bidi_stream.h" #include "quiche/quic/moqt/moqt_error.h" #include "quiche/quic/moqt/moqt_framer.h" @@ -62,6 +61,10 @@ if (monitoring_interface_ != nullptr) { monitoring_interface_->OnObjectAckSupportKnown(parameters.oack_window_size); } + if (is_publish) { + parameters_.group_order = + track_publisher_->properties().default_publisher_group_order(); + } QUIC_DLOG(INFO) << "Created subscription for " << track_publisher_->GetTrackName(); // TODO(martinduke): Handle NEW_GROUP_REQUEST @@ -147,6 +150,12 @@ } void LivePublisher::OnSubscribeAccepted() { + publisher_subgroup_delivery_timeout_ = + track_publisher_->properties().subgroup_delivery_timeout(); + publisher_object_delivery_timeout_ = + track_publisher_->properties().object_delivery_timeout(); + default_publisher_priority_ = + track_publisher_->properties().default_publisher_priority(); if (established_) { return; // It's a PUBLISH. } @@ -165,10 +174,6 @@ parameters_.group_order = subscribe_ok.properties.default_publisher_group_order(); } - // TODO(martinduke): Support sending DELIVERY_TIMEOUT parameter as the - // publisher. - default_publisher_priority_ = - subscribe_ok.properties.default_publisher_priority(); bidi_stream_->SendOrBufferMessageOrFatal( framer_.SerializeSubscribeOk(subscribe_ok)); // It is now safe to process buffered REQUEST_UPDATEs. @@ -216,7 +221,6 @@ monitoring_interface_->OnNewObjectEnqueued(location); } } - // TODO(vasilvv): This currently sends UINT64_MAX for datagram subgroups. // Maybe do something more satisfactory? SessionToPublisherInterface* session_info = visitor(); @@ -236,25 +240,6 @@ } stream_id = stream_map_.GetStreamFor(index); } - if (session_info->alternate_delivery_timeout() && - !delivery_timeout().IsInfinite() && largest_sent_.has_value() && - location.group >= largest_sent_->group) { - // Start the delivery timeout timer on all previous groups. - for (uint64_t group = first_active_group_; group < location.group; - ++group) { - for (webtransport::StreamId stream_to_update : - stream_map_.GetStreamsForGroup(group)) { - webtransport::Stream* raw_stream = GetStreamById(stream_to_update); - if (raw_stream == nullptr) { - continue; - } - OutgoingSubgroupStream* stream = - absl::down_cast<OutgoingSubgroupStream*>(raw_stream->visitor()); - stream->CreateAndSetAlarm(session_info->clock()->ApproximateNow() + - delivery_timeout()); - } - } - } QUICHE_DCHECK_GE(location.group, first_active_group_); if (!subgroup.has_value()) { SendDatagram(location); @@ -289,6 +274,10 @@ default_publisher_priority())}), MoqtTrackPriority{subscriber_priority(), publisher_priority}); } + // TODO(martinduke): We are not properly enforcing SUBGROUP_DELIVERY_TIMEOUT + // for pending streams. When LivePublisher receives OnNewFinAvailable(), it + // should check pending streams (which ought to be accessible by index, not + // just rank) and start a timer on them. pending_streams_.emplace(rank, parameters); } } @@ -349,7 +338,8 @@ } std::vector<webtransport::StreamId> streams = stream_map_.GetStreamsForGroup(group_id); - if (delivery_timeout().IsInfinite() && largest_sent_.has_value() && + if (subgroup_delivery_timeout().IsInfinite() && + object_delivery_timeout().IsInfinite() && largest_sent_.has_value() && largest_sent_->group <= group_id) { PublishIsDone(PublishDoneCode::kTooFarBehind, ""); // No class access below this line!
diff --git a/quiche/quic/moqt/moqt_live_publisher.h b/quiche/quic/moqt/moqt_live_publisher.h index c2a5c95..757cd9a 100644 --- a/quiche/quic/moqt/moqt_live_publisher.h +++ b/quiche/quic/moqt/moqt_live_publisher.h
@@ -75,7 +75,6 @@ class QUICHE_EXPORT SessionToPublisherInterface { public: virtual ~SessionToPublisherInterface() = default; - virtual bool alternate_delivery_timeout() const = 0; // If |old_priority| is nullopt, the subscription does not have any pending // streams. If it has a value, |old_priority| is the old value to be // replaced by |new_priority|. @@ -142,22 +141,21 @@ (parameters_.subscription_filter->WindowKnown() && parameters_.subscription_filter->InWindow(location))); }; - bool alternate_delivery_timeout() override { - if (visitor() == nullptr) { - return false; - } - return visitor()->alternate_delivery_timeout(); - } const quic::QuicClock* clock() override { if (visitor() == nullptr) { return nullptr; } return visitor()->clock(); } - quic::QuicTimeDelta delivery_timeout() override { + quic::QuicTimeDelta object_delivery_timeout() override { return std::min( - parameters_.delivery_timeout.value_or(kDefaultDeliveryTimeout), - publisher_delivery_timeout_.value_or(kDefaultDeliveryTimeout)); + parameters_.object_delivery_timeout.value_or(kDefaultDeliveryTimeout), + publisher_object_delivery_timeout_); + } + quic::QuicTimeDelta subgroup_delivery_timeout() override { + return std::min( + parameters_.subgroup_delivery_timeout.value_or(kDefaultDeliveryTimeout), + publisher_subgroup_delivery_timeout_); } quic::QuicAlarmFactory* alarm_factory() override { if (visitor() == nullptr) { @@ -168,9 +166,6 @@ void OnObjectSent(Location sequence) override; void OnStreamTimeout(DataStreamIndex index) override { reset_subgroups_.insert(index); - if (visitor()->alternate_delivery_timeout()) { - first_active_group_ = std::max(first_active_group_, index.group + 1); - } } // OnSubgroupAbandoned() is declared above with MoqtObjectListener. void OnDataStreamDestroyed(DataStreamIndex) override; @@ -270,7 +265,10 @@ // group_order and largest_object may be updated by SUBSCRIBE_OK because // have no effect in a future REQUEST_UPDATE message. MessageParameters parameters_; - std::optional<quic::QuicTimeDelta> publisher_delivery_timeout_; + quic::QuicTimeDelta publisher_object_delivery_timeout_ = + kDefaultDeliveryTimeout; + quic::QuicTimeDelta publisher_subgroup_delivery_timeout_ = + kDefaultDeliveryTimeout; std::optional<MoqtPriority> default_publisher_priority_; uint64_t streams_opened_ = 0;
diff --git a/quiche/quic/moqt/moqt_live_publisher_test.cc b/quiche/quic/moqt/moqt_live_publisher_test.cc index 299a191..67883cf 100644 --- a/quiche/quic/moqt/moqt_live_publisher_test.cc +++ b/quiche/quic/moqt/moqt_live_publisher_test.cc
@@ -33,6 +33,7 @@ #include "quiche/quic/moqt/moqt_session_interface.h" #include "quiche/quic/moqt/moqt_trace_recorder.h" #include "quiche/quic/moqt/moqt_types.h" +#include "quiche/quic/moqt/moqt_uni_stream.h" #include "quiche/quic/moqt/test_tools/mock_moqt_session.h" #include "quiche/quic/moqt/test_tools/moqt_framer_utils.h" #include "quiche/quic/moqt/test_tools/moqt_mock_visitor.h" @@ -115,7 +116,8 @@ trace_recorder_(nullptr) { bidi_stream_.BindStream(&mock_bidi_stream_); parameters_.set_forward(true); - parameters_.delivery_timeout = quic::QuicTimeDelta::FromSeconds(1); + parameters_.object_delivery_timeout = quic::QuicTimeDelta::FromSeconds(1); + parameters_.subgroup_delivery_timeout = quic::QuicTimeDelta::FromSeconds(2); parameters_.group_order = MoqtDeliveryOrder::kAscending; EXPECT_CALL(monitoring_interface_, OnObjectAckSupportKnown) .Times(AtLeast(0)); @@ -126,7 +128,6 @@ framer_, track_publisher_, &bidi_stream_, kRequestId, kTrackAlias, parameters_, visitor_.weak_ptr_factory_.Create(), /*is_publish=*/false); - ON_CALL(visitor_, alternate_delivery_timeout).WillByDefault(Return(false)); ON_CALL(webtrans_, GetStreamById(kStreamId)) .WillByDefault(Return(&mock_uni_stream_)); ON_CALL(visitor_, alarm_factory).WillByDefault(Return(&alarm_factory_)); @@ -256,7 +257,7 @@ TEST_F(LivePublisherTest, OnSubscribeAcceptedWithFilter) { publisher_->parameters().subscription_filter = SubscriptionFilter(MoqtFilterType::kLargestObject); - const TrackProperties properties(std::nullopt, std::nullopt, + const TrackProperties properties(std::nullopt, std::nullopt, std::nullopt, /*default_publisher_priority=*/64, std::nullopt, std::nullopt, std::nullopt); EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true)); @@ -295,7 +296,7 @@ TEST_F(LivePublisherTest, OnSubscribeAcceptedWithQueuedUpdates) { MessageParameters update1; - update1.delivery_timeout = quic::QuicTimeDelta::FromSeconds(5); + update1.object_delivery_timeout = quic::QuicTimeDelta::FromSeconds(5); update1.set_forward(false); bidi_stream_.QueueIncomingUpdate(update1); @@ -328,7 +329,7 @@ publisher_->OnSubscribeAccepted(); EXPECT_TRUE(publisher_->established()); - EXPECT_EQ(publisher_->parameters().delivery_timeout, + EXPECT_EQ(publisher_->parameters().object_delivery_timeout, quic::QuicTimeDelta::FromSeconds(5)); EXPECT_TRUE(publisher_->parameters().forward()); EXPECT_EQ(publisher_->parameters().largest_object, Location(2, 5)); @@ -348,13 +349,13 @@ TEST_F(LivePublisherTest, Update) { MessageParameters new_params; - new_params.delivery_timeout = quic::QuicTimeDelta::FromSeconds(5); + new_params.object_delivery_timeout = quic::QuicTimeDelta::FromSeconds(5); EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true)); EXPECT_CALL(mock_bidi_stream_, Writev(SerializedControlMessage(MoqtRequestOk()), _)) .WillOnce(Return(absl::OkStatus())); publisher_->Update(new_params, false); - EXPECT_EQ(publisher_->parameters().delivery_timeout, + EXPECT_EQ(publisher_->parameters().object_delivery_timeout, quic::QuicTimeDelta::FromSeconds(5)); // Changing forward preference updates can_have_joining_fetch_ @@ -426,7 +427,7 @@ TEST_F(LivePublisherTest, UpdateRejectedByPublisher) { MessageParameters new_params; - new_params.delivery_timeout = quic::QuicTimeDelta::FromSeconds(5); + new_params.object_delivery_timeout = quic::QuicTimeDelta::FromSeconds(5); EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true)); // When from_request_ok is false, sends REQUEST_ERROR. @@ -436,7 +437,7 @@ Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _)) .WillOnce(Return(absl::OkStatus())); publisher_->Update(new_params, /*from_request_ok=*/false); - EXPECT_EQ(publisher_->parameters().delivery_timeout, + EXPECT_EQ(publisher_->parameters().object_delivery_timeout, quic::QuicTimeDelta::FromSeconds(1)); // When from_request_ok is true, resets the bidi stream. @@ -602,7 +603,8 @@ TEST_F(LivePublisherTest, OnGroupAbandonedTooFarBehind) { // Set the delivery timeout to infinite so that TooFarBehind is possible. - parameters_.delivery_timeout = quic::QuicTimeDelta::Infinite(); + parameters_.object_delivery_timeout = quic::QuicTimeDelta::Infinite(); + parameters_.subgroup_delivery_timeout = quic::QuicTimeDelta::Infinite(); publisher_->Update(parameters_, true); CreateStream(Location(5, 0), 0, 128); struct MoqtPublishDone expected_publish_done = { @@ -635,14 +637,13 @@ publisher_->OnCanCreateNewUniStream(); } -TEST_F(LivePublisherTest, AlternateDeliveryTimeoutSetAlarm) { - ON_CALL(visitor_, alternate_delivery_timeout).WillByDefault(Return(true)); +TEST_F(LivePublisherTest, SubgroupDeliveryTimeoutSetAlarm) { // Create a stream for group 1. CreateStream(Location(1, 0), 0, 128); // Create a pending stream for group 2, which should start the timer but does // less work than an active stream. EXPECT_CALL(visitor_, alarm_factory).WillOnce(Return(&alarm_factory_)); - CreatePendingStream(Location(2, 0), 0, 128); + publisher_->OnNewFinAvailable(Location(1, 7), 0); } TEST_F(LivePublisherTest, OnTrackPublisherGone) { @@ -808,38 +809,19 @@ Location(1, 0)); } -TEST_F(LivePublisherTest, AlternateDeliveryTimeout) { - EXPECT_CALL(visitor_, alternate_delivery_timeout) - .WillRepeatedly(Return(true)); +TEST_F(LivePublisherTest, SubgroupDeliveryTimeout) { CreateStream(Location(0, 0), 0, 128); - // Save the visitor before it's overwritten. - std::unique_ptr<webtransport::StreamVisitor> uni_stream = - std::move(uni_stream_); - CreateStream(Location(0, 1), 1, 200); - std::unique_ptr<webtransport::StreamVisitor> uni_stream1 = - std::move(uni_stream_); // Timers aren't running. EXPECT_EQ(OutgoingSubgroupStreamPeer::GetAlarm( - absl::down_cast<OutgoingSubgroupStream*>(uni_stream.get())), + absl::down_cast<OutgoingSubgroupStream*>(uni_stream_.get())), nullptr); - EXPECT_EQ(OutgoingSubgroupStreamPeer::GetAlarm( - absl::down_cast<OutgoingSubgroupStream*>(uni_stream1.get())), - nullptr); - // Second group starts the timer. + // FIN starts the timer. EXPECT_CALL(mock_uni_stream_, visitor) - .WillOnce(Return(uni_stream.get())) - .WillOnce(Return(uni_stream1.get())) + .WillOnce(Return(uni_stream_.get())) .WillRepeatedly([&]() { return uni_stream_.get(); }); - CreateStream(Location(1, 0), 0, 128); + publisher_->OnNewFinAvailable(Location(0, 7), 0); // Group 0 streams now have a timer running. EXPECT_NE(OutgoingSubgroupStreamPeer::GetAlarm( - absl::down_cast<OutgoingSubgroupStream*>(uni_stream.get())), - nullptr); - EXPECT_NE(OutgoingSubgroupStreamPeer::GetAlarm( - absl::down_cast<OutgoingSubgroupStream*>(uni_stream1.get())), - nullptr); - // No timer on group 1. - EXPECT_EQ(OutgoingSubgroupStreamPeer::GetAlarm( absl::down_cast<OutgoingSubgroupStream*>(uni_stream_.get())), nullptr); } @@ -909,6 +891,38 @@ publisher_->OnNewObjectAvailable(Location(1, 1), 0, 128); } +TEST_F(LivePublisherTest, PublishPropertiesAndTimeouts) { + TrackProperties properties( + /*object_delivery_timeout=*/quic::QuicTimeDelta::FromMilliseconds(250), + /*max_cache_duration=*/std::nullopt, + /*subgroup_delivery_timeout=*/quic::QuicTimeDelta::FromMilliseconds(500), + /*publisher_priority=*/42, + /*group_order=*/MoqtDeliveryOrder::kDescending, + /*dynamic_groups=*/std::nullopt, + /*immutable_properties=*/std::nullopt); + EXPECT_CALL(*track_publisher_, properties()) + .WillRepeatedly(ReturnRef(properties)); + + MessageParameters publish_params; + publish_params.set_forward(true); + LivePublisher publish_publisher(framer_, track_publisher_, &bidi_stream_, + kRequestId, kTrackAlias, publish_params, + visitor_.weak_ptr_factory_.Create(), + /*is_publish=*/true); + EXPECT_EQ(publish_publisher.parameters().group_order, + MoqtDeliveryOrder::kDescending); + + // OnSubscribeAccepted initializes publisher delivery timeouts and default + // priority without sending SUBSCRIBE_OK since established_ is already true. + EXPECT_CALL(mock_bidi_stream_, Writev).Times(0); + publish_publisher.OnSubscribeAccepted(); + EXPECT_EQ(publish_publisher.subgroup_delivery_timeout(), + quic::QuicTimeDelta::FromMilliseconds(500)); + EXPECT_EQ(publish_publisher.object_delivery_timeout(), + quic::QuicTimeDelta::FromMilliseconds(250)); + EXPECT_CALL(*track_publisher_, RemoveObjectListener(&publish_publisher)); +} + } // namespace } // namespace moqt::test
diff --git a/quiche/quic/moqt/moqt_object_subscriber.cc b/quiche/quic/moqt/moqt_object_subscriber.cc index ad4b321..4ee2751 100644 --- a/quiche/quic/moqt/moqt_object_subscriber.cc +++ b/quiche/quic/moqt/moqt_object_subscriber.cc
@@ -53,7 +53,10 @@ parameters().subscription_filter->OnLargestObject( data.parameters.largest_object); } - publisher_delivery_timeout_ = data.properties.delivery_timeout(); + publisher_object_delivery_timeout_ = + data.properties.object_delivery_timeout(); + publisher_subgroup_delivery_timeout_ = + data.properties.subgroup_delivery_timeout(); // TODO(martinduke): Is there anything to do with EXPIRES? default_publisher_priority_ = data.properties.default_publisher_priority(); dynamic_groups_ = data.properties.dynamic_groups(); @@ -110,9 +113,8 @@ void LiveSubscriber::MaybeSetPublishDoneAlarm() { if (currently_open_streams_ == 0 && total_streams_.has_value() && clock_ != nullptr) { - quic::QuicTimeDelta timeout = std::min( - parameters().delivery_timeout.value_or(kDefaultDeliveryTimeout), - publisher_delivery_timeout_); + quic::QuicTimeDelta timeout = + std::max(object_delivery_timeout(), subgroup_delivery_timeout()); timeout = std::min(timeout, kMaxPublishDoneTimeout); timeout = std::max(timeout, kMinPublishDoneTimeout); publish_done_alarm_->Set(clock_->ApproximateNow() + timeout);
diff --git a/quiche/quic/moqt/moqt_object_subscriber.h b/quiche/quic/moqt/moqt_object_subscriber.h index 7bc8b87..94ce1c5 100644 --- a/quiche/quic/moqt/moqt_object_subscriber.h +++ b/quiche/quic/moqt/moqt_object_subscriber.h
@@ -7,6 +7,7 @@ #ifndef QUICHE_QUIC_MOQT_MOQT_TRACK_H_ #define QUICHE_QUIC_MOQT_MOQT_TRACK_H_ +#include <algorithm> #include <cstdint> #include <memory> #include <optional> @@ -146,8 +147,21 @@ return default_publisher_priority_; } - quic::QuicTimeDelta publisher_delivery_timeout() const { - return publisher_delivery_timeout_; + quic::QuicTimeDelta publisher_object_delivery_timeout() const { + return publisher_object_delivery_timeout_; + } + quic::QuicTimeDelta publisher_subgroup_delivery_timeout() const { + return publisher_subgroup_delivery_timeout_; + } + quic::QuicTimeDelta object_delivery_timeout() const { + return std::min(const_parameters().object_delivery_timeout.value_or( + kDefaultDeliveryTimeout), + publisher_object_delivery_timeout_); + } + quic::QuicTimeDelta subgroup_delivery_timeout() const { + return std::min(const_parameters().subgroup_delivery_timeout.value_or( + kDefaultDeliveryTimeout), + publisher_subgroup_delivery_timeout_); } SubscribeVisitor* visitor() const { return visitor_; } @@ -192,7 +206,10 @@ // is a protocol violation. bool error_is_allowed_ = true; - quic::QuicTimeDelta publisher_delivery_timeout_ = kDefaultDeliveryTimeout; + quic::QuicTimeDelta publisher_object_delivery_timeout_ = + kDefaultDeliveryTimeout; + quic::QuicTimeDelta publisher_subgroup_delivery_timeout_ = + kDefaultDeliveryTimeout; MoqtPriority default_publisher_priority_ = kDefaultPublisherPriority; bool dynamic_groups_ = kDefaultDynamicGroups; void FetchObjects();
diff --git a/quiche/quic/moqt/moqt_outgoing_queue.cc b/quiche/quic/moqt/moqt_outgoing_queue.cc index 29dda1a..7090dd1 100644 --- a/quiche/quic/moqt/moqt_outgoing_queue.cc +++ b/quiche/quic/moqt/moqt_outgoing_queue.cc
@@ -91,6 +91,9 @@ queue_.back().push_back( std::make_unique<CachedObject>(metadata, std::move(payload), fin)); for (MoqtObjectListener* listener : listeners_) { + if (fin) { + listener->OnNewFinAvailable(sequence, 0); + } listener->OnNewObjectAvailable(sequence, /*subgroup=*/0, default_publisher_priority()); }
diff --git a/quiche/quic/moqt/moqt_outgoing_queue.h b/quiche/quic/moqt/moqt_outgoing_queue.h index 991a07d..0a792a5 100644 --- a/quiche/quic/moqt/moqt_outgoing_queue.h +++ b/quiche/quic/moqt/moqt_outgoing_queue.h
@@ -51,6 +51,7 @@ : clock_(clock), track_(std::move(track)), properties_(std::nullopt, std::nullopt, std::nullopt, std::nullopt, + std::nullopt, new_group_callback != nullptr ? std::optional<bool>(true) : std::nullopt, std::nullopt),
diff --git a/quiche/quic/moqt/moqt_outgoing_queue_test.cc b/quiche/quic/moqt/moqt_outgoing_queue_test.cc index a94bcf5..3372306 100644 --- a/quiche/quic/moqt/moqt_outgoing_queue_test.cc +++ b/quiche/quic/moqt/moqt_outgoing_queue_test.cc
@@ -209,6 +209,7 @@ EXPECT_CALL(queue, PublishObject(0, 0, "a")); EXPECT_CALL(queue, PublishObject(0, 1, "b")); EXPECT_CALL(queue, PublishObject(0, 2, "c")); + EXPECT_CALL(queue, OnNewFinAvailable(Location(0, 3), 0)); EXPECT_CALL(queue, CloseStreamForGroup(0)); EXPECT_CALL(queue, PublishObject(1, 0, "d")); EXPECT_CALL(queue, PublishObject(1, 1, "e"));
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc index a9518a8..d57ef11 100644 --- a/quiche/quic/moqt/moqt_parser.cc +++ b/quiche/quic/moqt/moqt_parser.cc
@@ -354,16 +354,16 @@ value) { last_key = key; switch (static_cast<MessageParameter>(key)) { - case MessageParameter::kDeliveryTimeout: - if (delivery_timeout.has_value()) { + case MessageParameter::kObjectDeliveryTimeout: + if (object_delivery_timeout.has_value()) { status = absl::InvalidArgumentError("Duplicate Message Parameter"); return false; } if (std::get<uint64_t>(value) == 0) { - status = absl::InvalidArgumentError("DELIVERY_TIMEOUT cannot be 0"); - return false; + object_delivery_timeout = quic::QuicTimeDelta::Infinite(); + break; } - delivery_timeout = + object_delivery_timeout = quic::QuicTimeDelta::TryFromMilliseconds(std::get<uint64_t>(value)) .value_or(quic::QuicTimeDelta::Infinite()); break; @@ -374,6 +374,19 @@ return false; } break; + case MessageParameter::kSubgroupDeliveryTimeout: + if (subgroup_delivery_timeout.has_value()) { + status = absl::InvalidArgumentError("Duplicate Message Parameter"); + return false; + } + if (std::get<uint64_t>(value) == 0) { + subgroup_delivery_timeout = quic::QuicTimeDelta::Infinite(); + break; + } + subgroup_delivery_timeout = + quic::QuicTimeDelta::TryFromMilliseconds(std::get<uint64_t>(value)) + .value_or(quic::QuicTimeDelta::Infinite()); + break; case MessageParameter::kExpires: if (expires.has_value()) { status = absl::InvalidArgumentError("Duplicate Message Parameter");
diff --git a/quiche/quic/moqt/moqt_parser_test.cc b/quiche/quic/moqt/moqt_parser_test.cc index fda2c12..bb16687 100644 --- a/quiche/quic/moqt/moqt_parser_test.cc +++ b/quiche/quic/moqt/moqt_parser_test.cc
@@ -760,14 +760,14 @@ EXPECT_EQ(message.options, SetupOptions()); } -TEST_F(MoqtMessageSpecificTest, SubscribeDeliveryTimeoutTwice) { +TEST_F(MoqtMessageSpecificTest, SubscribeObjectDeliveryTimeoutTwice) { char subscribe[] = { 0x03, 0x00, 0x12, 0x01, 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x02, // two params - 0x02, 0x67, 0x10, // delivery_timeout = 10000 - 0x00, 0x67, 0x10, // delivery_timeout = 10000 + 0x02, 0x67, 0x10, // object_delivery_timeout = 10000 + 0x00, 0x67, 0x10, // object_delivery_timeout = 10000 }; absl::StatusOr<std::vector<AnyMoqtControlMessage>> parsed = ParseAllMessages(absl::string_view(subscribe, sizeof(subscribe)), @@ -1121,7 +1121,8 @@ ASSERT_TRUE(parsed.ok()); ASSERT_EQ(parsed->size(), 1); MoqtSubscribe message = std::get<MoqtSubscribe>((*parsed)[0]); - EXPECT_FALSE(message.parameters.delivery_timeout.has_value()); + EXPECT_FALSE(message.parameters.object_delivery_timeout.has_value()); + EXPECT_FALSE(message.parameters.subgroup_delivery_timeout.has_value()); EXPECT_FALSE(message.parameters.forward_has_value()); EXPECT_FALSE(message.parameters.subscription_filter.has_value()); EXPECT_FALSE(message.parameters.group_order.has_value());
diff --git a/quiche/quic/moqt/moqt_publish_stream.cc b/quiche/quic/moqt/moqt_publish_stream.cc index 261d8f0..71ac1d7 100644 --- a/quiche/quic/moqt/moqt_publish_stream.cc +++ b/quiche/quic/moqt/moqt_publish_stream.cc
@@ -78,6 +78,9 @@ if (response_callback_ != nullptr) { // PUBLISH_OK publisher_->Update(message.parameters, /*from_request_ok=*/true); + // In draft-18, PUBLISH_OK can update the group order. This has been + // eliminated since. This is not implemented because it is likely to be + // buggy to change the group order mid-subscription. MoqtResponseCallback callback = std::move(response_callback_); response_callback_ = nullptr; std::move(callback)(message.parameters);
diff --git a/quiche/quic/moqt/moqt_publish_stream_test.cc b/quiche/quic/moqt/moqt_publish_stream_test.cc index a9552c7..e896e57 100644 --- a/quiche/quic/moqt/moqt_publish_stream_test.cc +++ b/quiche/quic/moqt/moqt_publish_stream_test.cc
@@ -86,7 +86,7 @@ // Construct the LivePublisher. parameters_.set_forward(true); - parameters_.delivery_timeout = quic::QuicTimeDelta::FromSeconds(1); + parameters_.object_delivery_timeout = quic::QuicTimeDelta::FromSeconds(1); parameters_.group_order = MoqtDeliveryOrder::kAscending; EXPECT_CALL(visitor_, session).WillRepeatedly(Return(&webtrans_)); @@ -133,7 +133,8 @@ stream_->BindStream(&mock_stream_); // Calls OnStreamBound MoqtRequestOk request_ok; - request_ok.parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(2); + request_ok.parameters.object_delivery_timeout = + quic::QuicTimeDelta::FromSeconds(2); request_ok.parameters.group_order = MoqtDeliveryOrder::kDescending; QUICHE_EXPECT_OK(stream_->OnControlMessage(request_ok)); @@ -141,15 +142,15 @@ ASSERT_TRUE(response_.has_value()); ASSERT_TRUE(std::holds_alternative<MessageParameters>(*response_)); MessageParameters resp_params = std::get<MessageParameters>(*response_); - EXPECT_EQ(resp_params.delivery_timeout, - request_ok.parameters.delivery_timeout); + EXPECT_EQ(resp_params.object_delivery_timeout, + request_ok.parameters.object_delivery_timeout); EXPECT_EQ(resp_params.group_order, request_ok.parameters.group_order); // Verify publisher parameters were updated. const MessageParameters& pub_params = LivePublisherPeer::parameters(*publisher_); - EXPECT_EQ(pub_params.delivery_timeout, - request_ok.parameters.delivery_timeout); + EXPECT_EQ(pub_params.object_delivery_timeout, + request_ok.parameters.object_delivery_timeout); // Group order cannot be updated. EXPECT_EQ(pub_params.group_order, parameters_.group_order); } @@ -183,7 +184,7 @@ MoqtRequestUpdate request_update; request_update.request_id = kRequestId + 2; request_update.existing_request_id = kRequestId; - request_update.parameters.delivery_timeout = + request_update.parameters.object_delivery_timeout = quic::QuicTimeDelta::FromSeconds(3); request_update.parameters.subscriber_priority = 5; request_update.parameters.subscription_filter.emplace( @@ -196,8 +197,8 @@ // Verify publisher parameters were updated. const MessageParameters& pub_params = LivePublisherPeer::parameters(*publisher_); - EXPECT_EQ(pub_params.delivery_timeout, - request_update.parameters.delivery_timeout); + EXPECT_EQ(pub_params.object_delivery_timeout, + request_update.parameters.object_delivery_timeout); EXPECT_EQ(pub_params.subscriber_priority, request_update.parameters.subscriber_priority); @@ -298,7 +299,8 @@ return true; }); MoqtPublish publish = DefaultPublish(); - publish.parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(1); + publish.parameters.object_delivery_timeout = + quic::QuicTimeDelta::FromSeconds(1); QUICHE_EXPECT_OK(stream_->OnControlMessage(publish)); // Verify subscriber is created. @@ -309,7 +311,8 @@ // Verify REQUEST_OK response was sent with new_group_request filtered out // because DYNAMIC_GROUPS is absent/false. MoqtRequestOk expected_ok; - expected_ok.parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(2); + expected_ok.parameters.object_delivery_timeout = + quic::QuicTimeDelta::FromSeconds(2); EXPECT_CALL(mock_stream_, Writev(SerializedControlMessage(expected_ok), _)) .WillOnce(Return(absl::OkStatus())); MessageParameters response_parameters = expected_ok.parameters; @@ -320,7 +323,8 @@ // filtered out. const MessageParameters& sub_params = LiveSubscriberPeer::parameters(*captured_subscriber); - EXPECT_EQ(sub_params.delivery_timeout, response_parameters.delivery_timeout); + EXPECT_EQ(sub_params.object_delivery_timeout, + response_parameters.object_delivery_timeout); EXPECT_EQ(sub_params.new_group_request, std::nullopt); EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone); } @@ -346,8 +350,9 @@ }); MoqtPublish publish = DefaultPublish(); publish.properties = TrackProperties( - /*delivery_timeout=*/std::nullopt, + /*object_delivery_timeout=*/std::nullopt, /*max_cache_duration=*/std::nullopt, + /*subgroup_delivery_timeout=*/std::nullopt, /*publisher_priority=*/std::nullopt, /*group_order=*/std::nullopt, /*dynamic_groups=*/true, @@ -465,7 +470,7 @@ MoqtRequestUpdate request_update; request_update.request_id = kRequestId + 2; request_update.existing_request_id = kRequestId; - request_update.parameters.delivery_timeout = + request_update.parameters.object_delivery_timeout = quic::QuicTimeDelta::FromSeconds(3); EXPECT_CALL(mock_stream_, Writev(ControlMessageOfType(MoqtMessageType::kRequestOk), _)) @@ -476,8 +481,8 @@ ASSERT_NE(captured_subscriber, nullptr); const MessageParameters& sub_params = LiveSubscriberPeer::parameters(*captured_subscriber); - EXPECT_EQ(sub_params.delivery_timeout, - request_update.parameters.delivery_timeout); + EXPECT_EQ(sub_params.object_delivery_timeout, + request_update.parameters.object_delivery_timeout); EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName)); }
diff --git a/quiche/quic/moqt/moqt_publisher.h b/quiche/quic/moqt/moqt_publisher.h index bcbd0d2..c33d991 100644 --- a/quiche/quic/moqt/moqt_publisher.h +++ b/quiche/quic/moqt/moqt_publisher.h
@@ -47,9 +47,9 @@ virtual void OnNewObjectAvailable(Location sequence, std::optional<uint64_t> subgroup, MoqtPriority publisher_priority) = 0; - // Notifies that a pure FIN has arrived following |sequence|. Should not be - // called unless all objects have already been delivered. If not delivered, - // instead set the fin_after_this flag in the PublishedObject. + // Notifies that a pure FIN has arrived following |sequence|. Listeners can + // also find out about FINs by later retrieving an object with fin_after_this + // set, so this doesn't have to be stored if the sender isn't ready for it. virtual void OnNewFinAvailable(Location final_object_in_subgroup, uint64_t subgroup_id) = 0; // Notifies that the a stream is being abandoned (via RESET_STREAM) before @@ -59,7 +59,11 @@ webtransport::StreamErrorCode error_code) = 0; // No further object will be published for the given group, usually due to a - // timeout. The owner of the Listener may want to reset the relevant streams. + // timeout. The owner of the Listener may want to reset the relevant streams, + // MoqtSession will ignore all groups <= |group_id|. MoqtSession is also + // dependent on the caller to periodically retire groups to clear state; state + // will be unbounded if streams are frequently reset and OnGroupAbandoned is + // never called. virtual void OnGroupAbandoned(uint64_t group_id) = 0; // Notifies that the Publisher is being destroyed, so no more objects are
diff --git a/quiche/quic/moqt/moqt_relay_track_publisher_test.cc b/quiche/quic/moqt/moqt_relay_track_publisher_test.cc index 30bf951..46146a5 100644 --- a/quiche/quic/moqt/moqt_relay_track_publisher_test.cc +++ b/quiche/quic/moqt/moqt_relay_track_publisher_test.cc
@@ -649,8 +649,9 @@ ok_parameters.largest_object = kLargestLocation; // Location(3, 2) ok_parameters.expires = quic::QuicTimeDelta::FromSeconds(30); TrackProperties properties( - /*delivery_timeout=*/std::nullopt, + /*object_delivery_timeout=*/std::nullopt, /*max_cache_duration=*/std::nullopt, + /*subgroup_delivery_timeout=*/std::nullopt, /*publisher_priority=*/std::nullopt, /*group_order=*/std::nullopt, /*dynamic_groups=*/true, @@ -801,8 +802,9 @@ ok_parameters.largest_object = Location(5, 2); ok_parameters.expires = quic::QuicTimeDelta::FromSeconds(30); TrackProperties properties( - /*delivery_timeout=*/std::nullopt, + /*object_delivery_timeout=*/std::nullopt, /*max_cache_duration=*/std::nullopt, + /*subgroup_delivery_timeout=*/std::nullopt, /*publisher_priority=*/std::nullopt, /*group_order=*/std::nullopt, /*dynamic_groups=*/true, @@ -851,6 +853,7 @@ TrackProperties properties( /*delivery_timeout=*/std::nullopt, /*max_cache_duration=*/std::nullopt, + /*subgroup_delivery_timeout=*/std::nullopt, /*publisher_priority=*/std::nullopt, /*group_order=*/std::nullopt, /*dynamic_groups=*/true, @@ -974,8 +977,9 @@ ok_parameters.largest_object = kLargestLocation; // Location(3, 2) ok_parameters.expires = quic::QuicTimeDelta::FromSeconds(30); TrackProperties properties( - /*delivery_timeout=*/std::nullopt, + /*object_delivery_timeout=*/std::nullopt, /*max_cache_duration=*/std::nullopt, + /*subgroup_delivery_timeout=*/std::nullopt, /*publisher_priority=*/std::nullopt, /*group_order=*/std::nullopt, /*dynamic_groups=*/true,
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h index f823cbc..01727a9 100644 --- a/quiche/quic/moqt/moqt_session.h +++ b/quiche/quic/moqt/moqt_session.h
@@ -154,10 +154,6 @@ return weak_ptr_factory_.Create(); } - // SessionToPublisherInterface implementation. - bool alternate_delivery_timeout() const override { - return alternate_delivery_timeout_; - } // If |old_priority| is nullopt, the subscription does not have any pending // streams. If it has a value, |old_priority| is the old value to be replaced // by |new_priority|. Subgroup streams send |name| as the first argument. @@ -234,8 +230,6 @@ CleanUpState(); } - void UseAlternateDeliveryTimeout() { alternate_delivery_timeout_ = true; } - private: friend class ControlMessageDispatcher; friend class test::MoqtSessionPeer; @@ -494,10 +488,6 @@ // a GOAWAY. std::unique_ptr<quic::QuicAlarm> goaway_timeout_alarm_; - // If true, use a non-standard design where a timer starts for group n when - // the first object of group n+1 arrives. - bool alternate_delivery_timeout_ = false; - quiche::QuicheWeakPtrFactory<MoqtSessionInterface> weak_ptr_factory_; quiche::QuicheWeakPtrFactory<SessionToPublisherInterface> weak_ptr_factory_for_publishers_;
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc index c981d33..e4961b5 100644 --- a/quiche/quic/moqt/moqt_session_test.cc +++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -95,7 +95,8 @@ MessageParameters SubscribeForTest() { MessageParameters parameters; - parameters.delivery_timeout = quic::QuicTimeDelta::FromMilliseconds(10000); + parameters.object_delivery_timeout = + quic::QuicTimeDelta::FromMilliseconds(10000); parameters.authorization_tokens.emplace_back(AuthTokenType::kOutOfBand, "bar"); parameters.set_forward(true); @@ -1229,8 +1230,9 @@ const MoqtPriority kLocalDefaultPriority = 0x20; bidi_wrapper_ = std::make_unique<MoqtBidiStreamTestWrapper>( ResponseStream(kSubscribeByte)); - TrackProperties properties(std::nullopt, std::nullopt, kLocalDefaultPriority, - std::nullopt, std::nullopt, std::nullopt); + TrackProperties properties(std::nullopt, std::nullopt, std::nullopt, + kLocalDefaultPriority, std::nullopt, std::nullopt, + std::nullopt); EXPECT_CALL(*track, properties) .WillRepeatedly(testing::ReturnRef(properties)); MoqtObjectListener* listener = ReceiveSubscribeSynchronousOk( @@ -1247,10 +1249,10 @@ EXPECT_CALL(mock_bidi_stream_, Writev(ControlMessageOfType(MoqtMessageType::kSubscribe), _)); session_.Subscribe(ftn, &remote_track_visitor_, MessageParameters()); - MoqtSubscribeOk ok( - 2, MessageParameters(), - TrackProperties(std::nullopt, std::nullopt, kPeerDefaultPriority, - std::nullopt, std::nullopt, std::nullopt)); + MoqtSubscribeOk ok(2, MessageParameters(), + TrackProperties(std::nullopt, std::nullopt, std::nullopt, + kPeerDefaultPriority, std::nullopt, + std::nullopt, std::nullopt)); EXPECT_CALL(remote_track_visitor_, OnReply); bidi_wrapper_->ReceiveMessage(ok); @@ -1291,10 +1293,10 @@ EXPECT_CALL(mock_bidi_stream_, Writev(ControlMessageOfType(MoqtMessageType::kSubscribe), _)); session_.Subscribe(ftn, &remote_track_visitor_, MessageParameters()); - MoqtSubscribeOk ok( - 2, MessageParameters(), - TrackProperties(std::nullopt, std::nullopt, kPeerDefaultPriority, - std::nullopt, std::nullopt, std::nullopt)); + MoqtSubscribeOk ok(2, MessageParameters(), + TrackProperties(std::nullopt, std::nullopt, std::nullopt, + kPeerDefaultPriority, std::nullopt, + std::nullopt, std::nullopt)); EXPECT_CALL(remote_track_visitor_, OnReply); bidi_wrapper_->ReceiveMessage(ok); // Omit priority from a datagram. @@ -1336,8 +1338,9 @@ // Create the publisher and the SUBSCRIBE with kLocalDefaultPriority. MockTrackPublisher* track = CreateTrackPublisher(); std::make_shared<MockTrackPublisher>(request.full_track_name); - TrackProperties properties(std::nullopt, std::nullopt, kLocalDefaultPriority, - std::nullopt, std::nullopt, std::nullopt); + TrackProperties properties(std::nullopt, std::nullopt, std::nullopt, + kLocalDefaultPriority, std::nullopt, std::nullopt, + std::nullopt); EXPECT_CALL(*track, properties) .WillRepeatedly(testing::ReturnRef(properties)); MoqtObjectListener* listener = ReceiveSubscribeSynchronousOk( @@ -1421,10 +1424,10 @@ MessageParameters params; params.subscription_filter.emplace(Location(1, 0)); session_.Subscribe(ftn, &remote_track_visitor_, params); - MoqtSubscribeOk ok( - 2, MessageParameters(), - TrackProperties(std::nullopt, std::nullopt, kPeerDefaultPriority, - std::nullopt, std::nullopt, std::nullopt)); + MoqtSubscribeOk ok(2, MessageParameters(), + TrackProperties(std::nullopt, std::nullopt, std::nullopt, + kPeerDefaultPriority, std::nullopt, + std::nullopt, std::nullopt)); EXPECT_CALL(remote_track_visitor_, OnReply); bidi_wrapper_->ReceiveMessage(ok); char datagram[] = {0x01, 0x02, 0x00, 0x00, 0x80, 0x00, 0x08, 0x64, @@ -2341,13 +2344,14 @@ TEST_F(MoqtSessionTest, DeliveryTimeoutParameter) { MoqtSubscribe request = DefaultSubscribe(); - request.parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(1); + request.parameters.object_delivery_timeout = + quic::QuicTimeDelta::FromSeconds(1); bidi_wrapper_ = std::make_unique<MoqtBidiStreamTestWrapper>( ResponseStream(kSubscribeByte)); MockTrackPublisher* track = CreateTrackPublisher(); ReceiveSubscribeSynchronousOk(track, request, bidi_wrapper_.get()); std::optional<quic::QuicTimeDelta> delivery_timeout = - MoqtSessionPeer::GetDeliveryTimeout(&session_, request.request_id); + MoqtSessionPeer::GetObjectDeliveryTimeout(&session_, request.request_id); EXPECT_TRUE(delivery_timeout.has_value() && *delivery_timeout == quic::QuicTimeDelta::FromSeconds(1)); } @@ -2875,12 +2879,13 @@ })); MoqtRequestOk request_ok; - request_ok.parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(2); + request_ok.parameters.object_delivery_timeout = + quic::QuicTimeDelta::FromSeconds(2); bidi_wrapper_->ReceiveMessage(request_ok); ASSERT_TRUE(response.has_value()); EXPECT_TRUE(std::holds_alternative<MessageParameters>(*response)); - EXPECT_EQ(std::get<MessageParameters>(*response).delivery_timeout, + EXPECT_EQ(std::get<MessageParameters>(*response).object_delivery_timeout, quic::QuicTimeDelta::FromSeconds(2)); }
diff --git a/quiche/quic/moqt/moqt_subscribe_stream_test.cc b/quiche/quic/moqt/moqt_subscribe_stream_test.cc index 499bc47..36d24b3 100644 --- a/quiche/quic/moqt/moqt_subscribe_stream_test.cc +++ b/quiche/quic/moqt/moqt_subscribe_stream_test.cc
@@ -162,7 +162,8 @@ MoqtRequestOk request_ok( MessageParameters(), TrackProperties(quic::QuicTimeDelta::FromSeconds(5), std::nullopt, - std::nullopt, std::nullopt, std::nullopt, std::nullopt)); + std::nullopt, std::nullopt, std::nullopt, std::nullopt, + std::nullopt)); QUICHE_EXPECT_OK(stream_->OnControlMessage(request_ok)); // Test cleanup. EXPECT_CALL(mock_remove_callback_, Call);
diff --git a/quiche/quic/moqt/moqt_track_status_stream_test.cc b/quiche/quic/moqt/moqt_track_status_stream_test.cc index d527775..5fda85a 100644 --- a/quiche/quic/moqt/moqt_track_status_stream_test.cc +++ b/quiche/quic/moqt/moqt_track_status_stream_test.cc
@@ -89,7 +89,7 @@ TEST_F(MoqtTrackStatusRequestStreamTest, SendRequestWithParameters) { MessageParameters parameters; - parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(5); + parameters.object_delivery_timeout = quic::QuicTimeDelta::FromSeconds(5); MoqtTrackStatusRequestStream stream = CreateStream(parameters); EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(true)); MoqtTrackStatus expected_message; @@ -118,7 +118,8 @@ MoqtRequestOk ok( MessageParameters(), TrackProperties(quic::QuicTimeDelta::FromSeconds(5), - quic::QuicTimeDelta::FromSeconds(10), std::nullopt, + quic::QuicTimeDelta::FromSeconds(10), + quic::QuicTimeDelta::FromSeconds(5), std::nullopt, std::nullopt, std::nullopt, std::nullopt)); ok.parameters.expires = quic::QuicTimeDelta::FromSeconds(10); ok.parameters.largest_object = Location(1, 2);
diff --git a/quiche/quic/moqt/moqt_uni_stream.cc b/quiche/quic/moqt/moqt_uni_stream.cc index f8776cb..4b11ed9 100644 --- a/quiche/quic/moqt/moqt_uni_stream.cc +++ b/quiche/quic/moqt/moqt_uni_stream.cc
@@ -164,10 +164,9 @@ return; } - quic::QuicTimeDelta delivery_timeout = visitor->delivery_timeout(); - if (!visitor->alternate_delivery_timeout() && - visitor->clock()->ApproximateNow() - object->metadata.arrival_time > - delivery_timeout) { + quic::QuicTimeDelta delivery_timeout = visitor->object_delivery_timeout(); + if (visitor->clock()->ApproximateNow() - object->metadata.arrival_time > + delivery_timeout) { visitor->OnStreamTimeout(index_); stream().ResetWithUserCode(kResetCodeDeliveryTimeout); // No class access below this line. @@ -223,15 +222,20 @@ } ++next_object_; already_delivered_ = 0; - if (object->fin_after_this && !delivery_timeout.IsInfinite() && - !visitor->alternate_delivery_timeout()) { - CreateAndSetAlarm(object->metadata.arrival_time + delivery_timeout); - } } } void OutgoingSubgroupStream::Fin(Location last_object) { QUICHE_DCHECK_EQ(last_object.group, index_.group); + LivePublisherInterface* visitor = visitor_.GetIfAvailable(); + if (visitor != nullptr) { + quic::QuicTimeDelta subgroup_delivery_timeout = + visitor->subgroup_delivery_timeout(); + if (!subgroup_delivery_timeout.IsInfinite()) { + CreateAndSetAlarm(visitor->clock()->ApproximateNow() + + subgroup_delivery_timeout); + } + } if (next_object_ <= last_object.object) { // There is still data to send, do nothing. return; @@ -240,14 +244,6 @@ absl::Status status = webtransport::SendFinOnStream(stream()); QUICHE_BUG_IF(OutgoingSubgroupStream_fin_failed, !status.ok()) << "Writing pure FIN failed."; - LivePublisherInterface* visitor = visitor_.GetIfAvailable(); - if (visitor == nullptr) { - return; - } - quic::QuicTimeDelta delivery_timeout = visitor->delivery_timeout(); - if (!delivery_timeout.IsInfinite()) { - CreateAndSetAlarm(visitor->clock()->ApproximateNow() + delivery_timeout); - } } void OutgoingSubgroupStream::CreateAndSetAlarm(quic::QuicTime deadline) {
diff --git a/quiche/quic/moqt/moqt_uni_stream.h b/quiche/quic/moqt/moqt_uni_stream.h index 0e555c6..e93b620 100644 --- a/quiche/quic/moqt/moqt_uni_stream.h +++ b/quiche/quic/moqt/moqt_uni_stream.h
@@ -96,9 +96,9 @@ public: virtual ~LivePublisherInterface() = default; virtual bool InWindow(Location) = 0; - virtual bool alternate_delivery_timeout() = 0; virtual const quic::QuicClock* clock() = 0; - virtual quic::QuicTimeDelta delivery_timeout() = 0; + virtual quic::QuicTimeDelta object_delivery_timeout() = 0; + virtual quic::QuicTimeDelta subgroup_delivery_timeout() = 0; virtual quic::QuicAlarmFactory* alarm_factory() = 0; // Called when the first byte of an object is written to the stream. virtual void OnObjectSent(Location) = 0; @@ -125,6 +125,14 @@ // webtransport::StreamVisitor overrides. void OnCanWrite() override; void OnStopSendingReceived(webtransport::StreamErrorCode error_code) override; + void OnWriteSideInDataRecvdState() override { + // Any SUBGROUP_DELIVERY_TIMEOUT should be cancelled, as everything is + // acknowledged. + if (delivery_timeout_alarm_ != nullptr) { + delivery_timeout_alarm_->Cancel(); + delivery_timeout_alarm_ = nullptr; + } + } class DeliveryTimeoutDelegate : public quic::QuicAlarm::DelegateWithoutContext {
diff --git a/quiche/quic/moqt/moqt_uni_stream_test.cc b/quiche/quic/moqt/moqt_uni_stream_test.cc index bb981dc..b898a05 100644 --- a/quiche/quic/moqt/moqt_uni_stream_test.cc +++ b/quiche/quic/moqt/moqt_uni_stream_test.cc
@@ -66,9 +66,9 @@ MockLivePublisherInterface() : weak_ptr_factory_(this) {} MOCK_METHOD(bool, InWindow, (Location), (override)); - MOCK_METHOD(bool, alternate_delivery_timeout, (), (override)); MOCK_METHOD(quic::QuicClock*, clock, (), (override)); - MOCK_METHOD(quic::QuicTimeDelta, delivery_timeout, (), (override)); + MOCK_METHOD(quic::QuicTimeDelta, object_delivery_timeout, (), (override)); + MOCK_METHOD(quic::QuicTimeDelta, subgroup_delivery_timeout, (), (override)); MOCK_METHOD(quic::QuicAlarmFactory*, alarm_factory, (), (override)); MOCK_METHOD(void, OnObjectSent, (Location), (override)); MOCK_METHOD(void, OnStreamTimeout, (DataStreamIndex), (override)); @@ -158,9 +158,8 @@ EXPECT_CALL(*track_publisher_, GetCachedObject(0, Optional(0), 0, 0)) .WillOnce(Return(std::move(obj0))); EXPECT_CALL(visitor_, InWindow(Location(0, 0))).WillOnce(Return(true)); - EXPECT_CALL(visitor_, delivery_timeout()) + EXPECT_CALL(visitor_, object_delivery_timeout()) .WillOnce(Return(quic::QuicTimeDelta::FromSeconds(1))); - EXPECT_CALL(visitor_, alternate_delivery_timeout()).WillOnce(Return(false)); EXPECT_CALL(visitor_, clock()).WillOnce(Return(&mock_clock_)); EXPECT_CALL(*track_publisher_, properties()) .WillRepeatedly(ReturnRef(track_properties_)); @@ -186,9 +185,8 @@ EXPECT_CALL(*track_publisher_, GetCachedObject(0, Optional(0), 0, 0)) .WillOnce(Return(std::move(obj0))); EXPECT_CALL(visitor_, InWindow(Location(0, 0))).WillOnce(Return(true)); - EXPECT_CALL(visitor_, delivery_timeout()) + EXPECT_CALL(visitor_, object_delivery_timeout()) .WillOnce(Return(quic::QuicTimeDelta::FromSeconds(1))); - EXPECT_CALL(visitor_, alternate_delivery_timeout()).WillOnce(Return(false)); mock_clock_.AdvanceTime(quic::QuicTimeDelta::FromSeconds(2)); EXPECT_CALL(visitor_, clock()).WillOnce(Return(&mock_clock_)); EXPECT_CALL(visitor_, OnStreamTimeout(index_)); @@ -202,9 +200,8 @@ EXPECT_CALL(*track_publisher_, GetCachedObject(0, Optional(0), 0, 0)) .WillOnce(Return(std::move(obj0))); EXPECT_CALL(visitor_, InWindow(Location(0, 0))).WillOnce(Return(true)); - EXPECT_CALL(visitor_, delivery_timeout()) + EXPECT_CALL(visitor_, object_delivery_timeout()) .WillOnce(Return(quic::QuicTimeDelta::FromSeconds(1))); - EXPECT_CALL(visitor_, alternate_delivery_timeout()).WillOnce(Return(false)); EXPECT_CALL(visitor_, clock).WillOnce(Return(&mock_clock_)); EXPECT_CALL(*track_publisher_, properties()) .WillRepeatedly(ReturnRef(track_properties_)); @@ -217,44 +214,13 @@ "status: INTERNAL: error"); } -TEST_F(OutgoingSubgroupStreamTest, OnCanWriteSetsAlarm) { - PublishedObject obj0 = DefaultObject(); - obj0.fin_after_this = true; - EXPECT_CALL(mock_stream_, CanWrite()) - .WillOnce(Return(true)) - .WillOnce(Return(false)); - EXPECT_CALL(*track_publisher_, GetCachedObject(0, Optional(0), 0, 0)) - .WillOnce(Return(std::move(obj0))); - EXPECT_CALL(visitor_, InWindow(Location(0, 0))).WillOnce(Return(true)); - EXPECT_CALL(visitor_, delivery_timeout()) - .WillRepeatedly(Return(quic::QuicTimeDelta::FromSeconds(1))); - EXPECT_CALL(visitor_, alternate_delivery_timeout()) - .WillRepeatedly(Return(false)); - EXPECT_CALL(visitor_, clock).WillOnce(Return(&mock_clock_)); - - EXPECT_CALL(*track_publisher_, properties()) - .WillRepeatedly(ReturnRef(track_properties_)); - EXPECT_CALL(mock_stream_, Writev) - .WillOnce([&](absl::Span<quiche::QuicheMemSlice> data, - const webtransport::StreamWriteOptions& options) { - EXPECT_TRUE(options.send_fin()); - return absl::OkStatus(); - }); - EXPECT_CALL(visitor_, OnObjectSent(Location(0, 0))); - ExpectAlarm(); - stream_->OnCanWrite(); - EXPECT_CALL(mock_stream_, ResetWithUserCode(kResetCodeDeliveryTimeout)); - EXPECT_CALL(visitor_, OnStreamTimeout(index_)); - alarm_factory_.FireAlarm(OutgoingSubgroupStreamPeer::GetAlarm(stream_.get())); -} - TEST_F(OutgoingSubgroupStreamTest, Fin) { // Replace stream_ with one where next_object_ is 1. EXPECT_CALL(visitor_, OnDataStreamDestroyed(index_)); CreateStream(1); // last_object.object < next_object: sends pure FIN ExpectFin(); - EXPECT_CALL(visitor_, delivery_timeout()) + EXPECT_CALL(visitor_, subgroup_delivery_timeout()) .WillOnce(Return(quic::QuicTimeDelta::FromSeconds(1))); EXPECT_CALL(visitor_, clock()).WillOnce(Return(&mock_clock_)); ExpectAlarm(); @@ -268,8 +234,12 @@ // Delivery is blocked. EXPECT_CALL(mock_stream_, CanWrite).WillOnce(Return(false)); stream_->OnCanWrite(); - // FIN does nothing because last object hasn't been sent. Rely on the cache - // to set object.fin_after_this. + // FIN does nothing except set the delivery timeout, because last object + // hasn't been sent. Rely on the cache to set object.fin_after_this. + EXPECT_CALL(visitor_, subgroup_delivery_timeout()) + .WillOnce(Return(quic::QuicTimeDelta::FromSeconds(1))); + EXPECT_CALL(visitor_, clock()).WillOnce(Return(&mock_clock_)); + ExpectAlarm(); EXPECT_CALL(mock_stream_, Writev).Times(0); stream_->Fin(Location(0, 0)); } @@ -291,10 +261,10 @@ .WillOnce(Return(std::move(obj0))); EXPECT_CALL(visitor_, InWindow).WillRepeatedly(Return(true)); EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(true)); - EXPECT_CALL(visitor_, delivery_timeout()) + EXPECT_CALL(visitor_, object_delivery_timeout()) .WillRepeatedly(Return(quic::QuicTimeDelta::FromSeconds(1))); - EXPECT_CALL(visitor_, alternate_delivery_timeout()) - .WillRepeatedly(Return(false)); + EXPECT_CALL(visitor_, subgroup_delivery_timeout()) + .WillRepeatedly(Return(quic::QuicTimeDelta::FromSeconds(1))); EXPECT_CALL(visitor_, clock()).WillRepeatedly(Return(&mock_clock_)); EXPECT_CALL(*track_publisher_, properties()) .WillRepeatedly(ReturnRef(track_properties_)); @@ -327,7 +297,6 @@ return absl::OkStatus(); }); EXPECT_CALL(visitor_, OnObjectSent).Times(0); - ExpectAlarm(); stream_->OnCanWrite(); }
diff --git a/quiche/quic/moqt/test_tools/mock_moqt_session.h b/quiche/quic/moqt/test_tools/mock_moqt_session.h index 5aa79d4..55d5220 100644 --- a/quiche/quic/moqt/test_tools/mock_moqt_session.h +++ b/quiche/quic/moqt/test_tools/mock_moqt_session.h
@@ -45,7 +45,6 @@ public: MockSessionToPublisherInterface() : weak_ptr_factory_(this) {} ~MockSessionToPublisherInterface() override = default; - MOCK_METHOD(bool, alternate_delivery_timeout, (), (const, override)); MOCK_METHOD(void, UpdateTrackPriority, (const FullTrackName&, std::optional<MoqtTrackPriority>, MoqtTrackPriority),
diff --git a/quiche/quic/moqt/test_tools/moqt_mock_visitor.h b/quiche/quic/moqt/test_tools/moqt_mock_visitor.h index a24483f..409f860 100644 --- a/quiche/quic/moqt/test_tools/moqt_mock_visitor.h +++ b/quiche/quic/moqt/test_tools/moqt_mock_visitor.h
@@ -197,6 +197,9 @@ largest_location_ = location; } for (MoqtObjectListener* listener : listeners_) { + if (fin) { + listener->OnNewFinAvailable(location, subgroup); + } listener->OnNewObjectAvailable(location, subgroup, 128); } }
diff --git a/quiche/quic/moqt/test_tools/moqt_session_peer.h b/quiche/quic/moqt/test_tools/moqt_session_peer.h index 6f7d673..bf7e327 100644 --- a/quiche/quic/moqt/test_tools/moqt_session_peer.h +++ b/quiche/quic/moqt/test_tools/moqt_session_peer.h
@@ -181,13 +181,22 @@ return session->goaway_timeout_alarm_.get(); } - static quic::QuicTimeDelta GetDeliveryTimeout(MoqtSession* session, - uint64_t request_id) { + static quic::QuicTimeDelta GetObjectDeliveryTimeout(MoqtSession* session, + uint64_t request_id) { auto it = session->published_subscriptions_.find(request_id); if (it == session->published_subscriptions_.end()) { return quic::QuicTimeDelta::Zero(); } - return it->second->delivery_timeout(); + return it->second->object_delivery_timeout(); + } + + static quic::QuicTimeDelta GetSubgroupDeliveryTimeout(MoqtSession* session, + uint64_t request_id) { + auto it = session->published_subscriptions_.find(request_id); + if (it == session->published_subscriptions_.end()) { + return quic::QuicTimeDelta::Zero(); + } + return it->second->subgroup_delivery_timeout(); } static absl::string_view GetImplementationString(MoqtSession* session) {
diff --git a/quiche/quic/moqt/test_tools/moqt_simulator.cc b/quiche/quic/moqt/test_tools/moqt_simulator.cc index 5090baa..5d643e4 100644 --- a/quiche/quic/moqt/test_tools/moqt_simulator.cc +++ b/quiche/quic/moqt/test_tools/moqt_simulator.cc
@@ -306,9 +306,6 @@ if (parameters_.bitrate_adaptation) { client_session()->SetMonitoringInterfaceForTrack(TrackName(), &adjuster_); } - if (parameters_.alternative_timeout) { - client_session()->UseAlternateDeliveryTimeout(); - } publisher_.Add(generator_.queue()); modification_box_.OnBeforeSimulationStart(); @@ -323,7 +320,13 @@ subscription_parameters.oack_window_size = parameters_.deadline; } if (!parameters_.delivery_timeout.IsInfinite()) { - subscription_parameters.delivery_timeout = parameters_.delivery_timeout; + if (parameters_.alternative_timeout) { + subscription_parameters.subgroup_delivery_timeout = + parameters_.delivery_timeout; + } else { + subscription_parameters.object_delivery_timeout = + parameters_.delivery_timeout; + } } server_session()->RelativeJoiningFetch(TrackName(), &receiver_, 0, subscription_parameters);
diff --git a/quiche/quic/moqt/test_tools/moqt_test_message.h b/quiche/quic/moqt/test_tools/moqt_test_message.h index 51dc7fe..6c023c7 100644 --- a/quiche/quic/moqt/test_tools/moqt_test_message.h +++ b/quiche/quic/moqt/test_tools/moqt_test_message.h
@@ -92,7 +92,8 @@ inline MessageParameters SubscribeForTest() { MessageParameters parameters; - parameters.delivery_timeout = quic::QuicTimeDelta::FromMilliseconds(10000); + parameters.object_delivery_timeout = + quic::QuicTimeDelta::FromMilliseconds(10000); parameters.authorization_tokens.emplace_back(AuthTokenType::kOutOfBand, "bar"); parameters.set_forward(true); @@ -734,11 +735,11 @@ private: uint8_t raw_packet_[36] = { - 0x03, 0x00, 0x21, 0x01, // request_id = 1 - 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" - 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" - 0x06, // 6 parameters - 0x02, 0xa7, 0x10, // delivery_timeout = 10000 ms + 0x03, 0x00, 0x21, 0x01, // request_id = 1 + 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" + 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" + 0x06, // 6 parameters + 0x02, 0xa7, 0x10, // object_delivery_timeout = 10000 ms 0x01, 0x05, 0x03, 0x00, 0x62, 0x61, 0x72, // authorization_tag = "bar" 0x0d, 0x01, // forward = true 0x10, 0x20, // subscriber_priority = 0x20 @@ -789,8 +790,10 @@ /*track_alias=*/2, MessageParameters(), // Set in the constructor. TrackProperties( - /*delivery_timeout=*/quic::QuicTimeDelta::FromMilliseconds(10000), + /*object_delivery_timeout=*/ + quic::QuicTimeDelta::FromMilliseconds(10000), /*max_cache_duration=*/quic::QuicTimeDelta::FromMilliseconds(10000), + /*subgroup_delivery_timeout=*/std::nullopt, /*publisher_priority=*/std::nullopt, /*group_order=*/MoqtDeliveryOrder::kDescending, /*dynamic_groups=*/std::nullopt, @@ -803,7 +806,7 @@ 0x08, 0x03, // expires = 3 0x01, 0x02, 0x0c, 0x14, // largest_location = (12, 20) // Properties - 0x02, 0xa7, 0x10, // delivery_timeout = 10000 + 0x02, 0xa7, 0x10, // object_delivery_timeout = 10000 0x02, 0xa7, 0x10, // max_cache_duration = 10000 0x1e, 0x02 // default_publisher_group_order = 2 }; @@ -908,7 +911,7 @@ public: RequestUpdateMessage() : TestMessageBase() { SetWireImage(raw_packet_, sizeof(raw_packet_)); - request_update_.parameters.delivery_timeout = + request_update_.parameters.object_delivery_timeout = quic::QuicTimeDelta::FromMilliseconds(10000); request_update_.parameters.set_forward(true); request_update_.parameters.subscriber_priority = 0xaa; @@ -942,7 +945,7 @@ uint8_t raw_packet_[20] = { 0x02, 0x00, 0x11, 0x02, 0x00, // request IDs 2 and 0 0x04, // Four parameters - 0x02, 0xa7, 0x10, // delivery_timeout = 10000 + 0x02, 0xa7, 0x10, // object_delivery_timeout = 10000 0x0e, 0x01, // forward = true 0x10, 0x80, 0xaa, // subscriber_priority = 0xaa 0x01, 0x04, 0x04, 0x03, 0x01, 0x02, // Absolute Range: (3, 1) to 5 @@ -1097,7 +1100,7 @@ 0x01, // 1 parameter 0x09, 0x02, 0x05, 0x01, // Largest Object = (5, 1) // Properties - 0x02, 0xa7, 0x10, // delivery_timeout = 10000 + 0x02, 0xa7, 0x10, // object_delivery_timeout = 10000 0x02, 0xa7, 0x10, // max_cache_duration = 10000 0x1e, 0x02 // default_publisher_group_order = 2 }; @@ -1105,8 +1108,10 @@ MoqtRequestOk request_ok_ = { MessageParameters(), // Set in the constructor. TrackProperties( - /*delivery_timeout=*/quic::QuicTimeDelta::FromMilliseconds(10000), + /*object_delivery_timeout=*/ + quic::QuicTimeDelta::FromMilliseconds(10000), /*max_cache_duration=*/quic::QuicTimeDelta::FromMilliseconds(10000), + /*subgroup_delivery_timeout=*/std::nullopt, /*publisher_priority=*/std::nullopt, /*group_order=*/MoqtDeliveryOrder::kDescending, /*dynamic_groups=*/std::nullopt, @@ -1517,10 +1522,10 @@ /*end_of_track=*/false, /*end_location=*/Location{5, 3}, MessageParameters(), - TrackProperties(std::nullopt, - quic::QuicTimeDelta::FromMilliseconds(10000), - std::nullopt, MoqtDeliveryOrder::kDescending, - std::nullopt, std::nullopt), + TrackProperties( + std::nullopt, quic::QuicTimeDelta::FromMilliseconds(10000), + std::nullopt, std::nullopt, MoqtDeliveryOrder::kDescending, + std::nullopt, std::nullopt), }; }; @@ -1585,7 +1590,7 @@ FullTrackName("foo", "bar"), /*track_alias=*/4, MessageParameters(), - TrackProperties(std::nullopt, std::nullopt, std::nullopt, + TrackProperties(std::nullopt, std::nullopt, std::nullopt, std::nullopt, MoqtDeliveryOrder::kDescending, std::nullopt, std::nullopt), };