diff --git a/docs/dev/local-forwarder-flow.md b/docs/dev/local-forwarder-flow.md index 58821ce8..0e2b8ab2 100644 --- a/docs/dev/local-forwarder-flow.md +++ b/docs/dev/local-forwarder-flow.md @@ -48,9 +48,13 @@ Executor legend used below: ## Publish -A publisher's local forwarder **is** the publisher forwarder — the same object serves as the -on-thread data-plane forwarder and as the registry's publisher entry (no second forwarder is -constructed on `relayExec_`, unlike the `RelayExec`-mode branch in `publishWithSession`). +A publisher's local forwarder **is** the publisher forwarder — a single forwarder serves the +whole track, living on the publisher's executor rather than on `relayExec_`. Ownership lives in +`tlForwarders_` on the publisher's executor, **not** in the registry: once registration +completes, `registerPublishOnRelayExec` calls `registry_.clearForwarder(ftn)`, so the registry +entry keeps its subscription/namespace bookkeeping but drops its forwarder ref. Code that needs +the forwarder resolves it by `FullTrackName` from `tlForwarders_` on the owning executor, rather +than holding a registry ref. ### Setup @@ -65,7 +69,8 @@ constructed on `relayExec_`, unlike the `RelayExec`-mode branch in `publishWithS [Relay] └─▶ registerPublishOnRelayExec() [Relay] ├─▶ publishWithSession() # registry + namespace tree [Relay] │ └─▶ addSubscriberAndPublish() # attach existing SUBSCRIBE_NAMESPACE subs (see Publish fan-out) -[Relay] └─▶ relayChainFilter->setDownstream(topNFilter) +[Relay] ├─▶ relayChainFilter->setDownstream(topNFilter) +[Relay] └─▶ registry_.clearForwarder(ftn) # drop registry ref; tlForwarders_ owns the fwd [Pub] return {consumer, replyTask} # immediate ``` @@ -133,11 +138,13 @@ upstream work, and nests a single sortie to `[Pub]` to wire the channel sub. [Sub] ├─▶ localFwd->addSubscriber() # so `forward` is right pre-hop [Sub] └─▶ ⇢⇢▶ [Relay] attachNewLocalForwarderOnRelayExec() [Relay] ├─▶ joinOrPrepareUpstreamSubscription() # registry: first vs subsequent -[Relay] ├─▶ buildLocalToPublisherCallbacks() [Relay] └─▶ ⇢⇢▶ [Pub] single sortie: +[Pub] ├─ if SUBSEQUENT: resolve live publisherFwd from tlForwarders_ +[Pub] ├─▶ buildLocalToPublisherCallbacks() [Pub] ├─ if FIRST subscriber: [Pub] │ └─▶ installPublisherForwarderCallbackChain() # chain + tlForwarders_ slot [Pub] ├─▶ installChannelSubscriber(localFwd ↔ publisherFwd) +[Pub] ├─ if SUBSEQUENT: read seed largest/extensions [Pub] └─ if FIRST subscriber: [Pub] ├─▶ addChannelSubscriber(relayChain, passive) [Pub] └─▶ subscribeUpstreamAndApplyOk() @@ -147,7 +154,7 @@ upstream work, and nests a single sortie to `[Pub]` to wire the channel sub. [Sub] ◀── back on subscriberExec (tail) [Sub] ├─ sawOnEmpty? ─▶ teardownLocalForwarderOnFailure() [Sub] ├─ error? ─▶ publishDone + remove -[Sub] ├─▶ apply upstreamOk extensions / largest +[Sub] ├─▶ apply largest/extensions (upstreamOk if FIRST, else seeded) [Sub] ├─▶ replayPendingFowarderEvents() # drain buffered events [Sub] └─▶ return sub ``` @@ -158,12 +165,16 @@ publisher-forwarder control chain and claims the `tlForwarders_` slot via `installPublisherForwarderCallbackChain` — the **same** wiring the publish path uses, so a subscribe-initiated publisher forwarder is symmetric with a publish-initiated one. Subsequent subscribers on the same thread hit the `attachSubscriber` fast path (now also for -subscribe-initiated tracks); on other threads they take only the `installChannelSubscriber` half -of the sortie. +subscribe-initiated tracks); on other threads they take the non-first half of the sortie: +resolve the live publisher forwarder from `tlForwarders_` on the publisher exec, install the +channel sub, and seed `largest`/extensions from it. The forwarder is created on `relayExec_` (in `joinOrPrepareUpstreamSubscription`) with a **null** callback, then gets its real callback + tl slot installed on `[Pub]` in the first-subscriber -sortie — `tlForwarders_.get()` must run on the forwarder's own exec. It uses `removeOnEmpty=true` +sortie — `tlForwarders_.get()` must run on the forwarder's own exec. Once that sortie claims the +tl slot, the relay-exec continuation calls `registry_.clearForwarder(ftn)`, transferring ownership +to `tlForwarders_` and leaving the registry with only bookkeeping — the same as the publish path. +It uses `removeOnEmpty=true` (subscribe-initiated tracks drop on empty, unlike publish): `LocalForwarderCallback` vacates the tl slot on `[Pub]` when the last subscriber leaves, while `onEmptyImpl` unsubscribes upstream and removes the registry entry on `[Relay]`. diff --git a/src/MoqxRelay.cpp b/src/MoqxRelay.cpp index f03b3c37..04a0084d 100644 --- a/src/MoqxRelay.cpp +++ b/src/MoqxRelay.cpp @@ -505,6 +505,11 @@ Subscriber::PublishResult MoqxRelay::publishFromPublisherExec( return folly::makeUnexpected(std::move(*err)); } + // Capture a same-exec replace before createPublisherForwarder overwrites the tl slot. + std::shared_ptr displacedLocalFwd; + if (auto* localReg = tlForwarders_.get()) { + displacedLocalFwd = localReg->get(pub.fullTrackName); + } // createPublisherForwarder claims the tlForwarders_ slot on this exec. auto localPubFwd = createPublisherForwarder(pub); @@ -524,7 +529,9 @@ Subscriber::PublishResult MoqxRelay::publishFromPublisherExec( handle = std::move(handle), session = std::move(session), localPubFwd, - crossExecFilter]() mutable -> folly::coro::Task> { + crossExecFilter, + displacedLocalFwd = std::move(displacedLocalFwd)]( + ) mutable -> folly::coro::Task> { co_return co_await folly::coro::co_withExecutor( folly::getKeepAliveToken(exec), relay->registerPublishOnRelayExec( @@ -532,7 +539,8 @@ Subscriber::PublishResult MoqxRelay::publishFromPublisherExec( std::move(handle), std::move(session), std::move(localPubFwd), - std::move(crossExecFilter) + std::move(crossExecFilter), + std::move(displacedLocalFwd) ) ); } @@ -549,11 +557,17 @@ folly::coro::Task> MoqxRelay::registerP std::shared_ptr handle, std::shared_ptr session, std::shared_ptr publisherFwd, - std::shared_ptr relayChainFilter + std::shared_ptr relayChainFilter, + std::shared_ptr displacedLocalFwd ) { auto ftn = pub.fullTrackName; - auto setup = - publishWithSession(std::move(pub), std::move(handle), std::move(session), publisherFwd); + auto setup = publishWithSession( + std::move(pub), + std::move(handle), + std::move(session), + publisherFwd, + std::move(displacedLocalFwd) + ); if (setup.hasError()) { co_return folly::makeUnexpected(setup.error()); } @@ -563,6 +577,9 @@ folly::coro::Task> MoqxRelay::registerP << "registerPublishOnRelayExec: topNFilter always present in MT mode"; relayChainFilter->setDownstream(topNView->topNFilter); + // tlForwarders_ on the publisher exec owns the forwarder now; relayExec_ must not hold it. + registry_.clearForwarder(ftn); + co_return setup.value().publishOk; } @@ -597,7 +614,8 @@ MoqxRelay::PublishSetupResult MoqxRelay::publishWithSession( PublishRequest pub, std::shared_ptr handle, std::shared_ptr session, - std::shared_ptr forwarder + std::shared_ptr forwarder, + std::shared_ptr displacedLocalFwd ) { // Handle duplicate publisher at relay level before registering in the tree. if (!forwarder) { @@ -622,18 +640,38 @@ MoqxRelay::PublishSetupResult MoqxRelay::publishWithSession( // drainSubscriber) can't destroy the forwarder mid-forEachSubscriber. XLOG(DBG1) << "New publisher for existing subscription"; auto& evicted = *publishEntry.evicted; + PublishDone done{ + RequestID(0), + PublishDoneStatusCode::SUBSCRIPTION_ENDED, + 0, // filled in by session + "upstream disconnect" + }; // Null handle => previous publisher already terminated and onPublishDone() tore it down; skip. - if (evicted.handle) { + if (evicted.handle && mode() == Mode::LocalForwarder) { + // The old forwarder lives on its publisher's exec; unsubscribe and drain it there. + runOnSessionExec( + relayExec_, + evicted.publisherExec, + [this, ftn = pub.fullTrackName, h = evicted.handle, displacedLocalFwd, done]() mutable { + h->unsubscribe(); + // Set when the new publisher displaced it from this same exec's slot; otherwise + // the old publisher was elsewhere, so it still holds ftn in that exec's registry. + auto oldFwd = displacedLocalFwd; + if (!oldFwd) { + auto* localReg = tlForwarders_.get(); + oldFwd = localReg ? localReg->get(ftn) : nullptr; + } + if (oldFwd) { + oldFwd->publishDone(std::move(done)); + } + } + ); + } else if (evicted.handle) { // unsubscribe mutates the old publisher's session inline, so hop to its exec. runOnSessionExec(relayExec_, evicted.publisherExec, [h = evicted.handle] { h->unsubscribe(); }); - evicted.forwarder->publishDone( - {RequestID(0), - PublishDoneStatusCode::SUBSCRIPTION_ENDED, - 0, // filled in by session - "upstream disconnect"} - ); + evicted.forwarder->publishDone(std::move(done)); } } @@ -687,6 +725,7 @@ MoqxRelay::PublishSetupResult MoqxRelay::publishWithSession( auto* publisherExec = relayExec_ ? session->getExecutor() : nullptr; if (!addSubscriberAndPublish( outSession, + pub.fullTrackName, forwarder, info.forward, /*pinned=*/true, @@ -722,6 +761,7 @@ MoqxRelay::PublishSetupResult MoqxRelay::publishWithSession( auto* publisherExec = relayExec_ ? session->getExecutor() : nullptr; if (!addSubscriberAndPublish( outSession, + pub.fullTrackName, forwarder, info.forward, /*pinned=*/true, @@ -816,6 +856,7 @@ std::optional MoqxRelay::startPublish( // Returns false on synchronous failure. bool MoqxRelay::addSubscriberAndPublish( std::shared_ptr subscriberSession, + FullTrackName ftn, std::shared_ptr forwarder, bool forward, bool pinned, @@ -828,7 +869,7 @@ bool MoqxRelay::addSubscriberAndPublish( folly::getKeepAliveToken(subscriberSession->getExecutor()), addSubscriberAndPublishViaLocalForwarder( subscriberSession, - forwarder, + std::move(ftn), publisherExec, forward, pinned @@ -837,6 +878,14 @@ bool MoqxRelay::addSubscriberAndPublish( .start(); return true; } + // Non-LF: relayExec_ owns the forwarder; the registry still holds it when not passed in. + if (!forwarder) { + forwarder = registry_.getForwarder(ftn); + } + if (!forwarder) { + XLOG(ERR) << "addSubscriberAndPublish: no forwarder in registry for " << ftn; + return false; + } folly::Executor* subscriberExec = relayExec_ ? subscriberSession->getExecutor() : nullptr; auto p = startPublish(subscriberSession, forwarder, forward, pinned, subscriberExec); if (!p) { @@ -859,19 +908,26 @@ namespace { // publishDone if given. void teardownLocalForwarderOnFailure( folly::Executor* publisherExec, - std::shared_ptr publisherFwd, + std::weak_ptr publisherFwd, folly::Executor* subscriberExec, folly::Executor* relayExec, const std::shared_ptr& localFwd = nullptr, std::string publishDoneReason = {} ) { - if (publisherFwd && publisherExec) { + if (publisherExec) { folly::via( publisherExec, [pf = std::move(publisherFwd), ex = subscriberExec, re = relayExec]() noexcept { - pf->removeChannelSubscriberByExec(ex); + // lock on the pub exec: resolves the exact wired forwarder or null if it is gone. + // Gone when the tl slot was vacated before this hop ran — publishDone, or onEmpty + // for subscribe-initiated forwarders — since that slot holds the only long-lived ref. + auto fwd = pf.lock(); + if (!fwd) { + return; + } + fwd->removeChannelSubscriberByExec(ex); if (re) { - pf->removeChannelSubscriberByExec(re); + fwd->removeChannelSubscriberByExec(re); } } ); @@ -1119,13 +1175,12 @@ void replayPendingFowarderEvents( // it to publisherFwd as a channel subscriber (isNew path), and awaits the publish reply. folly::coro::Task MoqxRelay::addSubscriberAndPublishViaLocalForwarder( std::shared_ptr subscriberSession, - std::shared_ptr publisherFwd, + FullTrackName ftn, folly::Executor* publisherExec, bool forward, bool pinned ) { auto* subscriberExec = subscriberSession->getExecutor(); - const auto& ftn = publisherFwd->fullTrackName(); // Fast path: local forwarder already exists on this thread. if (auto* localReg = tlForwarders_.get()) { @@ -1138,18 +1193,27 @@ folly::coro::Task MoqxRelay::addSubscriberAndPublishViaLocalForwarder( } } - // Capture largest/extensions on publisherExec; reading them on subscriberExec would - // race the publisher advancing largest_. + // Resolve the publisher forwarder from tlForwarders_ and seed largest/extensions on its own + // exec; the registry no longer holds it, and reading those fields elsewhere races the publisher. + std::shared_ptr publisherFwd; std::optional seedLargest; Extensions seedExtensions; co_await folly::coro::co_withExecutor( folly::getKeepAliveToken(publisherExec), [&]() -> folly::coro::Task { - seedLargest = publisherFwd->largest(); - seedExtensions = publisherFwd->extensions(); + if (auto* publisherReg = tlForwarders_.get()) { + publisherFwd = publisherReg->get(ftn); + } + if (publisherFwd) { + seedLargest = publisherFwd->largest(); + seedExtensions = publisherFwd->extensions(); + } co_return; }() ); + if (!publisherFwd) { + co_return; + } // ready: the forwarder is built from the publisher's snapshot, so no attacher waits. auto [localFwd, isNew, localReg] = acquireLocalForwarder( @@ -1497,7 +1561,7 @@ folly::coro::Task MoqxRelay::subscribeNames node->forEachPublish([&](const std::string& trackName, const std::shared_ptr& publishSession) { FullTrackName ftn{prefix, trackName}; - auto forwarder = registry_.getForwarder(ftn); + auto forwarder = registry_.getForwarderIfExists(ftn); if (!forwarder) { XLOG(ERR) << "Invalid state, no subscription for publish ftn=" << ftn; return; @@ -1518,7 +1582,8 @@ folly::coro::Task MoqxRelay::subscribeNames auto* publisherExec = relayExec_ ? publishSession->getExecutor() : nullptr; if (!addSubscriberAndPublish( session, - forwarder, + ftn, + *forwarder, subNs.forward, /*pinned=*/true, publisherExec @@ -1612,14 +1677,15 @@ folly::coro::Task MoqxRelay::subscribeTracks( return; } FullTrackName ftn{prefix, trackName}; - auto forwarder = registry_.getForwarder(ftn); + auto forwarder = registry_.getForwarderIfExists(ftn); if (!forwarder) { return; } auto* publisherExec = relayExec_ ? publishSession->getExecutor() : nullptr; if (!addSubscriberAndPublish( session, - forwarder, + ftn, + *forwarder, subTracks.forward, /*pinned=*/true, publisherExec @@ -1789,51 +1855,73 @@ folly::coro::Task MoqxRelay::attachNewLocalForwa attach.error = std::move(*sr.error); co_return attach; // pending dtor fires when sr is destroyed, cleaning the registry } - attach.publisherFwd = sr.publisherForwarder; attach.publisherExec = sr.publisherExec; - if (!attach.publisherFwd || !attach.publisherExec) { + if (!attach.publisherExec) { co_return attach; } attach.ownsRelayChain = sr.firstSetup.has_value(); - // Single publisherExec sortie for all publisherFwd mutation: the local channel sub and, - // for the first subscriber, the passive relay chain + upstream subscribe. Merging the - // two installs drops a relayExec_ round-trip. - auto cbs = buildLocalToPublisherCallbacks( - localReg, - ftn, - localFwd, - attach.publisherFwd, - attach.publisherExec, - subscriberExec - ); - attach.finalCallback = cbs.finalCallback; - std::shared_ptr relayChainFilter; std::optional> upstreamResult; + // Local shared lives only across this hop; only a weak escapes via attach.publisherFwd. + std::shared_ptr publisherFwd = sr.publisherForwarder; + // One publisherExec sortie does all publisherFwd work: resolve the live forwarder + // (subsequent), build+install the channel sub, and read seeds / run the first-subscriber + // upstream subscribe — so a subsequent subscriber wires and seeds in a single hop. co_await folly::coro::co_withExecutor( folly::getKeepAliveToken(attach.publisherExec), [&]() -> folly::coro::Task { + if (!sr.firstSetup) { + auto* publisherReg = tlForwarders_.get(); + publisherFwd = publisherReg ? publisherReg->get(ftn) : nullptr; + if (!publisherFwd) { + // Live forwarder vanished between the registry check and this regrab (teardown race). + attach.error = folly::makeUnexpected(SubscribeError{ + subReq.requestID, + SubscribeErrorCode::INTERNAL_ERROR, + "publisher forwarder gone" + }); + co_return; + } + } + attach.publisherFwd = publisherFwd; + auto cbs = buildLocalToPublisherCallbacks( + localReg, + ftn, + localFwd, + publisherFwd, + attach.publisherExec, + subscriberExec + ); + attach.finalCallback = cbs.finalCallback; + // First subscriber installs the publisher chain + tl slot before any sub is added. if (sr.firstSetup) { - installPublisherForwarderCallbackChain(ftn, attach.publisherFwd, /*removeOnEmpty=*/true); + installPublisherForwarderCallbackChain(ftn, publisherFwd, /*removeOnEmpty=*/true); } installChannelSubscriber( *cbs.channelCb, - *attach.publisherFwd, + *publisherFwd, subscriberExec, forward, crossExecFilter ); + if (!sr.firstSetup) { + // Read after install: largest reflects everything published so far, and the + // channel sub forwards strictly newer objects, so the OK boundary is exact. + attach.seedLargest = publisherFwd->largest(); + attach.seedExtensions = publisherFwd->extensions(); + } + if (sr.firstSetup) { auto& setup = *sr.firstSetup; // Passive relay chain (top-N/termination/cache): forward=true so it observes every // object, passive=true so it doesn't count as a forwarding subscriber or in the // onEmpty quorum (the publisher's onEmpty still fires when the last real sub leaves). relayChainFilter = std::make_shared(relayExec_, nullptr); - attach.publisherFwd->addChannelSubscriber( + publisherFwd->addChannelSubscriber( relayExec_, /*forward=*/true, relayChainFilter, @@ -1844,7 +1932,7 @@ folly::coro::Task MoqxRelay::attachNewLocalForwa setup.upstreamSession, std::move(setup.upstreamSubReq), std::move(setup.upstreamConsumer), - attach.publisherFwd, + publisherFwd, setup.clientRequestID ); } @@ -1853,7 +1941,7 @@ folly::coro::Task MoqxRelay::attachNewLocalForwa // Back on relayExec_. if (!sr.firstSetup) { - co_return attach; // subsequent subscriber: wired to the live publisher, done + co_return attach; // subsequent: wired to the live publisher (or attach.error on regrab miss) } if (upstreamResult->hasError()) { @@ -1893,6 +1981,8 @@ folly::coro::Task MoqxRelay::attachNewLocalForwa attach.error = folly::makeUnexpected(std::move(*err)); co_return attach; } + // tlForwarders_ on the publisher exec owns the forwarder now; relayExec_ must not retain it. + registry_.clearForwarder(ftn); attach.upstreamOk = std::move(upstreamOk); co_return attach; } @@ -1954,7 +2044,9 @@ MoqxRelay::joinOrPrepareUpstreamSubscription(SubscribeRequest subReq) { ); auto upstreamView = registry_.getUpstreamView(ftn); auto* publisherExec = upstreamView ? upstreamView->publisherExec : nullptr; - co_return StatefulSubscribeResult{sub.forwarder, publisherExec, std::nullopt, std::nullopt}; + // No forwarder: the registry no longer holds it (LF mode owns it in tlForwarders_ on + // the publisher exec). attachNewLocalForwarderOnRelayExec resolves it in its pub sortie. + co_return StatefulSubscribeResult{nullptr, publisherExec, std::nullopt, std::nullopt}; } } @@ -2070,6 +2162,14 @@ folly::coro::Task MoqxRelay::subscribeFromSubscriber // Seed the subscriber snapshot so a post-SUBSCRIBE_OK joining fetch resolves. sub->updateLargest(*attach.upstreamOk->largest); } + } else { + // Subsequent subscriber, first on this iothread: seed from the live publisher + // forwarder so the OK carries the established largest. + localFwd->setExtensions(attach.seedExtensions); + if (attach.seedLargest) { + localFwd->updateLargest(attach.seedLargest->group, attach.seedLargest->object); + sub->updateLargest(*attach.seedLargest); + } } replayPendingFowarderEvents(localFwd.get(), attach.finalCallback, *pendingCb, forward); localFwd->tryProcessNewGroupRequest(subReq.params); @@ -2573,18 +2673,23 @@ void MoqxRelay::onTrackSelected( return; } - auto trackForwarder = registry_.getForwarder(ftn); - if (!trackForwarder) { + auto upstreamView = registry_.getUpstreamView(ftn); + if (!upstreamView) { XLOG(DBG4) << "onTrackSelected: no subscription/forwarder for " << ftn; return; } - - auto upstreamView = registry_.getUpstreamView(ftn); - XCHECK(!relayExec_ || (upstreamView && upstreamView->publisherExec)) + XCHECK(!relayExec_ || upstreamView->publisherExec) << "onTrackSelected: relayExec set but no publisherExec for " << ftn; auto* publisherExec = relayExec_ ? upstreamView->publisherExec : nullptr; // TRACK_FILTER subscribers are unpinned so onTrackEvicted can remove them. - addSubscriberAndPublish(session, trackForwarder, forward, /*pinned=*/false, publisherExec); + addSubscriberAndPublish( + session, + ftn, + upstreamView->forwarder, + forward, + /*pinned=*/false, + publisherExec + ); } void MoqxRelay::onTrackEvicted(const FullTrackName& ftn, std::shared_ptr session) { @@ -2641,13 +2746,16 @@ void MoqxRelay::dumpState(RelayStateVisitor& visitor) const { RelayStateVisitor::SubscriptionInfo info{ .ftn = e.ftn, .isPublish = e.isPublish, - .subscribers = e.forwarder->subscriberCount(), - .forwardingSubscribers = e.forwarder->numForwardingSubscribers(), - .largest = e.forwarder->largest(), - .totalGroupsReceived = e.forwarder->totalGroupsReceived(), - .totalObjectsReceived = e.forwarder->totalObjectsReceived(), .sourceAddress = sourceAddr, }; + // LF mode clears the registry forwarder (owned by tlForwarders_); stats unavailable here. + if (e.forwarder) { + info.subscribers = e.forwarder->subscriberCount(); + info.forwardingSubscribers = e.forwarder->numForwardingSubscribers(); + info.largest = e.forwarder->largest(); + info.totalGroupsReceived = e.forwarder->totalGroupsReceived(); + info.totalObjectsReceived = e.forwarder->totalObjectsReceived(); + } visitor.onSubscription(info); }); visitor.onSubscriptionsEnd(); diff --git a/src/MoqxRelay.h b/src/MoqxRelay.h index 2ebbd108..668527a3 100644 --- a/src/MoqxRelay.h +++ b/src/MoqxRelay.h @@ -330,6 +330,7 @@ class MoqxRelay : public moxygen::Publisher, bool addSubscriberAndPublish( std::shared_ptr subscriberSession, + moxygen::FullTrackName ftn, std::shared_ptr forwarder, bool forward, bool pinned, @@ -338,7 +339,7 @@ class MoqxRelay : public moxygen::Publisher, folly::coro::Task addSubscriberAndPublishViaLocalForwarder( std::shared_ptr subscriberSession, - std::shared_ptr publisherFwd, + moxygen::FullTrackName ftn, folly::Executor* publisherExec, bool forward, bool pinned @@ -371,7 +372,8 @@ class MoqxRelay : public moxygen::Publisher, std::shared_ptr handle, std::shared_ptr session, std::shared_ptr publisherFwd, - std::shared_ptr relayChainFilter + std::shared_ptr relayChainFilter, + std::shared_ptr displacedLocalFwd ); // TRACK_FILTER support @@ -477,11 +479,16 @@ class MoqxRelay : public moxygen::Publisher, // chain filter is not exposed (setDownstream/teardown happen inside attach); the tail // needs only ownsRelayChain to gate the sawOnEmpty teardown. struct PublisherAttachment { - std::shared_ptr publisherFwd; + // weak teardown identity pin; must not keep the forwarder alive off the pub exec. + std::weak_ptr publisherFwd; folly::Executor* publisherExec{nullptr}; bool ownsRelayChain{false}; // firstSetup path installed the passive relay chain std::shared_ptr finalCallback; std::optional upstreamOk; + // Subsequent-subscriber seed (set only when !ownsRelayChain): the live publisher + // forwarder's largest/extensions, read on the publisher exec. + std::optional seedLargest; + moxygen::Extensions seedExtensions; std::optional error; // set => bail }; @@ -550,7 +557,8 @@ class MoqxRelay : public moxygen::Publisher, moxygen::PublishRequest pub, std::shared_ptr handle, std::shared_ptr session, - std::shared_ptr forwarder = nullptr + std::shared_ptr forwarder = nullptr, + std::shared_ptr displacedLocalFwd = nullptr ); std::shared_ptr ownedRelayExec_; diff --git a/src/SubscriptionRegistry.cpp b/src/SubscriptionRegistry.cpp index 8933b655..fcad1dd9 100644 --- a/src/SubscriptionRegistry.cpp +++ b/src/SubscriptionRegistry.cpp @@ -164,6 +164,15 @@ SubscriptionRegistry::getForwarder(const moxygen::FullTrackName& ftn) const { return it != subscriptions_.end() ? it->second.forwarder : nullptr; } +std::optional> +SubscriptionRegistry::getForwarderIfExists(const moxygen::FullTrackName& ftn) const { + auto it = subscriptions_.find(ftn); + if (it == subscriptions_.end()) { + return std::nullopt; + } + return it->second.forwarder; +} + std::optional SubscriptionRegistry::getTopNView(const moxygen::FullTrackName& ftn) const { auto it = subscriptions_.find(ftn); @@ -222,6 +231,13 @@ void SubscriptionRegistry::remove(const moxygen::FullTrackName& ftn) { subscriptions_.erase(ftn); } +void SubscriptionRegistry::clearForwarder(const moxygen::FullTrackName& ftn) { + auto it = subscriptions_.find(ftn); + if (it != subscriptions_.end()) { + it->second.forwarder.reset(); + } +} + void SubscriptionRegistry::removeIf( folly::FunctionRef predicate ) { diff --git a/src/SubscriptionRegistry.h b/src/SubscriptionRegistry.h index 3a3c3fdb..b6203e23 100644 --- a/src/SubscriptionRegistry.h +++ b/src/SubscriptionRegistry.h @@ -121,6 +121,8 @@ class SubscriptionRegistry { bool exists(const moxygen::FullTrackName& ftn) const; std::shared_ptr getForwarder(const moxygen::FullTrackName& ftn) const; + std::optional> + getForwarderIfExists(const moxygen::FullTrackName& ftn) const; struct TopNView { std::shared_ptr forwarder; @@ -160,6 +162,9 @@ class SubscriptionRegistry { // a publisher-terminated entry's forwarder goes empty. void remove(const moxygen::FullTrackName& ftn); + // LF mode: drop the registry's forwarder ref once tlForwarders_ owns it on the publisher exec. + void clearForwarder(const moxygen::FullTrackName& ftn); + // === Iteration === struct EntryView { diff --git a/test/MoqxRelaySubscribeTests.cpp b/test/MoqxRelaySubscribeTests.cpp index 5f150ce2..ffaa4e9a 100644 --- a/test/MoqxRelaySubscribeTests.cpp +++ b/test/MoqxRelaySubscribeTests.cpp @@ -406,4 +406,117 @@ TEST_P(MoQRelayTest, SubsequentSubscriberFailsWhenUpstreamSubscribeFails) { driveIfMultiThread(); } +// Cross-thread sibling of SubsequentSubscriberWaitsForUpstreamLargestSeeding: sub2 on its own +// OS thread is gated by the registry awaitSubsequent future (not the per-thread readiness gate). +TEST_P(MoQRelayTest, CrossThreadSubsequentSubscriberSeedingRace) { + auto publisherSession = createMockSession(); // upstream/publisher on exec_ + auto subSession1 = createMockSession(); // first subscriber on exec_ + + // Second subscriber on a dedicated OS thread => distinct thread-local tlForwarders_. + folly::ScopedEventBaseThread subThread("sub2-thread"); + auto subExec = std::make_shared(subThread.getEventBase()); + auto subSession2 = std::make_shared>(subExec); + ON_CALL(*subSession2, getNegotiatedVersion()) + .WillByDefault(Return(std::optional(kVersionDraftCurrent))); + getOrCreateMockState(subSession2); + + doPublishNamespace(publisherSession, kTestNamespace); + + const AbsoluteLocation kLargest{3, 0}; + SubscribeOk upstreamOk; + upstreamOk.requestID = RequestID(1); + upstreamOk.trackAlias = TrackAlias(1); + upstreamOk.expires = std::chrono::milliseconds(0); + upstreamOk.groupOrder = GroupOrder::OldestFirst; + upstreamOk.largest = kLargest; + + // Hold the upstream SUBSCRIBE in flight so sub2 races in while sub1's largest is unseeded. + folly::coro::Baton upstreamGate; + std::atomic upstreamSubscribeCalled{false}; + EXPECT_CALL(*publisherSession, subscribe(_, _)) + .WillOnce( + [&](const SubscribeRequest&, + std::shared_ptr) -> folly::coro::Task { + upstreamSubscribeCalled.store(true); + co_await upstreamGate; + auto handle = std::make_shared>(upstreamOk); + co_return folly::Expected, SubscribeError>(handle); + } + ); + + // The cross-exec sortie runs on exec_; drive it and flush the sub2 thread each iteration. + auto pumpExec = [&](auto pred) { + for (int i = 0; i < 2000 && !pred(); ++i) { + exec_->drive(); + subThread.getEventBase()->runInEventBaseThreadAndWait([] {}); + } + return pred(); + }; + + auto launchSubscribe = [&](std::shared_ptr session, + folly::Executor* startExec, + RequestID requestID, + std::shared_ptr> out, + std::atomic* done) { + withSessionContext(session, [&]() { + SubscribeRequest sub; + sub.fullTrackName = kTestTrackName; + sub.requestID = requestID; + sub.locType = LocationType::LargestObject; + auto task = publisherInterface()->subscribe(std::move(sub), createMockConsumer()); + co_withExecutor( + startExec, + folly::coro::co_invoke( + [t = std::move(task), out, done]() mutable -> folly::coro::Task { + *out = co_await std::move(t); + if (done) { + done->store(true); + } + } + ) + ).start(); + }); + }; + + // First subscriber on exec_: becomes firstSetup, suspends in the gated upstream SUBSCRIBE. + auto firstResult = std::make_shared>(); + launchSubscribe( + subSession1, + static_cast(exec_.get()), + RequestID(0), + firstResult, + nullptr + ); + ASSERT_TRUE(pumpExec([&] { return upstreamSubscribeCalled.load(); })) + << "relay should issue an upstream subscribe and suspend in it"; + + // Second subscriber on its own thread, while the first's seeding is still pending. + std::atomic sub2Done{false}; + auto secondResult = std::make_shared>(); + launchSubscribe(subSession2, subExec.get(), RequestID(2), secondResult, &sub2Done); + + // sub2 must stay blocked on awaitSubsequent until the upstream OK seeds largest; resolving + // early would return an empty largest a client reads as a track restart. + EXPECT_FALSE(pumpExec([&] { return sub2Done.load(); })) + << "cross-thread subsequent subscriber resolved before the upstream OK seeded largest"; + + upstreamGate.post(); + ASSERT_TRUE(pumpExec([&] { return firstResult->has_value() && sub2Done.load(); })); + + ASSERT_TRUE(firstResult->value().hasValue()); + EXPECT_EQ(firstResult->value().value()->subscribeOk().largest, kLargest); + ASSERT_TRUE(secondResult->value().hasValue()); + // Post-OK the established largest must hold regardless of when sub2 resolved. + EXPECT_EQ(secondResult->value().value()->subscribeOk().largest, kLargest); + + getOrCreateMockState(subSession1)->subscribeHandles.push_back(firstResult->value().value()); + getOrCreateMockState(subSession2)->subscribeHandles.push_back(secondResult->value().value()); + + removeSession(publisherSession); + removeSession(subSession1); + removeSession(subSession2); + driveIfMultiThread(); + subThread.getEventBase()->runInEventBaseThreadAndWait([] {}); +} + } // namespace moxygen::test