30067416336

Committed 24 Jul 2026 04:40AM UTC coverage: 50.693% (-2.0%) from 52.66%

Build # 30067416336

Build Type

Pull #99

github

Committed by

web-flow

Commit Message

Merge 464586f0d into 06bc84ec9

Pull Request Pull Request #99: Support CM time_metric (NS/MS/SEC/US) across reader and viz

Coverage Stats

18063 of 48203 branches covered (37.47%)

Branch coverage included in aggregate %.

1514 of 2204 new or added lines in 58 files covered. (68.69%)

741 existing lines in 114 files now uncovered.

23715 of 34210 relevant lines covered (69.32%)

39842.55 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

60.32

/src/dftracer/utils/core/runtime.cpp

#include <dftracer/utils/core/common/error.h>
#include <dftracer/utils/core/common/logging.h>
#include <dftracer/utils/core/common/platform_compat.h>
#include <dftracer/utils/core/env.h>
#include <dftracer/utils/core/runtime.h>

#include <algorithm>
#include <cstdlib>
#include <stdexcept>
#include <string>
#include <string_view>
#include <thread>

namespace dftracer::utils {

namespace {
// Resolve the worker-thread count. DFTRACER_UTILS_THREADS overrides everything
// (a debugging lever: force a fixed count, e.g. 1 for a single-threaded async
// loop). Otherwise 0 means hardware_concurrency.
std::size_t resolve_threads(std::size_t requested) {
    if (auto env = Env::get<std::string_view>("DFTRACER_UTILS_THREADS");
        env.has_value()) {
        const long long n =
            std::strtoll(std::string(*env).c_str(), nullptr, 10);
        if (n > 0) return static_cast<std::size_t>(n);
    }
    return requested == 0 ? dftracer_utils_hardware_concurrency() : requested;
}

// Resolve the I/O-pool size. DFTRACER_UTILS_IO_THREADS overrides everything (a
// lever to bound the epoll/kqueue thread pool independently of the compute
// pool, e.g. so the coordinator does not open a full-machine-sized pool).
// Otherwise 0 means hardware_concurrency, matching the executor default.
std::size_t resolve_io_threads(std::size_t requested) {
    if (auto env = Env::get<std::string_view>("DFTRACER_UTILS_IO_THREADS");
        env.has_value()) {
        const long long n =
            std::strtoll(std::string(*env).c_str(), nullptr, 10);
        if (n > 0) return static_cast<std::size_t>(n);
    }
    return requested == 0 ? dftracer_utils_hardware_concurrency() : requested;
}
}  // namespace

Runtime::Runtime(std::size_t threads) : threads_(resolve_threads(threads)) {
    ExecutorConfig config;
    config.num_threads = threads_;
    config.io_pool_size = resolve_io_threads(config.io_pool_size);
    executor_ = std::make_unique<Executor>(config);
    executor_->start();

    watchdog_ = std::make_unique<Watchdog>();
    watchdog_->set_executor(executor_.get());
}

Runtime::Runtime(const ExecutorConfig& config, bool enable_watchdog)
    : threads_(resolve_threads(config.num_threads)) {
    ExecutorConfig cfg = config;
    cfg.num_threads = threads_;
    cfg.io_pool_size = resolve_io_threads(config.io_pool_size);
    executor_ = std::make_unique<Executor>(cfg);
    executor_->start();

    if (enable_watchdog) {
        watchdog_ = std::make_unique<Watchdog>();
        watchdog_->set_executor(executor_.get());
    }
}

Runtime::Runtime(const ExecutorConfig& config,
                 std::unique_ptr<Watchdog> watchdog)
    : threads_(resolve_threads(config.num_threads)) {
    ExecutorConfig cfg = config;
    cfg.num_threads = threads_;
    cfg.io_pool_size = resolve_io_threads(config.io_pool_size);
    executor_ = std::make_unique<Executor>(cfg);
    executor_->start();

    watchdog_ = std::move(watchdog);
    if (watchdog_) {
        watchdog_->set_executor(executor_.get());
    }
}

Runtime::~Runtime() { shutdown(); }

TaskHandle Runtime::submit(coro::CoroTask<void> task, std::string name) {
    if (shutdown_called_.load(std::memory_order_acquire)) {
        throw DFTUtilsException(ErrorCode::PIPELINE, "Runtime is shut down");
    }
    if (name.empty()) {
        name = "task-" + std::to_string(task_name_counter_++);
    }

    auto promise = std::make_shared<std::promise<void>>();
    auto future = promise->get_future().share();
    auto tid = std::make_shared<std::atomic<TaskIndex>>(-1);

    auto wrapper =
        [](coro::CoroTask<void> t, std::shared_ptr<std::promise<void>> p,
           Executor* exec,
           std::shared_ptr<std::atomic<TaskIndex>> task_id) -> coro::Coro {
        try {
            co_await std::move(t);
            t = coro::CoroTask<void>{
                std::coroutine_handle<coro::CoroTask<void>::promise_type>{}};
            exec->mark_coro_completed(task_id->load(std::memory_order_acquire));
        } catch (...) {
            t = coro::CoroTask<void>{
                std::coroutine_handle<coro::CoroTask<void>::promise_type>{}};
            exec->mark_coro_completed(task_id->load(std::memory_order_acquire));
            p->set_exception(std::current_exception());
            co_return;
        }
        p->set_value();
    };

    // Set the executor on the task's promise so awaitables (e.g. channels)
    // that capture `get_root_promise()->get_executor()` can schedule
    // resumption. Without this, awaiters end up with executor=nullptr because
    // the wrapping `coro::Coro` doesn't extend PromiseBase and the
    // root-promise chain stops at the user's CoroTask.
    if (task.handle()) {
        task.handle().promise().set_executor(executor_.get());
    }
    auto coro = wrapper(std::move(task), promise, executor_.get(), tid);
    TaskIndex id = executor_->enqueue_tracked(std::move(coro), name, tid);

    {
        std::lock_guard<std::mutex> lock(futures_mutex_);
        cleanup_completed_futures();
        outstanding_futures_.push_back(future);
    }

    return TaskHandle{future, id, std::move(name)};
}

void Runtime::wait_all() {
    std::vector<std::shared_future<void>> futures;
    {
        std::lock_guard<std::mutex> lock(futures_mutex_);
        futures = std::move(outstanding_futures_);
        outstanding_futures_.clear();
    }
    for (auto& f : futures) {
        f.wait();
    }
}

void Runtime::cleanup_completed_futures() {
    outstanding_futures_.erase(
        std::remove_if(outstanding_futures_.begin(), outstanding_futures_.end(),
                       [](const std::shared_future<void>& f) {
                           return f.wait_for(std::chrono::seconds(0)) ==
                                  std::future_status::ready;
                       }),
        outstanding_futures_.end());
}

ExecutorProgress Runtime::get_progress() const {
    return executor_->get_progress();
}

bool Runtime::is_responsive() const { return executor_->is_responsive(); }

void Runtime::set_global_timeout(std::chrono::milliseconds timeout) {
    if (!watchdog_) {
        throw DFTUtilsException(
            ErrorCode::PIPELINE,
            "Cannot set timeout: Runtime created without watchdog");
    }
    watchdog_->set_global_timeout(timeout);
}

void Runtime::set_default_task_timeout(std::chrono::milliseconds timeout) {
    if (!watchdog_) {
        throw DFTUtilsException(
            ErrorCode::PIPELINE,
            "Cannot set timeout: Runtime created without watchdog");
    }
    watchdog_->set_default_task_timeout(timeout);
}

void Runtime::shutdown() {
    bool expected = false;
    if (!shutdown_called_.compare_exchange_strong(expected, true)) return;
    if (watchdog_) watchdog_->stop();
    if (executor_) executor_->shutdown();
}

std::size_t Runtime::threads() const { return threads_; }

std::size_t Runtime::io_threads() const {
    return executor_ ? executor_->get_io_pool_size() : 0;
}

}  // namespace dftracer::utils

1	#include <dftracer/utils/core/common/error.h>
2	#include <dftracer/utils/core/common/logging.h>
3	#include <dftracer/utils/core/common/platform_compat.h>
4	#include <dftracer/utils/core/env.h>
5	#include <dftracer/utils/core/runtime.h>
6
7	#include <algorithm>
8	#include <cstdlib>
9	#include <stdexcept>
10	#include <string>
11	#include <string_view>
12	#include <thread>
13
14	namespace dftracer::utils {
15
16	namespace {
17	// Resolve the worker-thread count. DFTRACER_UTILS_THREADS overrides everything
18	// (a debugging lever: force a fixed count, e.g. 1 for a single-threaded async
19	// loop). Otherwise 0 means hardware_concurrency.
20	std::size_t resolve_threads(std::size_t requested) {	455✔
21	if (auto env = Env::get<std::string_view>("DFTRACER_UTILS_THREADS");	910!
22	env.has_value()) {	455✔
23	const long long n =
24	std::strtoll(std::string(*env).c_str(), nullptr, 10);	×
25	if (n > 0) return static_cast<std::size_t>(n);	×
26	}
27	return requested == 0 ? dftracer_utils_hardware_concurrency() : requested;	455✔
28	}
29
30	// Resolve the I/O-pool size. DFTRACER_UTILS_IO_THREADS overrides everything (a
31	// lever to bound the epoll/kqueue thread pool independently of the compute
32	// pool, e.g. so the coordinator does not open a full-machine-sized pool).
33	// Otherwise 0 means hardware_concurrency, matching the executor default.
34	std::size_t resolve_io_threads(std::size_t requested) {	455✔
35	if (auto env = Env::get<std::string_view>("DFTRACER_UTILS_IO_THREADS");	910!
36	env.has_value()) {	455✔
37	const long long n =
NEW 38	std::strtoll(std::string(*env).c_str(), nullptr, 10);	×
NEW 39	if (n > 0) return static_cast<std::size_t>(n);	×
40	}
41	return requested == 0 ? dftracer_utils_hardware_concurrency() : requested;	455✔
42	}
43	} // namespace
44
45	Runtime::Runtime(std::size_t threads) : threads_(resolve_threads(threads)) {	146!
46	ExecutorConfig config;	146✔
47	config.num_threads = threads_;	146✔
48	config.io_pool_size = resolve_io_threads(config.io_pool_size);	146!
49	executor_ = std::make_unique<Executor>(config);	146!
50	executor_->start();	146!
51
52	watchdog_ = std::make_unique<Watchdog>();	146!
53	watchdog_->set_executor(executor_.get());	146!
54	}	146✔
55
56	Runtime::Runtime(const ExecutorConfig& config, bool enable_watchdog)	78✔
57	: threads_(resolve_threads(config.num_threads)) {	78!
58	ExecutorConfig cfg = config;	78✔
59	cfg.num_threads = threads_;	78✔
60	cfg.io_pool_size = resolve_io_threads(config.io_pool_size);	78!
61	executor_ = std::make_unique<Executor>(cfg);	78!
62	executor_->start();	78!
63
64	if (enable_watchdog) {	78!
65	watchdog_ = std::make_unique<Watchdog>();	78!
66	watchdog_->set_executor(executor_.get());	78!
67	}
68	}	78✔
69
70	Runtime::Runtime(const ExecutorConfig& config,	231✔
71	std::unique_ptr<Watchdog> watchdog)	231✔
72	: threads_(resolve_threads(config.num_threads)) {	231!
73	ExecutorConfig cfg = config;	231✔
74	cfg.num_threads = threads_;	231✔
75	cfg.io_pool_size = resolve_io_threads(config.io_pool_size);	231!
76	executor_ = std::make_unique<Executor>(cfg);	231!
77	executor_->start();	231!
78
79	watchdog_ = std::move(watchdog);	231✔
80	if (watchdog_) {	231✔
81	watchdog_->set_executor(executor_.get());	81!
82	}
83	}	231✔
84
85	Runtime::~Runtime() { shutdown(); }	455✔
86
87	TaskHandle Runtime::submit(coro::CoroTask<void> task, std::string name) {	525✔
88	if (shutdown_called_.load(std::memory_order_acquire)) {	525✔
89	throw DFTUtilsException(ErrorCode::PIPELINE, "Runtime is shut down");	2!
90	}
91	if (name.empty()) {	523✔
92	name = "task-" + std::to_string(task_name_counter_++);	3!
93	}
94
95	auto promise = std::make_shared<std::promise<void>>();	523!
96	auto future = promise->get_future().share();	523!
97	auto tid = std::make_shared<std::atomic<TaskIndex>>(-1);	523!
98
99	auto wrapper =
100	[](coro::CoroTask<void> t, std::shared_ptr<std::promise<void>> p,	523!
101	Executor* exec,
102	std::shared_ptr<std::atomic<TaskIndex>> task_id) -> coro::Coro {
103	try {
104	co_await std::move(t);
105	t = coro::CoroTask<void>{
106	std::coroutine_handle<coro::CoroTask<void>::promise_type>{}};
107	exec->mark_coro_completed(task_id->load(std::memory_order_acquire));
108	} catch (...) {
109	t = coro::CoroTask<void>{
110	std::coroutine_handle<coro::CoroTask<void>::promise_type>{}};
111	exec->mark_coro_completed(task_id->load(std::memory_order_acquire));
112	p->set_exception(std::current_exception());
113	co_return;
114	}
115	p->set_value();
116	};	1,046!
117
118	// Set the executor on the task's promise so awaitables (e.g. channels)
119	// that capture `get_root_promise()->get_executor()` can schedule
120	// resumption. Without this, awaiters end up with executor=nullptr because
121	// the wrapping `coro::Coro` doesn't extend PromiseBase and the
122	// root-promise chain stops at the user's CoroTask.
123	if (task.handle()) {	523!
124	task.handle().promise().set_executor(executor_.get());	523✔
125	}
126	auto coro = wrapper(std::move(task), promise, executor_.get(), tid);	1,046!
127	TaskIndex id = executor_->enqueue_tracked(std::move(coro), name, tid);	523!
128
129	{
130	std::lock_guard<std::mutex> lock(futures_mutex_);	523!
131	cleanup_completed_futures();	523!
132	outstanding_futures_.push_back(future);	523!
133	}	523✔
134
135	return TaskHandle{future, id, std::move(name)};	1,046✔
136	}	523✔
137
138	void Runtime::wait_all() {	100✔
139	std::vector<std::shared_future<void>> futures;	100✔
140	{
141	std::lock_guard<std::mutex> lock(futures_mutex_);	100!
142	futures = std::move(outstanding_futures_);	100✔
143	outstanding_futures_.clear();	100✔
144	}	100✔
145	for (auto& f : futures) {	141✔
146	f.wait();	41!
147	}
148	}	100✔
149
150	void Runtime::cleanup_completed_futures() {	609✔
151	outstanding_futures_.erase(	1,218!
152	std::remove_if(outstanding_futures_.begin(), outstanding_futures_.end(),	609!
153	[](const std::shared_future<void>& f) {	474✔
154	return f.wait_for(std::chrono::seconds(0)) ==	474!
155	std::future_status::ready;	474✔
156	}),
157	outstanding_futures_.end());	609✔
158	}	609✔
159
160	ExecutorProgress Runtime::get_progress() const {	19✔
161	return executor_->get_progress();	19✔
162	}
163
164	bool Runtime::is_responsive() const { return executor_->is_responsive(); }	2✔
165
166	void Runtime::set_global_timeout(std::chrono::milliseconds timeout) {	×
167	if (!watchdog_) {	×
168	throw DFTUtilsException(	×
169	ErrorCode::PIPELINE,
170	"Cannot set timeout: Runtime created without watchdog");	×
171	}
172	watchdog_->set_global_timeout(timeout);	×
173	}	×
174
175	void Runtime::set_default_task_timeout(std::chrono::milliseconds timeout) {	×
176	if (!watchdog_) {	×
177	throw DFTUtilsException(	×
178	ErrorCode::PIPELINE,
179	"Cannot set timeout: Runtime created without watchdog");	×
180	}
181	watchdog_->set_default_task_timeout(timeout);	×
182	}	×
183
184	void Runtime::shutdown() {	862✔
185	bool expected = false;	862✔
186	if (!shutdown_called_.compare_exchange_strong(expected, true)) return;	862✔
187	if (watchdog_) watchdog_->stop();	455!
188	if (executor_) executor_->shutdown();	455!
189	}
190
191	std::size_t Runtime::threads() const { return threads_; }	167✔
192
193	std::size_t Runtime::io_threads() const {	×
194	return executor_ ? executor_->get_io_pool_size() : 0;	×
195	}
196
197	} // namespace dftracer::utils

llnl / dftracer-utils / 30067416336

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous