26195612357

Committed 20 May 2026 11:19PM UTC coverage: 49.859% (-2.3%) from 52.2%

Build # 26195612357

Build Type

push

github

Committed by

hariharan-devarajan

Commit Message

feat(aggregator): improve system metrics scanning and persistence error handling

Coverage Stats

16041 of 43831 branches covered (36.6%)

Branch coverage included in aggregate %.

6 of 17 new or added lines in 2 files covered. (35.29%)

1072 existing lines in 104 files now uncovered.

21423 of 31309 relevant lines covered (68.42%)

13054.31 hits per line

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

35.21

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

#include <dftracer/utils/core/common/logging.h>
#include <dftracer/utils/core/pipeline/error.h>
#include <dftracer/utils/core/tasks/coro_scope.h>
#include <dftracer/utils/core/tasks/task.h>

#include <algorithm>
#include <any>
#include <optional>
#include <sstream>

namespace dftracer::utils {

std::shared_ptr<Task> Task::depends_on(std::shared_ptr<Task> parent) {
    if (!parent) {
        throw PipelineError(PipelineError::VALIDATION_ERROR,
                            "Cannot add null parent to task");
    }

    // Add parent to this task
    parents_.push_back(parent);

    // Add this task as child to parent
    parent->add_child(shared_from_this());

    // Register as reader for smart value release
    parent->result_.add_reader();

    return shared_from_this();
}

std::shared_ptr<Task> Task::depends_on(
    std::initializer_list<std::shared_ptr<Task>> parent_list) {
    // Simply delegate to single-parent version for each parent
    for (auto& parent : parent_list) {
        depends_on(parent);
    }
    return shared_from_this();
}

std::shared_ptr<Task> Task::with_combiner(
    std::function<std::any(const std::vector<std::any>&)> combiner) {
    input_combiner_ = std::move(combiner);
    has_custom_combiner_ = true;
    return shared_from_this();
}

std::shared_ptr<Task> Task::with_name(std::string name) {
    name_ = std::move(name);
    return shared_from_this();
}

coro::CoroTask<std::any> Task::execute(CoroScope& context,
                                       const std::any& input) {
    try {
        std::any result = co_await func_(context, input);
        co_return result;
    } catch (const std::exception& e) {
        DFTRACER_UTILS_LOG_ERROR("Task '%s' execution failed: %s", get_name(),
                                 e.what());
        throw;
    }
}

void Task::set_result(std::any result) { result_.set_value(std::move(result)); }

void Task::set_exception(std::exception_ptr ex) { result_.set_exception(ex); }

std::any Task::apply_combiner(const std::vector<std::any>& inputs) const {
    if (!has_custom_combiner_) {
        throw PipelineError(PipelineError::VALIDATION_ERROR,
                            "Task has no custom combiner");
    }
    return input_combiner_(inputs);
}

bool Task::validate_connection(std::type_index from, std::type_index to) const {
    // Void can connect to anything (synchronization)
    if (from == typeid(void)) {
        return true;
    }

    // std::any can accept any type (for tap, logging, etc.)
    if (to == typeid(std::any)) {
        return true;
    }

    // Exact type match
    return from == to;
}

std::shared_ptr<Task> Task::operator&(std::shared_ptr<Task> other) {
    // Create a combiner task that depends on both this and other
    // When a task has multiple parents, it receives a vector<any> as input
    auto combiner = make_task(
        [](CoroScope&,
           const std::vector<std::any>& inputs) -> coro::CoroTask<std::any> {
            // inputs[0] is from first parent (this), inputs[1] is from second
            // (other)
            if (inputs.size() != 2) {
                throw std::runtime_error(
                    "AND combiner expects exactly 2 inputs");
            }
            // Return tuple of both results
            co_return std::make_any<std::tuple<std::any, std::any>>(inputs[0],
                                                                    inputs[1]);
        },
        "AND_combiner");

    // Combiner depends on both tasks
    combiner->depends_on(shared_from_this());
    combiner->depends_on(other);

    // Set a custom combiner to pass inputs as vector
    combiner->with_combiner(
        [](const std::vector<std::any>& inputs) -> std::any {
            return std::make_any<std::vector<std::any>>(inputs);
        });

    return combiner;
}

std::shared_ptr<Task> Task::operator^(std::shared_ptr<Task> tap_task) {
    if (!tap_task) {
        throw PipelineError(PipelineError::VALIDATION_ERROR,
                            "Tap task cannot be null");
    }
    auto passthrough = make_task(
        [](CoroScope&, const std::any& input) -> coro::CoroTask<std::any> {
            co_return input;
        },
        "TAP_passthrough");
    passthrough->depends_on(shared_from_this());
    tap_task->depends_on(shared_from_this());
    return passthrough;
}

}  // namespace dftracer::utils

1	#include <dftracer/utils/core/common/logging.h>
2	#include <dftracer/utils/core/pipeline/error.h>
3	#include <dftracer/utils/core/tasks/coro_scope.h>
4	#include <dftracer/utils/core/tasks/task.h>
5
6	#include <algorithm>
7	#include <any>
8	#include <optional>
9	#include <sstream>
10
11	namespace dftracer::utils {
12
13	std::shared_ptr<Task> Task::depends_on(std::shared_ptr<Task> parent) {	561✔
14	if (!parent) {	561!
15	throw PipelineError(PipelineError::VALIDATION_ERROR,	×
16	"Cannot add null parent to task");	×
17	}
18
19	// Add parent to this task
20	parents_.push_back(parent);	561!
21
22	// Add this task as child to parent
23	parent->add_child(shared_from_this());	561!
24
25	// Register as reader for smart value release
26	parent->result_.add_reader();	561✔
27
28	return shared_from_this();	561✔
29	}
30
31	std::shared_ptr<Task> Task::depends_on(	6✔
32	std::initializer_list<std::shared_ptr<Task>> parent_list) {
33	// Simply delegate to single-parent version for each parent
34	for (auto& parent : parent_list) {	19✔
35	depends_on(parent);	13!
36	}
37	return shared_from_this();	6✔
38	}
39
40	std::shared_ptr<Task> Task::with_combiner(	8✔
41	std::function<std::any(const std::vector<std::any>&)> combiner) {
42	input_combiner_ = std::move(combiner);	8✔
43	has_custom_combiner_ = true;	8✔
44	return shared_from_this();	8✔
45	}
46
47	std::shared_ptr<Task> Task::with_name(std::string name) {	×
48	name_ = std::move(name);	×
49	return shared_from_this();	×
50	}
51
52	coro::CoroTask<std::any> Task::execute(CoroScope& context,	768!
53	const std::any& input) {
54	try {
55	std::any result = co_await func_(context, input);
56	co_return result;
57	} catch (const std::exception& e) {
58	DFTRACER_UTILS_LOG_ERROR("Task '%s' execution failed: %s", get_name(),
59	e.what());
60	throw;
61	}
62	}	1,540!
63
64	void Task::set_result(std::any result) { result_.set_value(std::move(result)); }	744!
65
66	void Task::set_exception(std::exception_ptr ex) { result_.set_exception(ex); }	17!
67
68	std::any Task::apply_combiner(const std::vector<std::any>& inputs) const {	10✔
69	if (!has_custom_combiner_) {	10!
70	throw PipelineError(PipelineError::VALIDATION_ERROR,	×
71	"Task has no custom combiner");	×
72	}
73	return input_combiner_(inputs);	10✔
74	}
75
76	bool Task::validate_connection(std::type_index from, std::type_index to) const {	×
77	// Void can connect to anything (synchronization)
78	if (from == typeid(void)) {	×
79	return true;	×
80	}
81
82	// std::any can accept any type (for tap, logging, etc.)
83	if (to == typeid(std::any)) {	×
84	return true;	×
85	}
86
87	// Exact type match
88	return from == to;	×
89	}
90
91	std::shared_ptr<Task> Task::operator&(std::shared_ptr<Task> other) {	1✔
92	// Create a combiner task that depends on both this and other
93	// When a task has multiple parents, it receives a vector<any> as input
94	auto combiner = make_task(	1✔
95	[](CoroScope&,	1!
96	const std::vector<std::any>& inputs) -> coro::CoroTask<std::any> {
97	// inputs[0] is from first parent (this), inputs[1] is from second
98	// (other)
99	if (inputs.size() != 2) {
100	throw std::runtime_error(
101	"AND combiner expects exactly 2 inputs");
102	}
103	// Return tuple of both results
104	co_return std::make_any<std::tuple<std::any, std::any>>(inputs[0],
105	inputs[1]);
106	},	2!
107	"AND_combiner");	1!
108
109	// Combiner depends on both tasks
110	combiner->depends_on(shared_from_this());	1!
111	combiner->depends_on(other);	1!
112
113	// Set a custom combiner to pass inputs as vector
114	combiner->with_combiner(	1!
115	[](const std::vector<std::any>& inputs) -> std::any {	1✔
116	return std::make_any<std::vector<std::any>>(inputs);	1✔
117	});
118
119	return combiner;	1✔
UNCOV 120	}	×
121
122	std::shared_ptr<Task> Task::operator^(std::shared_ptr<Task> tap_task) {	×
123	if (!tap_task) {	×
124	throw PipelineError(PipelineError::VALIDATION_ERROR,	×
125	"Tap task cannot be null");	×
126	}
127	auto passthrough = make_task(	×
128	[](CoroScope&, const std::any& input) -> coro::CoroTask<std::any> {	×
129	co_return input;
130	},	×
131	"TAP_passthrough");	×
132	passthrough->depends_on(shared_from_this());	×
133	tap_task->depends_on(shared_from_this());	×
134	return passthrough;	×
135	}	×
136
137	} // namespace dftracer::utils

llnl / dftracer-utils / 26195612357

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