From d063243d0bffd955d712ed5204394b079d8503f2 Mon Sep 17 00:00:00 2001 From: Trevor Clinkenbeard Date: Mon, 10 Aug 2026 05:41:00 -0700 Subject: [PATCH 1/5] Convert ActorCollection to standard coroutines --- flow/ActorCollection.actor.cpp | 275 --------------- flow/ActorCollection.cpp | 587 +++++++++++++++++++++++++++++++++ 2 files changed, 587 insertions(+), 275 deletions(-) delete mode 100644 flow/ActorCollection.actor.cpp create mode 100644 flow/ActorCollection.cpp diff --git a/flow/ActorCollection.actor.cpp b/flow/ActorCollection.actor.cpp deleted file mode 100644 index 8e19bd7f557..00000000000 --- a/flow/ActorCollection.actor.cpp +++ /dev/null @@ -1,275 +0,0 @@ -/* - * ActorCollection.actor.cpp - * - * This source file is part of the FoundationDB open source project - * - * Copyright 2013-2026 Apple Inc. and the FoundationDB project authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "flow/ActorCollection.h" -#include "flow/IndexedSet.h" -#include "flow/UnitTest.h" -#include -#include "flow/actorcompiler.h" // This must be the last #include. - -class Runner final : public boost::intrusive::list_base_hook<>, - public Callback, - public FastAllocated, - NonCopyable { -public: - Runner(PromiseStream complete, PromiseStream errors) - : complete(std::move(complete)), errors(std::move(errors)) {} - - ~Runner() { detach(); } - - void start(Future task) { - if (!task.isReady()) { - registered = true; - task.addCallbackAndClear(this); - return; - } - if (task.isError()) { - error(task.getError()); - } else { - fire(Void()); - } - } - - void fire(Void const&) override { -#ifdef ENABLE_SAMPLING - LineageReference callbackLineage = lineage; - LineageScope scope(&callbackLineage); -#endif - auto output = complete; - detach(); - output.send(this); - } - - void error(Error e) override { -#ifdef ENABLE_SAMPLING - LineageReference callbackLineage = lineage; - LineageScope scope(&callbackLineage); -#endif - auto output = errors; - detach(); - if (e.code() != error_code_actor_cancelled) { - output.send(e); - } - } - -private: - void detach() { - if (registered) { - registered = false; - Callback::remove(); - } - } - - PromiseStream complete; - PromiseStream errors; - bool registered = false; -#ifdef ENABLE_SAMPLING - LineageReference lineage = *currentLineage; -#endif -}; - -// An intrusive list of Runners, which are FastAllocated. -using RunnerList = boost::intrusive::list>; - -// The runners list in the ActorCollection must be destroyed when the actor is destructed rather -// than before returning or throwing -struct RunnerListDestroyer : NonCopyable { - explicit RunnerListDestroyer(RunnerList* list) : list(list) {} - - ~RunnerListDestroyer() { - list->clear_and_dispose([](Runner* r) { delete r; }); - } - - RunnerList* list; -}; - -ACTOR Future actorCollection(FutureStream> addActor, - int* pCount, - double* lastChangeTime, - double* idleTime, - double* allTime, - bool returnWhenEmptied) { - state RunnerList runners; - state RunnerListDestroyer runnersDestroyer(&runners); - state PromiseStream complete; - state PromiseStream errors; - state int count = 0; - if (!pCount) - pCount = &count; - - loop choose { - when(Future f = waitNext(addActor)) { - auto runner = new Runner(complete, errors); - runners.insert(runners.end(), *runner); - runner->start(std::move(f)); - - ++*pCount; - if (*pCount == 1 && lastChangeTime && idleTime && allTime) { - double currentTime = now(); - *idleTime += currentTime - *lastChangeTime; - *allTime += currentTime - *lastChangeTime; - *lastChangeTime = currentTime; - } - } - when(Runner* runner = waitNext(complete.getFuture())) { - if (!--*pCount) { - if (lastChangeTime && idleTime && allTime) { - double currentTime = now(); - *allTime += currentTime - *lastChangeTime; - *lastChangeTime = currentTime; - } - if (returnWhenEmptied) - return Void(); - } - // If we didn't return then the entire list wasn't destroyed so erase/destroy runner - runners.erase_and_dispose(runners.iterator_to(*runner), [](Runner* r) { delete r; }); - } - when(Error e = waitNext(errors.getFuture())) { - throw e; - } - } -} - -template -struct Traceable> { - static constexpr bool value = Traceable::value && Traceable::value; - static std::string toString(const std::pair& p) { - auto tStr = Traceable::toString(p.first); - auto uStr = Traceable::toString(p.second); - std::string result(tStr.size() + uStr.size() + 3, 'x'); - std::copy(tStr.begin(), tStr.end(), result.begin()); - auto iter = result.begin() + tStr.size(); - *(iter++) = ' '; - *(iter++) = '-'; - *(iter++) = ' '; - std::copy(uStr.begin(), uStr.end(), iter); - return result; - } -}; - -void forceLinkActorCollectionTests() {} - -// The above implementation relies on the behavior that fulfilling a promise -// that another when clause in the same choose block is waiting on is not fired synchronously. -TEST_CASE("/flow/actorCollection/chooseWhen") { - state Promise promise; - choose { - when(wait(delay(0))) { - promise.send(Void()); - } - when(wait(promise.getFuture())) { - // Should be cancelled, since another when clause in this choose block has executed - ASSERT(false); - } - } - return Void(); -} - -ACTOR Future failIfNotCancelled() { - wait(delay(0)); - ASSERT(false); - return Void(); -} - -// test contract that actors are cancelled when the actor collection is cleared -TEST_CASE("/flow/actorCollection/testCancel") { - state ActorCollection actorCollection(false); - int actors = deterministicRandom()->randomInt(1, 1000); - for (int i = 0; i < actors; i++) { - actorCollection.add(failIfNotCancelled()); - } - actorCollection.clear(false); - wait(delay(0)); - return Void(); -} - -Future failedActor() { - return operation_failed(); -} - -TEST_CASE("/flow/actorCollection/testReady") { - state ActorCollection actorCollection(true); - actorCollection.add(Void()); - wait(actorCollection.getResult()); - return Void(); -} - -TEST_CASE("/flow/actorCollection/testReadyWhilePending") { - state ActorCollection actorCollection(true); - state Promise pending; - actorCollection.add(pending.getFuture()); - actorCollection.add(Void()); - wait(delay(0)); - ASSERT(!actorCollection.getResult().isReady()); - pending.send(Void()); - wait(actorCollection.getResult()); - return Void(); -} - -TEST_CASE("/flow/actorCollection/testReadyError") { - state ActorCollection actorCollection(false); - actorCollection.add(failedActor()); - try { - wait(actorCollection.getResult()); - ASSERT(false); - } catch (Error& e) { - ASSERT_EQ(e.code(), error_code_operation_failed); - } - return Void(); -} - -TEST_CASE("/flow/actorCollection/testPendingErrorCancels") { - state ActorCollection actorCollection(false); - state Promise pending; - actorCollection.add(failIfNotCancelled()); - actorCollection.add(pending.getFuture()); - pending.sendError(operation_failed()); - try { - wait(actorCollection.getResult()); - ASSERT(false); - } catch (Error& e) { - ASSERT_EQ(e.code(), error_code_operation_failed); - } - wait(delay(0)); - return Void(); -} - -// test contract that even if the actor collection has stopped and new actors are added to the promise stream, they are -// all cancelled when resetting actor -TEST_CASE("/flow/actorCollection/testCancelPromiseStream") { - state ActorCollection actorCollection(false); - int actors = deterministicRandom()->randomInt(1, 500); - for (int i = 0; i < actors; i++) { - actorCollection.add(failIfNotCancelled()); - } - // this actor should cause the actorCollection actor to exit, meaning the new futures just build up in the promise - // stream - actorCollection.add(failedActor()); - for (int i = 0; i < actors; i++) { - actorCollection.add(failIfNotCancelled()); - } - // Instead of doing actorCollection.clear(false) we reinitialize to also clear the promise stream. Otherwise on - // resetting the actor collection actor, the new actors will be pulled from the promise stream into the new instance - // Note that this test fails on the assert in failIfNotCancelled() when this is replaced with - // actorCollection.clear(false). - actorCollection = ActorCollection(false); - wait(delay(0)); - return Void(); -} diff --git a/flow/ActorCollection.cpp b/flow/ActorCollection.cpp new file mode 100644 index 00000000000..a7e6260613d --- /dev/null +++ b/flow/ActorCollection.cpp @@ -0,0 +1,587 @@ +/* + * ActorCollection.cpp + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2013-2026 Apple Inc. and the FoundationDB project authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "flow/ActorCollection.h" +#include "flow/CoroUtils.h" +#include "flow/IndexedSet.h" +#include "flow/UnitTest.h" +#include +#include + +class ActorCollectionRuntime; + +class Runner final : public boost::intrusive::list_base_hook<>, + public Callback, + public FastAllocated, + NonCopyable { +public: + explicit Runner(ActorCollectionRuntime* owner) : owner(owner) {} + + ~Runner() { detach(); } + + void start(Future task); + void fire(Void const&) override; + void error(Error e) override; + +private: + void detach() { + if (registered) { + registered = false; + Callback::remove(); + } + } + + ActorCollectionRuntime* owner; + Runner* nextCompleted = nullptr; + bool registered = false; +#ifdef ENABLE_SAMPLING + LineageReference lineage = *currentLineage; +#endif + + friend class ActorCollectionRuntime; +}; + +// An intrusive list of Runners, which are FastAllocated. +using RunnerList = boost::intrusive::list>; + +class RunnerListDestroyer final : NonCopyable { +public: + explicit RunnerListDestroyer(RunnerList* list) : list(list) {} + + ~RunnerListDestroyer() { + list->clear_and_dispose([](Runner* r) { delete r; }); + } + +private: + RunnerList* list; +}; + +class ActorCollectionRuntime final : NonCopyable { + class AddActorCallback final : public SingleCallback> { + public: + explicit AddActorCallback(ActorCollectionRuntime* owner) : owner(owner) {} + + void fire(Future const& actor) override; + void fire(Future&& actor) override; + void error(Error e) override; + + private: + ActorCollectionRuntime* owner; + }; + +public: + ActorCollectionRuntime(FutureStream> addActor, + int* pCount, + double* lastChangeTime, + double* idleTime, + double* allTime, + bool returnWhenEmptied) + : addActor(std::move(addActor)), pCount(pCount), lastChangeTime(lastChangeTime), idleTime(idleTime), + allTime(allTime), returnWhenEmptied(returnWhenEmptied), runnersDestroyer(&runners), addActorCallback(this) { + if (!this->pCount) { + this->pCount = &count; + } + } + + ~ActorCollectionRuntime() { + if (addCallbackRegistered) { + addActorCallback.remove(); + addCallbackRegistered = false; + } + } + + Future getResult() { return done.getFuture(); } + void start() { drain(); } + +private: + void onAdded(Future actor) { + if (finished) { + return; + } + if (draining) { + pendingAdds.emplace_back(std::move(actor)); + return; + } + + draining = true; + handleAdded(std::move(actor)); + draining = false; + drain(); + } + + void onAddError(Error e) { + if (!finished) { + requestError(e); + drain(); + } + } + + void onCompleted(Runner* runner) { + if (finished) { + return; + } + if (completedTail) { + completedTail->nextCompleted = runner; + } else { + completedHead = runner; + } + completedTail = runner; + drain(); + } + + void onError(Error e) { + if (finished) { + return; + } + if (!failure.present()) { + failure = e; + } + drain(); + } + + void incrementCount() { + ++*pCount; + if (*pCount == 1 && lastChangeTime && idleTime && allTime) { + double currentTime = now(); + *idleTime += currentTime - *lastChangeTime; + *allTime += currentTime - *lastChangeTime; + *lastChangeTime = currentTime; + } + } + + void decrementCount() { + if (!--*pCount && lastChangeTime && idleTime && allTime) { + double currentTime = now(); + *allTime += currentTime - *lastChangeTime; + *lastChangeTime = currentTime; + } + } + + void addRunner(Future actor) { + auto runner = runners.insert(runners.end(), *new Runner(this)); + runner->start(std::move(actor)); + incrementCount(); + } + + void handleAdded(Future actor) { + // Completing inline must not outrun an earlier queued addition. + if (!runners.empty() || !actor.isReady() || addActor.isReady() || !pendingAdds.empty()) { + addRunner(std::move(actor)); + return; + } + + if (actor.isError()) { + Error e = actor.getError(); + if (e.code() == error_code_actor_cancelled) { + addRunner(std::move(actor)); + return; + } + incrementCount(); + requestError(e); + return; + } + + incrementCount(); + decrementCount(); + if (!*pCount && returnWhenEmptied) { + terminalRequested = true; + } + } + + void handleCompleted(Runner* runner) { + decrementCount(); + if (!*pCount && returnWhenEmptied) { + terminalRequested = true; + return; + } + runners.erase_and_dispose(runners.iterator_to(*runner), [](Runner* runner) { delete runner; }); + } + + Runner* popCompleted() { + Runner* runner = completedHead; + completedHead = runner->nextCompleted; + if (!completedHead) { + completedTail = nullptr; + } + runner->nextCompleted = nullptr; + return runner; + } + + void armAddCallback() { + if (!addCallbackRegistered) { + addCallbackRegistered = true; + auto stream = addActor; + stream.addCallbackAndClear(&addActorCallback); + } + } + + void requestError(Error e) { + if (!terminalError.present()) { + terminalError = e; + } + terminalRequested = true; + } + + void deliverTerminal() { + // Sending can synchronously destroy this runtime, so retain the promise before notifying waiters. + Promise terminal = done; + Optional error = terminalError; + finished = true; + if (addCallbackRegistered) { + addActorCallback.remove(); + addCallbackRegistered = false; + } + if (error.present()) { + terminal.sendError(error.get()); + } else { + terminal.send(Void()); + } + } + + void drain() { + if (draining || finished) { + return; + } + draining = true; + + while (!terminalRequested) { + if (!pendingAdds.empty()) { + Future actor = std::move(pendingAdds[nextPendingAdd++]); + if (nextPendingAdd == pendingAdds.size()) { + pendingAdds.clear(); + nextPendingAdd = 0; + } + handleAdded(std::move(actor)); + continue; + } + if (addActor.isReady()) { + if (addActor.isError()) { + requestError(addActor.getError()); + } else { + handleAdded(addActor.pop()); + } + continue; + } + if (completedHead) { + handleCompleted(popCompleted()); + continue; + } + if (failure.present()) { + requestError(failure.get()); + continue; + } + armAddCallback(); + draining = false; + return; + } + + deliverTerminal(); + } + + FutureStream> addActor; + int* pCount; + double* lastChangeTime; + double* idleTime; + double* allTime; + bool returnWhenEmptied; + int count = 0; + RunnerList runners; + RunnerListDestroyer runnersDestroyer; + Promise done; + AddActorCallback addActorCallback; + Runner* completedHead = nullptr; + Runner* completedTail = nullptr; + std::vector> pendingAdds; + size_t nextPendingAdd = 0; + Optional failure; + Optional terminalError; + bool addCallbackRegistered = false; + bool draining = false; + bool terminalRequested = false; + bool finished = false; +#ifdef ENABLE_SAMPLING + LineageReference lineage = *currentLineage; +#endif + + friend class Runner; +}; + +void Runner::start(Future task) { + if (!task.isReady()) { + registered = true; + task.addCallbackAndClear(this); + return; + } + if (task.isError()) { + error(task.getError()); + } else { + fire(Void()); + } +} + +void Runner::fire(Void const&) { +#ifdef ENABLE_SAMPLING + LineageReference callbackLineage = lineage; + LineageScope scope(&callbackLineage); +#endif + ActorCollectionRuntime* runtime = owner; + detach(); + runtime->onCompleted(this); +} + +void Runner::error(Error e) { +#ifdef ENABLE_SAMPLING + LineageReference callbackLineage = lineage; + LineageScope scope(&callbackLineage); +#endif + ActorCollectionRuntime* runtime = owner; + detach(); + if (e.code() != error_code_actor_cancelled) { + runtime->onError(e); + } +} + +void ActorCollectionRuntime::AddActorCallback::fire(Future const& actor) { +#ifdef ENABLE_SAMPLING + LineageReference callbackLineage = owner->lineage; + LineageScope scope(&callbackLineage); +#endif + owner->onAdded(actor); +} + +void ActorCollectionRuntime::AddActorCallback::fire(Future&& actor) { +#ifdef ENABLE_SAMPLING + LineageReference callbackLineage = owner->lineage; + LineageScope scope(&callbackLineage); +#endif + owner->onAdded(std::move(actor)); +} + +void ActorCollectionRuntime::AddActorCallback::error(Error e) { +#ifdef ENABLE_SAMPLING + LineageReference callbackLineage = owner->lineage; + LineageScope scope(&callbackLineage); +#endif + owner->onAddError(e); +} + +static Future actorCollectionImpl(FutureStream> addActor, + int* pCount, + double* lastChangeTime, + double* idleTime, + double* allTime, + bool returnWhenEmptied, + NoThrowOnCancel = {}) { + ActorCollectionRuntime runtime(std::move(addActor), pCount, lastChangeTime, idleTime, allTime, returnWhenEmptied); + Future result = runtime.getResult(); + runtime.start(); + co_await result; +} + +static Future actorCollectionUntilEmpty(FutureStream> addActor, + int* pCount, + double* lastChangeTime, + double* idleTime, + double* allTime) { + ActorCollectionRuntime runtime(std::move(addActor), pCount, lastChangeTime, idleTime, allTime, true); + Future result = runtime.getResult(); + runtime.start(); + co_await result; +} + +Future actorCollection(FutureStream> const& addActor, + int* const& pCount, + double* const& lastChangeTime, + double* const& idleTime, + double* const& allTime, + bool const& returnWhenEmptied) { + if (returnWhenEmptied) { + return actorCollectionUntilEmpty(addActor, pCount, lastChangeTime, idleTime, allTime); + } + return actorCollectionImpl(addActor, pCount, lastChangeTime, idleTime, allTime, returnWhenEmptied); +} + +template +struct Traceable> { + static constexpr bool value = Traceable::value && Traceable::value; + static std::string toString(const std::pair& p) { + auto tStr = Traceable::toString(p.first); + auto uStr = Traceable::toString(p.second); + std::string result(tStr.size() + uStr.size() + 3, 'x'); + std::copy(tStr.begin(), tStr.end(), result.begin()); + auto iter = result.begin() + tStr.size(); + *(iter++) = ' '; + *(iter++) = '-'; + *(iter++) = ' '; + std::copy(uStr.begin(), uStr.end(), iter); + return result; + } +}; + +void forceLinkActorCollectionTests() {} + +TEST_CASE("/flow/actorCollection/chooseWhen") { + Promise promise; + auto result = co_await race(delay(0), promise.getFuture()); + ASSERT_EQ(result.index(), 0); + promise.send(Void()); +} + +Future failIfNotCancelled() { + co_await delay(0); + ASSERT(false); +} + +static Future recordActorCancellation(Future pending, std::vector* cancelled, int index) { + try { + co_await pending; + } catch (Error& e) { + if (e.code() == error_code_actor_cancelled) { + cancelled->push_back(index); + } + throw; + } +} + +// test contract that actors are cancelled when the actor collection is cleared +TEST_CASE("/flow/actorCollection/testCancel") { + ActorCollection actorCollection(false); + int actors = deterministicRandom()->randomInt(1, 1000); + for (int i = 0; i < actors; i++) { + actorCollection.add(failIfNotCancelled()); + } + actorCollection.clear(false); + co_await delay(0); +} + +TEST_CASE("/flow/actorCollection/testCancelOrder") { + constexpr int actorCount = 4; + for (bool returnWhenEmptied : { false, true }) { + ActorCollection collection(returnWhenEmptied); + std::vector> pending(actorCount); + std::vector cancelled; + for (int index = 0; index < actorCount; ++index) { + collection.add(recordActorCancellation(pending[index].getFuture(), &cancelled, index)); + } + collection.clear(returnWhenEmptied); + ASSERT_EQ(cancelled.size(), pending.size()); + for (int index = 0; index < actorCount; ++index) { + ASSERT_EQ(cancelled[index], index); + } + } + return Void(); +} + +Future failedActor() { + return operation_failed(); +} + +TEST_CASE("/flow/actorCollection/testReady") { + ActorCollection actorCollection(true); + actorCollection.add(Void()); + co_await actorCollection.getResult(); +} + +TEST_CASE("/flow/actorCollection/testReadyWhilePending") { + ActorCollection actorCollection(true); + Promise pending; + actorCollection.add(pending.getFuture()); + actorCollection.add(Void()); + co_await delay(0); + ASSERT(!actorCollection.getResult().isReady()); + pending.send(Void()); + co_await actorCollection.getResult(); +} + +TEST_CASE("/flow/actorCollection/testQueuedReadyWhilePending") { + PromiseStream> addActor; + Promise pending; + int count = 0; + addActor.send(Void()); + addActor.send(pending.getFuture()); + + Future collection = actorCollection(addActor.getFuture(), &count, nullptr, nullptr, nullptr, true); + ASSERT_EQ(count, 1); + ASSERT(!collection.isReady()); + + pending.send(Void()); + co_await collection; + ASSERT_EQ(count, 0); +} + +TEST_CASE("/flow/actorCollection/testReadyError") { + ActorCollection actorCollection(false); + actorCollection.add(failedActor()); + try { + co_await actorCollection.getResult(); + ASSERT(false); + } catch (Error& e) { + ASSERT_EQ(e.code(), error_code_operation_failed); + } +} + +TEST_CASE("/flow/actorCollection/testAddStreamError") { + PromiseStream> addActor; + Future collection = actorCollection(addActor.getFuture()); + addActor.sendError(operation_failed()); + try { + co_await collection; + ASSERT(false); + } catch (Error& e) { + ASSERT_EQ(e.code(), error_code_operation_failed); + } +} + +TEST_CASE("/flow/actorCollection/testPendingErrorCancels") { + ActorCollection actorCollection(false); + Promise pending; + actorCollection.add(failIfNotCancelled()); + actorCollection.add(pending.getFuture()); + pending.sendError(operation_failed()); + try { + co_await actorCollection.getResult(); + ASSERT(false); + } catch (Error& e) { + ASSERT_EQ(e.code(), error_code_operation_failed); + } + co_await delay(0); +} + +// test contract that even if the actor collection has stopped and new actors are added to the promise stream, they are +// all cancelled when resetting actor +TEST_CASE("/flow/actorCollection/testCancelPromiseStream") { + ActorCollection actorCollection(false); + int actors = deterministicRandom()->randomInt(1, 500); + for (int i = 0; i < actors; i++) { + actorCollection.add(failIfNotCancelled()); + } + // this actor should cause the actorCollection actor to exit, meaning the new futures just build up in the promise + // stream + actorCollection.add(failedActor()); + for (int i = 0; i < actors; i++) { + actorCollection.add(failIfNotCancelled()); + } + // Instead of doing actorCollection.clear(false) we reinitialize to also clear the promise stream. Otherwise on + // resetting the actor collection actor, the new actors will be pulled from the promise stream into the new instance + // Note that this test fails on the assert in failIfNotCancelled() when this is replaced with + // actorCollection.clear(false). + actorCollection = ActorCollection(false); + co_await delay(0); +} From 86712982f88243cb68f582b9d971a9778e5f082d Mon Sep 17 00:00:00 2001 From: Trevor Clinkenbeard Date: Mon, 10 Aug 2026 13:27:03 -0700 Subject: [PATCH 2/5] Fix ActorCollection cancellation teardown --- flow/ActorCollection.cpp | 33 +++++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/flow/ActorCollection.cpp b/flow/ActorCollection.cpp index a7e6260613d..345e8689268 100644 --- a/flow/ActorCollection.cpp +++ b/flow/ActorCollection.cpp @@ -101,10 +101,12 @@ class ActorCollectionRuntime final : NonCopyable { } ~ActorCollectionRuntime() { + finished = true; if (addCallbackRegistered) { addActorCallback.remove(); addCallbackRegistered = false; } + runners.clear_and_dispose([](Runner* runner) { delete runner; }); } Future getResult() { return done.getFuture(); } @@ -460,6 +462,21 @@ static Future recordActorCancellation(Future pending, std::vector signalSiblingOnActorCancellation(Future pending, Promise sibling, bool sendError) { + try { + co_await pending; + } catch (Error& e) { + if (e.code() == error_code_actor_cancelled) { + if (sendError) { + sibling.sendError(operation_failed()); + } else { + sibling.send(Void()); + } + } + throw; + } +} + // test contract that actors are cancelled when the actor collection is cleared TEST_CASE("/flow/actorCollection/testCancel") { ActorCollection actorCollection(false); @@ -489,6 +506,22 @@ TEST_CASE("/flow/actorCollection/testCancelOrder") { return Void(); } +TEST_CASE("/flow/actorCollection/testCancelReentrantSiblingCompletion") { + for (bool returnWhenEmptied : { false, true }) { + for (bool sendError : { false, true }) { + ActorCollection collection(returnWhenEmptied); + Promise pending; + Promise sibling; + collection.add(signalSiblingOnActorCancellation(pending.getFuture(), sibling, sendError)); + collection.add(sibling.getFuture()); + collection.clear(returnWhenEmptied); + ASSERT(sibling.isSet()); + ASSERT_EQ(sibling.isError(), sendError); + } + } + return Void(); +} + Future failedActor() { return operation_failed(); } From 39ab2d20498d6ba2ac083c0df356b1a23ef8e7b3 Mon Sep 17 00:00:00 2001 From: Trevor Clinkenbeard Date: Mon, 10 Aug 2026 15:02:52 -0700 Subject: [PATCH 3/5] Fix ActorCollection reentrant cancellation and sampling lineage --- flow/ActorCollection.cpp | 132 +++++++++++++++++++++++++--- flow/CoroTests.cpp | 29 ++++++ flow/include/flow/ActorCollection.h | 6 +- flow/include/flow/CoroutinesImpl.h | 5 +- 4 files changed, 159 insertions(+), 13 deletions(-) diff --git a/flow/ActorCollection.cpp b/flow/ActorCollection.cpp index 345e8689268..44a897969da 100644 --- a/flow/ActorCollection.cpp +++ b/flow/ActorCollection.cpp @@ -23,8 +23,17 @@ #include "flow/IndexedSet.h" #include "flow/UnitTest.h" #include +#include #include +#ifdef ENABLE_SAMPLING +static LineageReference actorCollectionLineage(LineageReference const& parent) { + LineageReference lineage = parent; + lineage.setActorName("actorCollection"); + return lineage; +} +#endif + class ActorCollectionRuntime; class Runner final : public boost::intrusive::list_base_hook<>, @@ -52,7 +61,7 @@ class Runner final : public boost::intrusive::list_base_hook<>, Runner* nextCompleted = nullptr; bool registered = false; #ifdef ENABLE_SAMPLING - LineageReference lineage = *currentLineage; + LineageReference lineage = actorCollectionLineage(*currentLineage); #endif friend class ActorCollectionRuntime; @@ -110,7 +119,13 @@ class ActorCollectionRuntime final : NonCopyable { } Future getResult() { return done.getFuture(); } - void start() { drain(); } + void start() { +#ifdef ENABLE_SAMPLING + LineageReference callbackLineage = actorCollectionLineage(lineage); + LineageScope scope(&callbackLineage); +#endif + drain(); + } private: void onAdded(Future actor) { @@ -319,7 +334,7 @@ class ActorCollectionRuntime final : NonCopyable { bool terminalRequested = false; bool finished = false; #ifdef ENABLE_SAMPLING - LineageReference lineage = *currentLineage; + LineageReference lineage = actorCollectionLineage(*currentLineage); #endif friend class Runner; @@ -340,7 +355,7 @@ void Runner::start(Future task) { void Runner::fire(Void const&) { #ifdef ENABLE_SAMPLING - LineageReference callbackLineage = lineage; + LineageReference callbackLineage = actorCollectionLineage(lineage); LineageScope scope(&callbackLineage); #endif ActorCollectionRuntime* runtime = owner; @@ -350,7 +365,7 @@ void Runner::fire(Void const&) { void Runner::error(Error e) { #ifdef ENABLE_SAMPLING - LineageReference callbackLineage = lineage; + LineageReference callbackLineage = actorCollectionLineage(lineage); LineageScope scope(&callbackLineage); #endif ActorCollectionRuntime* runtime = owner; @@ -362,7 +377,7 @@ void Runner::error(Error e) { void ActorCollectionRuntime::AddActorCallback::fire(Future const& actor) { #ifdef ENABLE_SAMPLING - LineageReference callbackLineage = owner->lineage; + LineageReference callbackLineage = actorCollectionLineage(owner->lineage); LineageScope scope(&callbackLineage); #endif owner->onAdded(actor); @@ -370,7 +385,7 @@ void ActorCollectionRuntime::AddActorCallback::fire(Future const& actor) { void ActorCollectionRuntime::AddActorCallback::fire(Future&& actor) { #ifdef ENABLE_SAMPLING - LineageReference callbackLineage = owner->lineage; + LineageReference callbackLineage = actorCollectionLineage(owner->lineage); LineageScope scope(&callbackLineage); #endif owner->onAdded(std::move(actor)); @@ -378,7 +393,7 @@ void ActorCollectionRuntime::AddActorCallback::fire(Future&& actor) { void ActorCollectionRuntime::AddActorCallback::error(Error e) { #ifdef ENABLE_SAMPLING - LineageReference callbackLineage = owner->lineage; + LineageReference callbackLineage = actorCollectionLineage(owner->lineage); LineageScope scope(&callbackLineage); #endif owner->onAddError(e); @@ -441,9 +456,11 @@ void forceLinkActorCollectionTests() {} TEST_CASE("/flow/actorCollection/chooseWhen") { Promise promise; - auto result = co_await race(delay(0), promise.getFuture()); - ASSERT_EQ(result.index(), 0); - promise.send(Void()); + co_await Choose() + .When(delay(0), [&promise](Void const&) { promise.send(Void()); }) + .When(promise.getFuture(), [](Void const&) { ASSERT(false); }) + .run(); + ASSERT(promise.isSet()); } Future failIfNotCancelled() { @@ -477,6 +494,26 @@ static Future signalSiblingOnActorCancellation(Future pending, Promi } } +static Future recancelCollectionOnActorCancellation(Future pending, + ActorCollection* collection, + bool returnWhenEmptied, + bool clearCollection, + int* cancellationCount) { + try { + co_await pending; + } catch (Error& e) { + if (e.code() == error_code_actor_cancelled) { + ++*cancellationCount; + if (clearCollection) { + collection->clear(returnWhenEmptied); + } else { + collection->getResult().cancel(); + } + } + throw; + } +} + // test contract that actors are cancelled when the actor collection is cleared TEST_CASE("/flow/actorCollection/testCancel") { ActorCollection actorCollection(false); @@ -522,6 +559,79 @@ TEST_CASE("/flow/actorCollection/testCancelReentrantSiblingCompletion") { return Void(); } +TEST_CASE("/flow/actorCollection/testCancelReentrantCollection") { + for (bool returnWhenEmptied : { false, true }) { + for (bool clearCollection : { false, true }) { + ActorCollection collection(returnWhenEmptied); + Promise pending; + int cancellationCount = 0; + collection.add(recancelCollectionOnActorCancellation( + pending.getFuture(), &collection, returnWhenEmptied, clearCollection, &cancellationCount)); + collection.clear(returnWhenEmptied); + ASSERT_EQ(cancellationCount, 1); + ASSERT(!collection.getResult().isReady()); + } + } + return Void(); +} + +#ifdef ENABLE_SAMPLING +class ActorCollectionLineageObserver final : public Callback, NonCopyable { +public: + void fire(Void const&) override { observe(); } + + void error(Error e) override { + observedError = e; + observe(); + } + + void assertObserved(bool expectError) const { + ASSERT_EQ(actorName, std::string("actorCollection")); + ASSERT_EQ(observedError.present(), expectError); + } + +private: + void observe() { + actorName = currentLineage->actorName(); + Callback::remove(); + } + + std::string actorName; + Optional observedError; +}; + +TEST_CASE("/flow/actorCollection/testSamplingLineage") { + for (bool readyActor : { false, true }) { + PromiseStream> addActor; + ActorCollectionRuntime collection(addActor.getFuture(), nullptr, nullptr, nullptr, nullptr, true); + collection.start(); + + ActorCollectionLineageObserver observer; + Future result = collection.getResult(); + result.addCallbackAndClear(&observer); + + if (readyActor) { + addActor.send(Future(Void())); + } else { + Promise pending; + addActor.send(pending.getFuture()); + pending.send(Void()); + } + observer.assertObserved(false); + } + + PromiseStream> addActor; + ActorCollectionRuntime collection(addActor.getFuture(), nullptr, nullptr, nullptr, nullptr, false); + collection.start(); + ActorCollectionLineageObserver observer; + Future result = collection.getResult(); + result.addCallbackAndClear(&observer); + addActor.sendError(operation_failed()); + observer.assertObserved(true); + return Void(); +} +#endif + Future failedActor() { return operation_failed(); } diff --git a/flow/CoroTests.cpp b/flow/CoroTests.cpp index dbf61e89221..8723d7cb46e 100644 --- a/flow/CoroTests.cpp +++ b/flow/CoroTests.cpp @@ -2079,6 +2079,18 @@ Future noThrowOnCancelTest(NoThrowOnCancelRecorder& recorder, Future recorder.record(NoThrowOnCancelEvent::AfterWait); } +Future noThrowOnCancelReentrantCancelTest(Future* result, + Future signal, + int* cleanupCount, + NoThrowOnCancel = {}) { + ScopeExit cleanup([result, cleanupCount]() { + ++*cleanupCount; + result->cancel(); + }); + + co_await signal; +} + Future noThrowOnCancelValueTest(NoThrowOnCancelRecorder& recorder, Future signal, NoThrowOnCancel = {}) { recorder.record(NoThrowOnCancelEvent::Start); @@ -2769,6 +2781,23 @@ TEST_CASE("/flow/coro/noThrowOnCancel/awaitedFutureErrorRunsCatch") { return Void(); } +TEST_CASE("/flow/coro/noThrowOnCancel/reentrantCancelDuringCleanup") { + Promise signal; + Future result; + int cleanupCount = 0; + result = noThrowOnCancelReentrantCancelTest(&result, signal.getFuture(), &cleanupCount); + ASSERT(signal.getFutureReferenceCount() > 0); + + result.cancel(); + ASSERT(result.isReady() && result.isError() && result.getError().code() == error_code_actor_cancelled); + ASSERT_EQ(cleanupCount, 1); + ASSERT_EQ(signal.getFutureReferenceCount(), 0); + + result.cancel(); + ASSERT_EQ(cleanupCount, 1); + return Void(); +} + TEST_CASE("/flow/coro/noThrowOnCancel/sequentialAwaitsCancelSecond") { NoThrowOnCancelRecorder recorder; Promise firstSignal; diff --git a/flow/include/flow/ActorCollection.h b/flow/include/flow/ActorCollection.h index fae1261351c..31133674da7 100644 --- a/flow/include/flow/ActorCollection.h +++ b/flow/include/flow/ActorCollection.h @@ -74,7 +74,11 @@ class ActorCollection : NonCopyable { void add(Future a) { m_add.send(a); } Future getResult() const { return m_out; } void clear(bool returnWhenEmptied) { - m_out.cancel(); + Future previous = m_out; + previous.cancel(); + if (m_out != previous) { + return; + } m_out = actorCollection(m_add.getFuture(), nullptr, nullptr, nullptr, nullptr, returnWhenEmptied); } }; diff --git a/flow/include/flow/CoroutinesImpl.h b/flow/include/flow/CoroutinesImpl.h index 8f5f075ac90..ed108ffc553 100644 --- a/flow/include/flow/CoroutinesImpl.h +++ b/flow/include/flow/CoroutinesImpl.h @@ -231,10 +231,13 @@ struct NoThrowOnCancelCoroActor final : Actor::canBeSet()) { + if (!SAV::canBeSet() || actorWaitStateIsCancelled(Actor::actor_wait_state)) { return; } + // Detaching a waiter or destroying the frame can synchronously reenter cancellation. + Actor::actor_wait_state = ACTOR_WAIT_STATE_CANCELLED; + if (cancelHandler) { // The handler object is stored in the coroutine frame, so unregister // it from its wait source before destroying the frame below. From e4360ee87aed736560d111ded34b0de29ad07d58 Mon Sep 17 00:00:00 2001 From: Trevor Clinkenbeard Date: Mon, 10 Aug 2026 15:19:20 -0700 Subject: [PATCH 4/5] Document ActorCollection runtime class invariants --- flow/ActorCollection.cpp | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/flow/ActorCollection.cpp b/flow/ActorCollection.cpp index 44a897969da..14aa303f950 100644 --- a/flow/ActorCollection.cpp +++ b/flow/ActorCollection.cpp @@ -36,6 +36,7 @@ static LineageReference actorCollectionLineage(LineageReference const& parent) { class ActorCollectionRuntime; +// Owns a child callback and preserves its intrusive-list membership until completion or cancellation. class Runner final : public boost::intrusive::list_base_hook<>, public Callback, public FastAllocated, @@ -70,6 +71,7 @@ class Runner final : public boost::intrusive::list_base_hook<>, // An intrusive list of Runners, which are FastAllocated. using RunnerList = boost::intrusive::list>; +// Disposes remaining runners in insertion order before their intrusive list is destroyed. class RunnerListDestroyer final : NonCopyable { public: explicit RunnerListDestroyer(RunnerList* list) : list(list) {} @@ -82,7 +84,9 @@ class RunnerListDestroyer final : NonCopyable { RunnerList* list; }; +// Coordinates queued additions, child completion, and reentrant-safe collection teardown. class ActorCollectionRuntime final : NonCopyable { + // Keeps the add stream's single callback registered for the runtime's active lifetime. class AddActorCallback final : public SingleCallback> { public: explicit AddActorCallback(ActorCollectionRuntime* owner) : owner(owner) {} @@ -576,6 +580,7 @@ TEST_CASE("/flow/actorCollection/testCancelReentrantCollection") { } #ifdef ENABLE_SAMPLING +// Captures the active actor lineage when a collection result is delivered synchronously. class ActorCollectionLineageObserver final : public Callback, NonCopyable { public: void fire(Void const&) override { observe(); } From 6ca397400d320dd4bdb6bc128b16c6e211cdefc8 Mon Sep 17 00:00:00 2001 From: Trevor Clinkenbeard Date: Fri, 14 Aug 2026 12:26:23 -0700 Subject: [PATCH 5/5] Harden ActorCollection reentrant cleanup --- flow/ActorCollection.cpp | 101 +++++++++++++++------------- flow/include/flow/ActorCollection.h | 7 +- 2 files changed, 61 insertions(+), 47 deletions(-) diff --git a/flow/ActorCollection.cpp b/flow/ActorCollection.cpp index 14aa303f950..001c8dde3db 100644 --- a/flow/ActorCollection.cpp +++ b/flow/ActorCollection.cpp @@ -20,7 +20,6 @@ #include "flow/ActorCollection.h" #include "flow/CoroUtils.h" -#include "flow/IndexedSet.h" #include "flow/UnitTest.h" #include #include @@ -36,7 +35,7 @@ static LineageReference actorCollectionLineage(LineageReference const& parent) { class ActorCollectionRuntime; -// Owns a child callback and preserves its intrusive-list membership until completion or cancellation. +// Owns a child callback and intrusive-list membership; completion may synchronously destroy it. class Runner final : public boost::intrusive::list_base_hook<>, public Callback, public FastAllocated, @@ -71,22 +70,9 @@ class Runner final : public boost::intrusive::list_base_hook<>, // An intrusive list of Runners, which are FastAllocated. using RunnerList = boost::intrusive::list>; -// Disposes remaining runners in insertion order before their intrusive list is destroyed. -class RunnerListDestroyer final : NonCopyable { -public: - explicit RunnerListDestroyer(RunnerList* list) : list(list) {} - - ~RunnerListDestroyer() { - list->clear_and_dispose([](Runner* r) { delete r; }); - } - -private: - RunnerList* list; -}; - // Coordinates queued additions, child completion, and reentrant-safe collection teardown. class ActorCollectionRuntime final : NonCopyable { - // Keeps the add stream's single callback registered for the runtime's active lifetime. + // Keeps the add callback registered; terminal delivery may synchronously destroy it. class AddActorCallback final : public SingleCallback> { public: explicit AddActorCallback(ActorCollectionRuntime* owner) : owner(owner) {} @@ -107,7 +93,7 @@ class ActorCollectionRuntime final : NonCopyable { double* allTime, bool returnWhenEmptied) : addActor(std::move(addActor)), pCount(pCount), lastChangeTime(lastChangeTime), idleTime(idleTime), - allTime(allTime), returnWhenEmptied(returnWhenEmptied), runnersDestroyer(&runners), addActorCallback(this) { + allTime(allTime), returnWhenEmptied(returnWhenEmptied), addActorCallback(this) { if (!this->pCount) { this->pCount = &count; } @@ -197,11 +183,13 @@ class ActorCollectionRuntime final : NonCopyable { void addRunner(Future actor) { auto runner = runners.insert(runners.end(), *new Runner(this)); + // Synchronous completion stays queued until this runner has been counted. runner->start(std::move(actor)); incrementCount(); } void handleAdded(Future actor) { + ASSERT(draining); // Completing inline must not outrun an earlier queued addition. if (!runners.empty() || !actor.isReady() || addActor.isReady() || !pendingAdds.empty()) { addRunner(std::move(actor)); @@ -211,6 +199,7 @@ class ActorCollectionRuntime final : NonCopyable { if (actor.isError()) { Error e = actor.getError(); if (e.code() == error_code_actor_cancelled) { + // Cancelled children do not report completion and remain counted until teardown. addRunner(std::move(actor)); return; } @@ -324,7 +313,6 @@ class ActorCollectionRuntime final : NonCopyable { bool returnWhenEmptied; int count = 0; RunnerList runners; - RunnerListDestroyer runnersDestroyer; Promise done; AddActorCallback addActorCallback; Runner* completedHead = nullptr; @@ -403,14 +391,14 @@ void ActorCollectionRuntime::AddActorCallback::error(Error e) { owner->onAddError(e); } +// Collections that never return on empty normally end by cancellation without unwinding. static Future actorCollectionImpl(FutureStream> addActor, int* pCount, double* lastChangeTime, double* idleTime, double* allTime, - bool returnWhenEmptied, NoThrowOnCancel = {}) { - ActorCollectionRuntime runtime(std::move(addActor), pCount, lastChangeTime, idleTime, allTime, returnWhenEmptied); + ActorCollectionRuntime runtime(std::move(addActor), pCount, lastChangeTime, idleTime, allTime, false); Future result = runtime.getResult(); runtime.start(); co_await result; @@ -436,26 +424,9 @@ Future actorCollection(FutureStream> const& addActor, if (returnWhenEmptied) { return actorCollectionUntilEmpty(addActor, pCount, lastChangeTime, idleTime, allTime); } - return actorCollectionImpl(addActor, pCount, lastChangeTime, idleTime, allTime, returnWhenEmptied); + return actorCollectionImpl(addActor, pCount, lastChangeTime, idleTime, allTime); } -template -struct Traceable> { - static constexpr bool value = Traceable::value && Traceable::value; - static std::string toString(const std::pair& p) { - auto tStr = Traceable::toString(p.first); - auto uStr = Traceable::toString(p.second); - std::string result(tStr.size() + uStr.size() + 3, 'x'); - std::copy(tStr.begin(), tStr.end(), result.begin()); - auto iter = result.begin() + tStr.size(); - *(iter++) = ' '; - *(iter++) = '-'; - *(iter++) = ' '; - std::copy(uStr.begin(), uStr.end(), iter); - return result; - } -}; - void forceLinkActorCollectionTests() {} TEST_CASE("/flow/actorCollection/chooseWhen") { @@ -518,6 +489,20 @@ static Future recancelCollectionOnActorCancellation(Future pending, } } +static Future reclearNoErrorsCollectionOnActorCancellation(Future pending, + ActorCollectionNoErrors* collection, + int* cancellationCount) { + try { + co_await pending; + } catch (Error& e) { + if (e.code() == error_code_actor_cancelled) { + ++*cancellationCount; + collection->clear(); + } + throw; + } +} + // test contract that actors are cancelled when the actor collection is cleared TEST_CASE("/flow/actorCollection/testCancel") { ActorCollection actorCollection(false); @@ -566,19 +551,43 @@ TEST_CASE("/flow/actorCollection/testCancelReentrantSiblingCompletion") { TEST_CASE("/flow/actorCollection/testCancelReentrantCollection") { for (bool returnWhenEmptied : { false, true }) { for (bool clearCollection : { false, true }) { - ActorCollection collection(returnWhenEmptied); - Promise pending; - int cancellationCount = 0; - collection.add(recancelCollectionOnActorCancellation( - pending.getFuture(), &collection, returnWhenEmptied, clearCollection, &cancellationCount)); - collection.clear(returnWhenEmptied); - ASSERT_EQ(cancellationCount, 1); - ASSERT(!collection.getResult().isReady()); + for (bool nestedReturnWhenEmptied : { false, true }) { + ActorCollection collection(returnWhenEmptied); + Promise pending; + int cancellationCount = 0; + collection.add(recancelCollectionOnActorCancellation( + pending.getFuture(), &collection, nestedReturnWhenEmptied, clearCollection, &cancellationCount)); + collection.clear(returnWhenEmptied); + ASSERT_EQ(cancellationCount, 1); + Future replacement = collection.getResult(); + ASSERT(!replacement.isReady()); + collection.add(Void()); + ASSERT_EQ(replacement.isReady(), clearCollection ? nestedReturnWhenEmptied : returnWhenEmptied); + } } } return Void(); } +TEST_CASE("/flow/actorCollection/testCancelReentrantNoErrorsCollection") { + ActorCollectionNoErrors collection; + Promise pending; + int cancellationCount = 0; + collection.add(reclearNoErrorsCollectionOnActorCancellation(pending.getFuture(), &collection, &cancellationCount)); + ASSERT_EQ(collection.size(), 1); + + collection.clear(); + ASSERT_EQ(cancellationCount, 1); + ASSERT_EQ(collection.size(), 0); + + Promise replacement; + collection.add(replacement.getFuture()); + ASSERT_EQ(collection.size(), 1); + replacement.send(Void()); + ASSERT_EQ(collection.size(), 0); + return Void(); +} + #ifdef ENABLE_SAMPLING // Captures the active actor lineage when a collection result is delivered synchronously. class ActorCollectionLineageObserver final : public Callback, NonCopyable { diff --git a/flow/include/flow/ActorCollection.h b/flow/include/flow/ActorCollection.h index 31133674da7..6efb7452cb3 100644 --- a/flow/include/flow/ActorCollection.h +++ b/flow/include/flow/ActorCollection.h @@ -54,7 +54,11 @@ struct ActorCollectionNoErrors : NonCopyable { public: ActorCollectionNoErrors() { init(); } void clear() { - m_ac = Future(); + Future previous = m_ac; + previous.cancel(); + if (m_ac != previous) { + return; + } init(); } void add(Future actor) { m_add.send(actor); } @@ -73,6 +77,7 @@ class ActorCollection : NonCopyable { void add(Future a) { m_add.send(a); } Future getResult() const { return m_out; } + // A reentrant clear retains the innermost replacement and its returnWhenEmptied setting. void clear(bool returnWhenEmptied) { Future previous = m_out; previous.cancel();