23529483807

Committed 25 Mar 2026 07:17AM UTC coverage: 48.515% (-1.6%) from 50.098%

Build # 23529483807

Build Type

Pull #57

github

Committed by

web-flow

Commit Message

Merge 5b1e117ad into 38f9f3616

Pull Request Pull Request #57: feat(comparator): add pairwise traces comparator

Coverage Stats

18829 of 49412 branches covered (38.11%)

Branch coverage included in aggregate %.

1584 of 1933 new or added lines in 14 files covered. (81.95%)

3552 existing lines in 135 files now uncovered.

18474 of 27477 relevant lines covered (67.23%)

241072.53 hits per line

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

59.91

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

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

#include <algorithm>
#include <stdexcept>
#include <thread>

namespace dftracer::utils {

Runtime::Runtime(std::size_t threads)
    : threads_(threads == 0 ? std::thread::hardware_concurrency() : threads) {
    ExecutorConfig config;
    config.num_threads = threads_;
    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_(config.num_threads == 0 ? std::thread::hardware_concurrency()
                                       : config.num_threads) {
    executor_ = std::make_unique<Executor>(config);
    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_(config.num_threads == 0 ? std::thread::hardware_concurrency()
                                       : config.num_threads) {
    executor_ = std::make_unique<Executor>(config);
    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 std::runtime_error("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);
            exec->mark_coro_completed(task_id->load(std::memory_order_acquire));
        } catch (...) {
            exec->mark_coro_completed(task_id->load(std::memory_order_acquire));
            p->set_exception(std::current_exception());
            co_return;
        }
        p->set_value();
    };

    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 std::runtime_error(
            "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 std::runtime_error(
            "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_; }

}  // namespace dftracer::utils

1	#include <dftracer/utils/core/common/logging.h>
2	#include <dftracer/utils/core/runtime.h>
3
4	#include <algorithm>
5	#include <stdexcept>
6	#include <thread>
7
8	namespace dftracer::utils {
9
10	Runtime::Runtime(std::size_t threads)	292✔
11	: threads_(threads == 0 ? std::thread::hardware_concurrency() : threads) {	292✔
12	ExecutorConfig config;	146!
13	config.num_threads = threads_;	146✔
14	executor_ = std::make_unique<Executor>(config);	146!
15	executor_->start();	146!
16
17	watchdog_ = std::make_unique<Watchdog>();	146!
18	watchdog_->set_executor(executor_.get());	146!
19	}	146✔
20
UNCOV 21	Runtime::Runtime(const ExecutorConfig& config, bool enable_watchdog)	×
22	: threads_(config.num_threads == 0 ? std::thread::hardware_concurrency()	×
NEW 23	: config.num_threads) {	×
24	executor_ = std::make_unique<Executor>(config);	×
25	executor_->start();	×
26
UNCOV 27	if (enable_watchdog) {	×
28	watchdog_ = std::make_unique<Watchdog>();	×
29	watchdog_->set_executor(executor_.get());	×
30	}	×
UNCOV 31	}	×
32
33	Runtime::Runtime(const ExecutorConfig& config,	350✔
34	std::unique_ptr<Watchdog> watchdog)
35	: threads_(config.num_threads == 0 ? std::thread::hardware_concurrency()	175✔
36	: config.num_threads) {	344✔
37	executor_ = std::make_unique<Executor>(config);	175!
38	executor_->start();	175!
39
40	watchdog_ = std::move(watchdog);	175✔
41	if (watchdog_) {	175✔
42	watchdog_->set_executor(executor_.get());	91!
43	}	91✔
44	}	175✔
45
46	Runtime::~Runtime() { shutdown(); }	642!
47
48	TaskHandle Runtime::submit(coro::CoroTask<void> task, std::string name) {	222✔
49	if (shutdown_called_.load(std::memory_order_acquire)) {	222✔
50	throw std::runtime_error("Runtime is shut down");	2!
51	}
52	if (name.empty()) {	220✔
53	name = "task-" + std::to_string(task_name_counter_++);	3!
54	}	3✔
55
56	auto promise = std::make_shared<std::promise<void>>();	220✔
57	auto future = promise->get_future().share();	220!
58	auto tid = std::make_shared<std::atomic<TaskIndex>>(-1);	220!
59
60	auto wrapper =	220✔
61	[](coro::CoroTask<void> t, std::shared_ptr<std::promise<void>> p,	1,534!
62	Executor* exec,
63	std::shared_ptr<std::atomic<TaskIndex>> task_id) -> coro::Coro {	220!
64	try {
65	co_await std::move(t);	1,094!
66	exec->mark_coro_completed(task_id->load(std::memory_order_acquire));	217!
67	} catch (...) {	220✔
68	exec->mark_coro_completed(task_id->load(std::memory_order_acquire));	3!
69	p->set_exception(std::current_exception());	3!
70	co_return;	3✔
71	}	3!
72	p->set_value();	217!
73	};	437✔
74
75	auto coro = wrapper(std::move(task), promise, executor_.get(), tid);	220!
76	TaskIndex id = executor_->enqueue_tracked(std::move(coro), name, tid);	220!
77
78	{
79	std::lock_guard<std::mutex> lock(futures_mutex_);	220!
80	cleanup_completed_futures();	220!
81	outstanding_futures_.push_back(future);	220!
82	}	220✔
83
84	return TaskHandle{future, id, std::move(name)};	220!
85	}	220✔
86
87	void Runtime::wait_all() {	77✔
88	std::vector<std::shared_future<void>> futures;	77✔
89	{
90	std::lock_guard<std::mutex> lock(futures_mutex_);	77!
91	futures = std::move(outstanding_futures_);	77✔
92	outstanding_futures_.clear();	77✔
93	}	77✔
94	for (auto& f : futures) {	95✔
95	f.wait();	18!
96	}
97	}	77✔
98
99	void Runtime::cleanup_completed_futures() {	312✔
100	outstanding_futures_.erase(	624✔
101	std::remove_if(outstanding_futures_.begin(), outstanding_futures_.end(),	312✔
102	[](const std::shared_future<void>& f) {	287✔
103	return f.wait_for(std::chrono::seconds(0)) ==	287✔
104	std::future_status::ready;
105	}),
106	outstanding_futures_.end());	312✔
107	}	312✔
108
109	ExecutorProgress Runtime::get_progress() const {	19✔
110	return executor_->get_progress();	19✔
111	}
112
113	bool Runtime::is_responsive() const { return executor_->is_responsive(); }	2✔
114
UNCOV 115	void Runtime::set_global_timeout(std::chrono::milliseconds timeout) {	×
116	if (!watchdog_) {	×
117	throw std::runtime_error(	×
118	"Cannot set timeout: Runtime created without watchdog");
119	}
UNCOV 120	watchdog_->set_global_timeout(timeout);	×
121	}	×
122
UNCOV 123	void Runtime::set_default_task_timeout(std::chrono::milliseconds timeout) {	×
124	if (!watchdog_) {	×
125	throw std::runtime_error(	×
126	"Cannot set timeout: Runtime created without watchdog");
127	}
UNCOV 128	watchdog_->set_default_task_timeout(timeout);	×
129	}	×
130
131	void Runtime::shutdown() {	609✔
132	bool expected = false;	609✔
133	if (!shutdown_called_.compare_exchange_strong(expected, true)) return;	609✔
134	if (watchdog_) watchdog_->stop();	321✔
135	if (executor_) executor_->shutdown();	321!
136	}	609✔
137
138	std::size_t Runtime::threads() const { return threads_; }	9✔
139
140	} // namespace dftracer::utils

llnl / dftracer-utils / 23529483807

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