28423703495

Committed 30 Jun 2026 05:59AM UTC coverage: 51.998% (-0.3%) from 52.278%

Build # 28423703495

Build Type

Pull #83

github

Committed by

web-flow

Commit Message

Merge fb542a938 into 2efed6649

Pull Request Pull Request #83: refactor and improve code QoL

Coverage Stats

37282 of 93303 branches covered (39.96%)

Branch coverage included in aggregate %.

801 of 1525 new or added lines in 78 files covered. (52.52%)

98 existing lines in 37 files now uncovered.

33674 of 43157 relevant lines covered (78.03%)

20306.85 hits per line

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

52.38

/src/dftracer/utils/core/tasks/task_result.cpp

#include <dftracer/utils/core/common/error.h>
#include <dftracer/utils/core/tasks/task_result.h>

#include <cassert>

namespace dftracer::utils {

void TaskResult::set_value(std::any value) {
    value_ = std::move(value);
    publish(State::value);
}

void TaskResult::set_exception(std::exception_ptr ex) {
    exception_ = ex;
    publish(State::exception);
}

void TaskResult::set_cancelled() { publish(State::cancelled); }

void TaskResult::mark_running() {
    auto expected = static_cast<std::uint8_t>(State::pending);
    state_.compare_exchange_strong(
        expected, static_cast<std::uint8_t>(State::running),
        std::memory_order_release, std::memory_order_relaxed);
}

void TaskResult::add_reader() {
    pending_readers_.fetch_add(1, std::memory_order_relaxed);
}

void TaskResult::release_reader() {
    int prev = pending_readers_.fetch_sub(1, std::memory_order_acq_rel);
    if (prev == 1) {
        // Last reader released -- free value memory
        std::lock_guard<std::mutex> lock(mutex_);
        value_ = std::any{};
    }
}

void TaskResult::publish(State s) {
    std::vector<std::coroutine_handle<>> to_resume;
    {
        std::lock_guard<std::mutex> lock(mutex_);
        state_.store(static_cast<std::uint8_t>(s), std::memory_order_release);
        to_resume.swap(continuations_);
    }
    cv_.notify_all();
    // Resume coroutine continuations OUTSIDE the lock.
    for (auto h : to_resume) {
        if (h && !h.done()) {
            h.resume();
        }
    }
}

bool TaskResult::wait(std::chrono::milliseconds timeout) const {
    auto ready = [this] {
        auto s = static_cast<State>(state_.load(std::memory_order_acquire));
        return s == State::value || s == State::exception ||
               s == State::cancelled;
    };
    if (ready()) return true;

    std::unique_lock<std::mutex> lock(mutex_);
    if (timeout.count() == 0) {
        cv_.wait(lock, ready);
        return true;
    }
    return cv_.wait_for(lock, timeout, ready);
}

std::any TaskResult::get() const {
    wait();
    std::lock_guard<std::mutex> lock(mutex_);
    auto s = static_cast<State>(state_.load(std::memory_order_acquire));
    if (s == State::exception) {
        std::rethrow_exception(exception_);
    }
    if (s == State::cancelled) {
        throw DFTUtilsException(ErrorCode::PIPELINE, "Task was cancelled");
    }
    return value_;  // returns COPY
}

std::any TaskResult::get_ready() const {
    std::lock_guard<std::mutex> lock(mutex_);
    auto s = static_cast<State>(state_.load(std::memory_order_acquire));
    assert(s == State::value || s == State::exception || s == State::cancelled);
    if (s == State::exception) {
        std::rethrow_exception(exception_);
    }
    if (s == State::cancelled) {
        throw DFTUtilsException(ErrorCode::PIPELINE, "Task was cancelled");
    }
    return value_;  // returns COPY
}

std::exception_ptr TaskResult::get_exception() const { return exception_; }

bool TaskResult::is_ready() const {
    auto s = static_cast<State>(state_.load(std::memory_order_acquire));
    return s == State::value || s == State::exception || s == State::cancelled;
}

bool TaskResult::has_exception() const {
    return static_cast<State>(state_.load(std::memory_order_acquire)) ==
           State::exception;
}

bool TaskResult::is_cancelled() const {
    return static_cast<State>(state_.load(std::memory_order_acquire)) ==
           State::cancelled;
}

TaskResult::State TaskResult::state() const {
    return static_cast<State>(state_.load(std::memory_order_acquire));
}

// WhenReadyAwaitable
bool TaskResult::WhenReadyAwaitable::await_ready() const noexcept {
    return result.is_ready();
}

bool TaskResult::WhenReadyAwaitable::await_suspend(std::coroutine_handle<> h) {
    std::lock_guard<std::mutex> lock(result.mutex_);
    // Double-check under lock -- result may have been published
    // between await_ready() and await_suspend().
    if (result.is_ready()) {
        return false;  // Don't suspend, resume immediately
    }
    result.continuations_.push_back(h);
    return true;       // Suspend
}

void TaskResult::WhenReadyAwaitable::await_resume() {
    if (result.has_exception()) {
        std::rethrow_exception(result.exception_);
    }
}

}  // namespace dftracer::utils

1	#include <dftracer/utils/core/common/error.h>
2	#include <dftracer/utils/core/tasks/task_result.h>
3
4	#include <cassert>
5
6	namespace dftracer::utils {
7
8	void TaskResult::set_value(std::any value) {	1,504✔
9	value_ = std::move(value);	1,504✔
10	publish(State::value);	1,519✔
11	}	1,510✔
12
13	void TaskResult::set_exception(std::exception_ptr ex) {	34✔
14	exception_ = ex;	34✔
15	publish(State::exception);	34✔
16	}	34✔
17
18	void TaskResult::set_cancelled() { publish(State::cancelled); }	×
19
20	void TaskResult::mark_running() {	1,548✔
21	auto expected = static_cast<std::uint8_t>(State::pending);	1,548✔
22	state_.compare_exchange_strong(	1,548✔
23	expected, static_cast<std::uint8_t>(State::running),
24	std::memory_order_release, std::memory_order_relaxed);
25	}	1,548✔
26
27	void TaskResult::add_reader() {	1,122✔
28	pending_readers_.fetch_add(1, std::memory_order_relaxed);	1,122✔
29	}	1,122✔
30
31	void TaskResult::release_reader() {	1,051✔
32	int prev = pending_readers_.fetch_sub(1, std::memory_order_acq_rel);	1,051✔
33	if (prev == 1) {	1,051✔
34	// Last reader released -- free value memory
35	std::lock_guard<std::mutex> lock(mutex_);	652!
36	value_ = std::any{};	652✔
37	}	652✔
38	}	1,051✔
39
40	void TaskResult::publish(State s) {	1,536✔
41	std::vector<std::coroutine_handle<>> to_resume;	1,536✔
42	{
43	std::lock_guard<std::mutex> lock(mutex_);	1,540✔
44	state_.store(static_cast<std::uint8_t>(s), std::memory_order_release);	1,553✔
45	to_resume.swap(continuations_);	1,547✔
46	}	1,536✔
47	cv_.notify_all();	1,554✔
48	// Resume coroutine continuations OUTSIDE the lock.
49	for (auto h : to_resume) {	1,548✔
50	if (h && !h.done()) {	5!
51	h.resume();	×
52	}
53	}
54	}	1,557✔
55
56	bool TaskResult::wait(std::chrono::milliseconds timeout) const {	852✔
57	auto ready = [this] {	1,278✔
58	auto s = static_cast<State>(state_.load(std::memory_order_acquire));	852✔
59	return s == State::value \|\| s == State::exception \|\|	852!
60	s == State::cancelled;	426✔
61	};	426✔
62	if (ready()) return true;	852!
63
64	std::unique_lock<std::mutex> lock(mutex_);	×
65	if (timeout.count() == 0) {	×
66	cv_.wait(lock, ready);	×
67	return true;	×
68	}
69	return cv_.wait_for(lock, timeout, ready);	×
70	}	426✔
71
72	std::any TaskResult::get() const {	674✔
73	wait();	674!
74	std::lock_guard<std::mutex> lock(mutex_);	674!
75	auto s = static_cast<State>(state_.load(std::memory_order_acquire));	674✔
76	if (s == State::exception) {	674✔
77	std::rethrow_exception(exception_);	9!
78	}
79	if (s == State::cancelled) {	668✔
NEW 80	throw DFTUtilsException(ErrorCode::PIPELINE, "Task was cancelled");	×
81	}
82	return value_; // returns COPY	1,002!
83	}	674✔
84
85	std::any TaskResult::get_ready() const {	1,053✔
86	std::lock_guard<std::mutex> lock(mutex_);	1,053!
87	auto s = static_cast<State>(state_.load(std::memory_order_acquire));	1,054✔
88	assert(s == State::value \|\| s == State::exception \|\| s == State::cancelled);	1,054!
89	if (s == State::exception) {	1,054✔
90	std::rethrow_exception(exception_);	×
91	}
92	if (s == State::cancelled) {	1,054✔
NEW 93	throw DFTUtilsException(ErrorCode::PIPELINE, "Task was cancelled");	×
94	}
95	return value_; // returns COPY	1,581!
96	}	1,053✔
97
98	std::exception_ptr TaskResult::get_exception() const { return exception_; }	10✔
99
100	bool TaskResult::is_ready() const {	14✔
101	auto s = static_cast<State>(state_.load(std::memory_order_acquire));	14✔
102	return s == State::value \|\| s == State::exception \|\| s == State::cancelled;	14!
103	}
104
105	bool TaskResult::has_exception() const {	1,563✔
106	return static_cast<State>(state_.load(std::memory_order_acquire)) ==	1,563✔
107	State::exception;	781✔
108	}
109
110	bool TaskResult::is_cancelled() const {	×
111	return static_cast<State>(state_.load(std::memory_order_acquire)) ==	×
112	State::cancelled;
113	}
114
115	TaskResult::State TaskResult::state() const {	×
116	return static_cast<State>(state_.load(std::memory_order_acquire));	×
117	}
118
119	// WhenReadyAwaitable
120	bool TaskResult::WhenReadyAwaitable::await_ready() const noexcept {	×
121	return result.is_ready();	×
122	}
123
124	bool TaskResult::WhenReadyAwaitable::await_suspend(std::coroutine_handle<> h) {	×
125	std::lock_guard<std::mutex> lock(result.mutex_);	×
126	// Double-check under lock -- result may have been published
127	// between await_ready() and await_suspend().
128	if (result.is_ready()) {	×
129	return false; // Don't suspend, resume immediately	×
130	}
131	result.continuations_.push_back(h);	×
132	return true; // Suspend	×
133	}	×
134
135	void TaskResult::WhenReadyAwaitable::await_resume() {	×
136	if (result.has_exception()) {	×
137	std::rethrow_exception(result.exception_);	×
138	}
139	}	×
140
141	} // namespace dftracer::utils

llnl / dftracer-utils / 28423703495

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