26043728131

Committed 18 May 2026 03:37PM UTC coverage: 51.706% (-0.4%) from 52.076%

Build # 26043728131

Build Type

push

github

Committed by

hariharan-devarajan

Commit Message

feat(perf): performance improvements for parallel reading, indexing, and aggregation

Indexer
- Streaming parse-and-emit worker pipeline with bounded memory usage
- Concurrent SST artifact ingestion with staging support
- Gzip member slicing for parallel indexing
- Lazy decoding for compressed value counts
- Bypass DOM wrapper for indexer hot path (simdjson on_demand)
- Decoupled write workers from parse workers
- --rebuild-summaries flag and optimized root summary rebuild

Aggregator / MPI
- Task-based DAG execution for aggregator pipeline
- Shared staging for multi-node artifact relocation
- Per-node thread scaling to avoid oversubscription
- Unified distributed aggregation tracking, removed manifest consolidation
- Deterministic aggregation and intra-file parallelism

Trace reader / query
- Compiled predicate evaluation for AND-of-EQ queries
- Uniform-match shortcut for AND-of-EQ queries
- Line-range support for work items and checkpoint processing
- Optimized chunk pruning and checkpoint handling

Replay
- Pipelined replay with coroutines and channels
- JsonParser-based trace processing
- Optimized string handling and i/o buffering

Organize / writer / dft
- Parallel slice creation and merging in organize visitor
- Inline indexer in organize
- Gzip member tracking in writer
- Coroutine-based event dispatcher with extracted parse logic
- Batch flushing in organize visitor

Arrow / call_tree
- Optimized arrow conversion
- Arrow IPC support and improved save/load in call_tree

Build / infrastructure
- zlib-ng option, system simdjson fallback
- cgroup v1/v2 memory limit detection
- Auto-computed per-file memory estimates and batch sizes
- CI: perf branch trigger, formatting

Docs
- Rewritten indexer and trace reader API references

Coverage Stats

35907 of 90345 branches covered (39.74%)

Branch coverage included in aggregate %.

16869 of 21880 new or added lines in 137 files covered. (77.1%)

273 existing lines in 39 files now uncovered.

32021 of 41028 relevant lines covered (78.05%)

13164.29 hits per line

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

37.46

/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) {	1,122✔
14	if (!parent) {	1,122✔
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);	1,122!
21
22	// Add this task as child to parent
23	parent->add_child(shared_from_this());	1,122!
24
25	// Register as reader for smart value release
26	parent->result_.add_reader();	1,122✔
27
28	return shared_from_this();	1,122✔
29	}
30
31	std::shared_ptr<Task> Task::depends_on(	12✔
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) {	38✔
35	depends_on(parent);	26!
36	}
37	return shared_from_this();	12✔
38	}
39
40	std::shared_ptr<Task> Task::with_combiner(	16✔
41	std::function<std::any(const std::vector<std::any>&)> combiner) {
42	input_combiner_ = std::move(combiner);	16✔
43	has_custom_combiner_ = true;	16✔
44	return shared_from_this();	16✔
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,	6,144!
53	const std::any& input) {	773!
54	try {
55	std::any result = co_await func_(context, input);	3,837!
56	co_return result;	759!
57	} catch (const std::exception& e) {	773!
58	DFTRACER_UTILS_LOG_ERROR("Task '%s' execution failed: %s", get_name(),	13!
59	e.what());
60	throw;	13✔
61	}	13!
62	}	3,087!
63
64	void Task::set_result(std::any result) { result_.set_value(std::move(result)); }	1,503!
65
66	void Task::set_exception(std::exception_ptr ex) { result_.set_exception(ex); }	34!
67
68	std::any Task::apply_combiner(const std::vector<std::any>& inputs) const {	20✔
69	if (!has_custom_combiner_) {	20✔
70	throw PipelineError(PipelineError::VALIDATION_ERROR,	×
71	"Task has no custom combiner");	×
72	}
73	return input_combiner_(inputs);	20✔
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) {	2✔
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(	3✔
95	[](CoroScope&,	6!
96	const std::vector<std::any>& inputs) -> coro::CoroTask<std::any> {	1!
97	// inputs[0] is from first parent (this), inputs[1] is from second
98	// (other)
99	if (inputs.size() != 2) {	1!
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],	3!
105	inputs[1]);	1✔
106	},	2!
107	"AND_combiner");	2!
108
109	// Combiner depends on both tasks
110	combiner->depends_on(shared_from_this());	2!
111	combiner->depends_on(other);	2!
112
113	// Set a custom combiner to pass inputs as vector
114	combiner->with_combiner(	2!
115	[](const std::vector<std::any>& inputs) -> std::any {	2✔
116	return std::make_any<std::vector<std::any>>(inputs);	2✔
117	});
118
119	return combiner;	2✔
120	}	1!
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	}
UNCOV 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 / 26043728131

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