Commit 0249f352 authored by Lewis Baker's avatar Lewis Baker Committed by Facebook GitHub Bot

Improve stack-tracing for tasks launched by the merge() internals

Summary:
The stack-traces for child worker tasks launched to process each
incoming AsyncGenerator now have the same common consumer-task
as the parent frame, which is then a detached-task, rather than
each of the child workers being their own detached tasks.

Reviewed By: andriigrynenko

Differential Revision: D24480161

fbshipit-source-id: c02887abf676bde9b297ee6d8f0fcabe959616e3
parent d35391f4
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include <folly/experimental/coro/WithCancellation.h> #include <folly/experimental/coro/WithCancellation.h>
#include <folly/experimental/coro/detail/Barrier.h> #include <folly/experimental/coro/detail/Barrier.h>
#include <folly/experimental/coro/detail/BarrierTask.h> #include <folly/experimental/coro/detail/BarrierTask.h>
#include <folly/experimental/coro/detail/CurrentAsyncFrame.h>
#include <folly/experimental/coro/detail/Helpers.h> #include <folly/experimental/coro/detail/Helpers.h>
#include <exception> #include <exception>
#include <memory> #include <memory>
...@@ -111,13 +112,15 @@ AsyncGenerator<Reference, Value> merge( ...@@ -111,13 +112,15 @@ AsyncGenerator<Reference, Value> merge(
detail::Barrier barrier{1}; detail::Barrier barrier{1};
auto& asyncFrame = co_await detail::co_current_async_stack_frame;
exception_wrapper ex; exception_wrapper ex;
try { try {
while (auto item = co_await sources_.next()) { while (auto item = co_await sources_.next()) {
if (state->cancelSource.isCancellationRequested()) { if (state->cancelSource.isCancellationRequested()) {
break; break;
} }
makeWorkerTask(state, *std::move(item)).start(&barrier); makeWorkerTask(state, *std::move(item)).start(&barrier, asyncFrame);
} }
} catch (const std::exception& e) { } catch (const std::exception& e) {
ex = exception_wrapper{std::current_exception(), e}; ex = exception_wrapper{std::current_exception(), e};
...@@ -135,7 +138,7 @@ AsyncGenerator<Reference, Value> merge( ...@@ -135,7 +138,7 @@ AsyncGenerator<Reference, Value> merge(
CallbackRecord<Reference>{callback_record_error, std::move(ex)}; CallbackRecord<Reference>{callback_record_error, std::move(ex)};
state->recordPublished.post(); state->recordPublished.post();
} }
}; }
// Wait for all worker tasks to finish consuming the entirety of their // Wait for all worker tasks to finish consuming the entirety of their
// input streams. // input streams.
......
...@@ -197,6 +197,15 @@ class DetachedBarrierTask { ...@@ -197,6 +197,15 @@ class DetachedBarrierTask {
std::move(*this).start(barrier, FOLLY_ASYNC_STACK_RETURN_ADDRESS()); std::move(*this).start(barrier, FOLLY_ASYNC_STACK_RETURN_ADDRESS());
} }
FOLLY_NOINLINE void start(
Barrier* barrier,
folly::AsyncStackFrame& parentFrame) &&
noexcept {
assert(coro_);
coro_.promise().getAsyncFrame().setParentFrame(parentFrame);
std::move(*this).start(barrier, FOLLY_ASYNC_STACK_RETURN_ADDRESS());
}
void start(Barrier* barrier, void* returnAddress) && noexcept { void start(Barrier* barrier, void* returnAddress) && noexcept {
assert(coro_); assert(coro_);
assert(barrier != nullptr); assert(barrier != nullptr);
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment