28982597416

Committed 08 Jul 2026 11:22PM UTC coverage: 50.709% (-1.9%) from 52.577%

Build # 28982597416

Build Type

Pull #87

github

Committed by

web-flow

Commit Message

Merge fdf73a826 into 4908d9921

Pull Request Pull Request #87: fix(comparator): fix wrong counting files

Coverage Stats

16259 of 43579 branches covered (37.31%)

Branch coverage included in aggregate %.

37 of 38 new or added lines in 4 files covered. (97.37%)

616 existing lines in 111 files now uncovered.

21634 of 31148 relevant lines covered (69.46%)

13117.41 hits per line

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

43.75

/src/dftracer/utils/server/trace_index.cpp

#include <dftracer/utils/core/common/filesystem.h>
#include <dftracer/utils/core/common/logging.h>
#include <dftracer/utils/core/coro/channel.h>
#include <dftracer/utils/core/io/io_backend.h>
#include <dftracer/utils/core/pipeline/pipeline.h>
#include <dftracer/utils/core/pipeline/pipeline_config.h>
#include <dftracer/utils/core/tasks/coro_scope.h>
#include <dftracer/utils/core/tasks/task.h>
#include <dftracer/utils/server/router.h>
#include <dftracer/utils/server/trace_index.h>
#include <dftracer/utils/utilities/composites/dft/internal/utils.h>
#include <dftracer/utils/utilities/composites/dft/metadata_collector_utility.h>
#include <dftracer/utils/utilities/filesystem/pattern_directory_scanner_utility.h>
#include <dftracer/utils/utilities/indexer/index_builder_utility.h>
#include <dftracer/utils/utilities/indexer/index_database.h>
#include <dftracer/utils/utilities/indexer/internal/helpers.h>

#include <cinttypes>
#include <limits>

namespace dftracer::utils::server {

using namespace dftracer::utils::utilities::composites::dft;
using namespace dftracer::utils::utilities::composites::dft::indexing;
using namespace dftracer::utils::utilities::filesystem;
namespace indexer = dftracer::utils::utilities::indexer;

TraceIndex::TraceIndex(const std::string& directory,
                       const std::string& index_dir, std::size_t max_concurrent)
    : directory_(directory),
      index_dir_(index_dir),
      max_concurrent_(max_concurrent == 0 ? 8 : max_concurrent) {}

coro::CoroTask<void> TraceIndex::initialize() {
    PatternDirectoryScannerUtility scanner;
    PatternDirectoryScannerUtilityInput scan_input{
        directory_, {".pfw", ".pfw.gz"}, false};
    auto entries = co_await scanner.process(scan_input);

    files_.clear();
    path_to_index_.clear();
    files_.reserve(entries.size());

    global_min_ts_ = std::numeric_limits<std::uint64_t>::max();
    global_max_ts_ = 0;

    std::vector<std::size_t> needs_build;
    std::vector<std::size_t> large_files;

    for (const auto& entry : entries) {
        FileInfo info;
        info.path = entry.path.string();
        info.index_path = internal::determine_index_path(info.path, index_dir_);

        std::error_code ec;
        auto fsize = fs::file_size(info.path, ec);
        info.compressed_size = (!ec && fsize > 0) ? fsize : 0;

        std::size_t idx = files_.size();
        path_to_index_[info.path] = idx;

        info.has_bloom_data = fs::exists(info.index_path);
        info.has_checkpoint_index = fs::exists(info.index_path);
        if (!info.has_bloom_data) {
            needs_build.push_back(idx);
        } else {
            large_files.push_back(idx);
        }

        files_.push_back(std::move(info));
    }

    if (!needs_build.empty() || !large_files.empty()) {
        auto pipeline_config =
            PipelineConfig()
                .with_name("TraceIndex Init")
                .with_compute_threads(max_concurrent_)
                .with_watchdog(false)
                .with_global_timeout(std::chrono::seconds(0))
                .with_task_timeout(std::chrono::seconds(0))
                .with_io_backend(io::IoBackendType::THREADPOOL)
                .with_io_batch_size(1);

        Pipeline pipeline(pipeline_config);

        auto* files_ptr = &files_;
        auto* needs_build_ptr = &needs_build;
        auto* large_files_ptr = &large_files;
        auto* global_min_ts_ptr = &global_min_ts_;
        auto* global_max_ts_ptr = &global_max_ts_;
        std::string index_dir = index_dir_;
        std::size_t max_concurrent = max_concurrent_;

        auto init_task = make_task(
            [files_ptr, needs_build_ptr, large_files_ptr, global_min_ts_ptr,
             global_max_ts_ptr, index_dir,
             max_concurrent](CoroScope& ctx) -> coro::CoroTask<void> {
                if (!needs_build_ptr->empty()) {
                    DFTRACER_UTILS_LOG_INFO(
                        "TraceIndex: building index for %zu file(s) ...",
                        needs_build_ptr->size());

                    auto file_chan =
                        coro::make_channel<std::size_t>(max_concurrent * 2);

                    const auto* index_dir_ptr = &index_dir;
                    co_await ctx.scope([&file_chan, files_ptr, needs_build_ptr,
                                        index_dir_ptr,
                                        max_concurrent](CoroScope& scope)
                                           -> coro::CoroTask<void> {
                        scope.spawn(
                            [ch = file_chan->producer(), needs_build_ptr](
                                CoroScope&) mutable -> coro::CoroTask<void> {
                                auto guard = ch.guard();
                                for (auto idx : *needs_build_ptr) {
                                    if (!co_await ch.send(idx)) co_return;
                                }
                                co_return;
                            });

                        for (std::size_t w = 0; w < max_concurrent; ++w) {
                            scope.spawn([ch = file_chan->consumer(), files_ptr,
                                         index_dir_ptr](CoroScope&)
                                            -> coro::CoroTask<void> {
                                while (auto fi_opt = co_await ch.receive()) {
                                    std::size_t fi = *fi_opt;
                                    auto* info = &(*files_ptr)[fi];

                                    indexer::IndexBuilderUtility builder;
                                    auto config =
                                        indexer::IndexBuildConfig::for_file(
                                            info->path)
                                            .with_index_dir(*index_dir_ptr);
                                    auto result =
                                        co_await builder.process(config);

                                    if (result.success) {
                                        info->index_path =
                                            internal::determine_index_path(
                                                info->path, *index_dir_ptr);
                                        info->has_bloom_data = true;
                                        info->has_checkpoint_index =
                                            fs::exists(info->index_path);
                                    } else {
                                        DFTRACER_UTILS_LOG_WARN(
                                            "TraceIndex: failed to "
                                            "index %s: %s",
                                            info->path.c_str(),
                                            result.error_message.c_str());
                                    }
                                }
                                co_return;
                            });
                        }
                        co_return;
                    });

                    for (auto idx : *needs_build_ptr) {
                        if ((*files_ptr)[idx].has_bloom_data) {
                            large_files_ptr->push_back(idx);
                        }
                    }
                }

                if (!large_files_ptr->empty()) {
                    auto meta_chan =
                        coro::make_channel<std::size_t>(max_concurrent * 2);

                    co_await ctx.scope([&meta_chan, files_ptr, large_files_ptr,
                                        max_concurrent](CoroScope& scope)
                                           -> coro::CoroTask<void> {
                        scope.spawn(
                            [ch = meta_chan->producer(), large_files_ptr](
                                CoroScope&) mutable -> coro::CoroTask<void> {
                                auto guard = ch.guard();
                                for (auto idx : *large_files_ptr) {
                                    if (!co_await ch.send(idx)) co_return;
                                }
                                co_return;
                            });

                        for (std::size_t w = 0; w < max_concurrent; ++w) {
                            scope.spawn([ch = meta_chan->consumer(),
                                         files_ptr](CoroScope&)
                                            -> coro::CoroTask<void> {
                                while (auto fi_opt = co_await ch.receive()) {
                                    std::size_t fi = *fi_opt;
                                    auto* info = &(*files_ptr)[fi];

                                    if (info->has_bloom_data) {
                                        try {
                                            indexer::IndexDatabase idx_db(
                                                info->index_path);
                                            auto logical = indexer::internal::
                                                get_logical_path(info->path);
                                            int fid = idx_db.get_file_info_id(
                                                logical);
                                            if (fid >= 0) {
                                                auto bounds =
                                                    idx_db.query_time_bounds(
                                                        fid);
                                                if (bounds.valid) {
                                                    info->min_timestamp_us =
                                                        bounds.min_timestamp_us;
                                                    info->max_timestamp_us =
                                                        bounds.max_timestamp_us;
                                                }
                                            }
                                        } catch (const std::exception& e) {
                                            DFTRACER_UTILS_LOG_WARN(
                                                "TraceIndex: failed to "
                                                "read time bounds from "
                                                "%s: %s",
                                                info->index_path.c_str(),
                                                e.what());
                                        }
                                    }

                                    auto meta_input =
                                        MetadataCollectorUtilityInput::
                                            from_file(info->path)
                                                .with_index(info->index_path);
                                    auto metadata =
                                        co_await MetadataCollectorUtility{}
                                            .process(meta_input);
                                    if (metadata.success) {
                                        info->uncompressed_size =
                                            metadata.uncompressed_size;
                                        info->num_checkpoints =
                                            metadata.num_checkpoints;
                                        info->checkpoint_size =
                                            metadata.checkpoint_size;
                                        info->compressed_size =
                                            metadata.compressed_size;
                                        info->num_lines = metadata.num_lines;
                                        info->size_mb = metadata.size_mb;
                                    }
                                }
                                co_return;
                            });
                        }
                        co_return;
                    });

                    for (auto fi : *large_files_ptr) {
                        const auto& info = (*files_ptr)[fi];
                        if (info.min_timestamp_us > 0 &&
                            info.min_timestamp_us < *global_min_ts_ptr)
                            *global_min_ts_ptr = info.min_timestamp_us;
                        if (info.max_timestamp_us > *global_max_ts_ptr)
                            *global_max_ts_ptr = info.max_timestamp_us;
                    }
                }

                co_return;
            },
            "TraceIndexInit");

        pipeline.set_source(init_task);
        pipeline.set_destination(init_task);
        pipeline.execute();
    }

    DFTRACER_UTILS_LOG_INFO("TraceIndex: found %zu trace files in %s",
                            files_.size(), directory_.c_str());
    if (global_max_ts_ > 0) {
        DFTRACER_UTILS_LOG_INFO("TraceIndex: global time range [%" PRIu64
                                ", %" PRIu64 "] us",
                                global_min_ts_, global_max_ts_);
    }
}

const TraceIndex::FileInfo* TraceIndex::find_file(
    const std::string& path) const {
    auto it = path_to_index_.find(path);
    if (it == path_to_index_.end()) return nullptr;
    return &files_[it->second];
}

const TraceIndex::FileInfo* TraceIndex::file_at(std::size_t index) const {
    if (index >= files_.size()) return nullptr;
    return &files_[index];
}

std::vector<const TraceIndex::FileInfo*> collect_candidate_files(
    TraceIndex& index, const QueryParams& params) {
    std::vector<const TraceIndex::FileInfo*> files;
    auto file_param = params.get("file");
    if (!file_param.empty()) {
        auto* f = index.find_file(std::string(file_param));
        if (f) files.push_back(f);
    } else {
        for (const auto& f : index.files()) {
            files.push_back(&f);
        }
    }
    return files;
}

}  // namespace dftracer::utils::server

1	#include <dftracer/utils/core/common/filesystem.h>
2	#include <dftracer/utils/core/common/logging.h>
3	#include <dftracer/utils/core/coro/channel.h>
4	#include <dftracer/utils/core/io/io_backend.h>
5	#include <dftracer/utils/core/pipeline/pipeline.h>
6	#include <dftracer/utils/core/pipeline/pipeline_config.h>
7	#include <dftracer/utils/core/tasks/coro_scope.h>
8	#include <dftracer/utils/core/tasks/task.h>
9	#include <dftracer/utils/server/router.h>
10	#include <dftracer/utils/server/trace_index.h>
11	#include <dftracer/utils/utilities/composites/dft/internal/utils.h>
12	#include <dftracer/utils/utilities/composites/dft/metadata_collector_utility.h>
13	#include <dftracer/utils/utilities/filesystem/pattern_directory_scanner_utility.h>
14	#include <dftracer/utils/utilities/indexer/index_builder_utility.h>
15	#include <dftracer/utils/utilities/indexer/index_database.h>
16	#include <dftracer/utils/utilities/indexer/internal/helpers.h>
17
18	#include <cinttypes>
19	#include <limits>
20
21	namespace dftracer::utils::server {
22
23	using namespace dftracer::utils::utilities::composites::dft;
24	using namespace dftracer::utils::utilities::composites::dft::indexing;
25	using namespace dftracer::utils::utilities::filesystem;
26	namespace indexer = dftracer::utils::utilities::indexer;
27
28	TraceIndex::TraceIndex(const std::string& directory,	8✔
29	const std::string& index_dir, std::size_t max_concurrent)	8✔
30	: directory_(directory),	8✔
31	index_dir_(index_dir),	8!
32	max_concurrent_(max_concurrent == 0 ? 8 : max_concurrent) {}	8!
33
34	coro::CoroTask<void> TraceIndex::initialize() {	7!
35	PatternDirectoryScannerUtility scanner;
36	PatternDirectoryScannerUtilityInput scan_input{
37	directory_, {".pfw", ".pfw.gz"}, false};
38	auto entries = co_await scanner.process(scan_input);
39
40	files_.clear();
41	path_to_index_.clear();
42	files_.reserve(entries.size());
43
44	global_min_ts_ = std::numeric_limits<std::uint64_t>::max();
45	global_max_ts_ = 0;
46
47	std::vector<std::size_t> needs_build;
48	std::vector<std::size_t> large_files;
49
50	for (const auto& entry : entries) {
51	FileInfo info;
52	info.path = entry.path.string();
53	info.index_path = internal::determine_index_path(info.path, index_dir_);
54
55	std::error_code ec;
56	auto fsize = fs::file_size(info.path, ec);
57	info.compressed_size = (!ec && fsize > 0) ? fsize : 0;
58
59	std::size_t idx = files_.size();
60	path_to_index_[info.path] = idx;
61
62	info.has_bloom_data = fs::exists(info.index_path);
63	info.has_checkpoint_index = fs::exists(info.index_path);
64	if (!info.has_bloom_data) {
65	needs_build.push_back(idx);
66	} else {
67	large_files.push_back(idx);
68	}
69
70	files_.push_back(std::move(info));
71	}
72
73	if (!needs_build.empty() \|\| !large_files.empty()) {
74	auto pipeline_config =
75	PipelineConfig()
76	.with_name("TraceIndex Init")
77	.with_compute_threads(max_concurrent_)
78	.with_watchdog(false)
79	.with_global_timeout(std::chrono::seconds(0))
80	.with_task_timeout(std::chrono::seconds(0))
81	.with_io_backend(io::IoBackendType::THREADPOOL)
82	.with_io_batch_size(1);
83
84	Pipeline pipeline(pipeline_config);
85
86	auto* files_ptr = &files_;
87	auto* needs_build_ptr = &needs_build;
88	auto* large_files_ptr = &large_files;
89	auto* global_min_ts_ptr = &global_min_ts_;
90	auto* global_max_ts_ptr = &global_max_ts_;
91	std::string index_dir = index_dir_;
92	std::size_t max_concurrent = max_concurrent_;
93
94	auto init_task = make_task(
95	[files_ptr, needs_build_ptr, large_files_ptr, global_min_ts_ptr,	6!
96	global_max_ts_ptr, index_dir,
97	max_concurrent](CoroScope& ctx) -> coro::CoroTask<void> {
98	if (!needs_build_ptr->empty()) {
99	DFTRACER_UTILS_LOG_INFO(
100	"TraceIndex: building index for %zu file(s) ...",
101	needs_build_ptr->size());
102
103	auto file_chan =
104	coro::make_channel<std::size_t>(max_concurrent * 2);
105
106	const auto* index_dir_ptr = &index_dir;
107	co_await ctx.scope([&file_chan, files_ptr, needs_build_ptr,	6!
108	index_dir_ptr,
109	max_concurrent](CoroScope& scope)
110	-> coro::CoroTask<void> {
111	scope.spawn(
112	[ch = file_chan->producer(), needs_build_ptr](	6!
113	CoroScope&) mutable -> coro::CoroTask<void> {
114	auto guard = ch.guard();
115	for (auto idx : *needs_build_ptr) {
116	if (!co_await ch.send(idx)) co_return;
117	}
118	co_return;
119	});	12!
120
121	for (std::size_t w = 0; w < max_concurrent; ++w) {
122	scope.spawn([ch = file_chan->consumer(), files_ptr,	36!
123	index_dir_ptr](CoroScope&)
124	-> coro::CoroTask<void> {
125	while (auto fi_opt = co_await ch.receive()) {
126	std::size_t fi = *fi_opt;
127	auto* info = &(*files_ptr)[fi];
128
129	indexer::IndexBuilderUtility builder;
130	auto config =
131	indexer::IndexBuildConfig::for_file(
132	info->path)
133	.with_index_dir(*index_dir_ptr);
134	auto result =
135	co_await builder.process(config);
136
137	if (result.success) {
138	info->index_path =
139	internal::determine_index_path(
140	info->path, *index_dir_ptr);
141	info->has_bloom_data = true;
142	info->has_checkpoint_index =
143	fs::exists(info->index_path);
144	} else {
145	DFTRACER_UTILS_LOG_WARN(
146	"TraceIndex: failed to "
147	"index %s: %s",
148	info->path.c_str(),
149	result.error_message.c_str());
150	}
151	}
152	co_return;
153	});	72!
154	}
155	co_return;
156	});	12!
157
158	for (auto idx : *needs_build_ptr) {
159	if ((*files_ptr)[idx].has_bloom_data) {
160	large_files_ptr->push_back(idx);
161	}
162	}
163	}
164
165	if (!large_files_ptr->empty()) {
166	auto meta_chan =
167	coro::make_channel<std::size_t>(max_concurrent * 2);
168
169	co_await ctx.scope([&meta_chan, files_ptr, large_files_ptr,	6!
170	max_concurrent](CoroScope& scope)
171	-> coro::CoroTask<void> {
172	scope.spawn(
173	[ch = meta_chan->producer(), large_files_ptr](	6!
174	CoroScope&) mutable -> coro::CoroTask<void> {
175	auto guard = ch.guard();
176	for (auto idx : *large_files_ptr) {
177	if (!co_await ch.send(idx)) co_return;
178	}
179	co_return;
180	});	12!
181
182	for (std::size_t w = 0; w < max_concurrent; ++w) {
183	scope.spawn([ch = meta_chan->consumer(),	34!
184	files_ptr](CoroScope&)
185	-> coro::CoroTask<void> {
186	while (auto fi_opt = co_await ch.receive()) {
187	std::size_t fi = *fi_opt;
188	auto* info = &(*files_ptr)[fi];
189
190	if (info->has_bloom_data) {
191	try {
192	indexer::IndexDatabase idx_db(
193	info->index_path);
194	auto logical = indexer::internal::
195	get_logical_path(info->path);
196	int fid = idx_db.get_file_info_id(
197	logical);
198	if (fid >= 0) {
199	auto bounds =
200	idx_db.query_time_bounds(
201	fid);
202	if (bounds.valid) {
203	info->min_timestamp_us =
204	bounds.min_timestamp_us;
205	info->max_timestamp_us =
206	bounds.max_timestamp_us;
207	}
208	}
209	} catch (const std::exception& e) {
210	DFTRACER_UTILS_LOG_WARN(
211	"TraceIndex: failed to "
212	"read time bounds from "
213	"%s: %s",
214	info->index_path.c_str(),
215	e.what());
216	}
217	}
218
219	auto meta_input =
220	MetadataCollectorUtilityInput::
221	from_file(info->path)
222	.with_index(info->index_path);
223	auto metadata =
224	co_await MetadataCollectorUtility{}
225	.process(meta_input);
226	if (metadata.success) {
227	info->uncompressed_size =
228	metadata.uncompressed_size;
229	info->num_checkpoints =
230	metadata.num_checkpoints;
231	info->checkpoint_size =
232	metadata.checkpoint_size;
233	info->compressed_size =
234	metadata.compressed_size;
235	info->num_lines = metadata.num_lines;
236	info->size_mb = metadata.size_mb;
237	}
238	}
239	co_return;
240	});	68!
241	}
242	co_return;
243	});	12!
244
245	for (auto fi : *large_files_ptr) {
246	const auto& info = (*files_ptr)[fi];
247	if (info.min_timestamp_us > 0 &&
248	info.min_timestamp_us < *global_min_ts_ptr)
249	*global_min_ts_ptr = info.min_timestamp_us;
250	if (info.max_timestamp_us > *global_max_ts_ptr)
251	*global_max_ts_ptr = info.max_timestamp_us;
252	}
253	}
254
255	co_return;
256	},	12!
257	"TraceIndexInit");
258
259	pipeline.set_source(init_task);
260	pipeline.set_destination(init_task);
261	pipeline.execute();
262	}
263
264	DFTRACER_UTILS_LOG_INFO("TraceIndex: found %zu trace files in %s",
265	files_.size(), directory_.c_str());
266	if (global_max_ts_ > 0) {
267	DFTRACER_UTILS_LOG_INFO("TraceIndex: global time range [%" PRIu64
268	", %" PRIu64 "] us",
269	global_min_ts_, global_max_ts_);
270	}
271	}	14!
272
273	const TraceIndex::FileInfo* TraceIndex::find_file(	3✔
274	const std::string& path) const {
275	auto it = path_to_index_.find(path);	3!
276	if (it == path_to_index_.end()) return nullptr;	3✔
277	return &files_[it->second];	2✔
278	}
279
280	const TraceIndex::FileInfo* TraceIndex::file_at(std::size_t index) const {	2✔
281	if (index >= files_.size()) return nullptr;	2✔
282	return &files_[index];	1✔
283	}
284
285	std::vector<const TraceIndex::FileInfo*> collect_candidate_files(	8✔
286	TraceIndex& index, const QueryParams& params) {
287	std::vector<const TraceIndex::FileInfo*> files;	8✔
288	auto file_param = params.get("file");	8!
289	if (!file_param.empty()) {	8!
290	auto* f = index.find_file(std::string(file_param));	×
291	if (f) files.push_back(f);	×
292	} else {
293	for (const auto& f : index.files()) {	16✔
294	files.push_back(&f);	8!
295	}
296	}
297	return files;	16✔
UNCOV 298	}	×
299
300	} // namespace dftracer::utils::server

llnl / dftracer-utils / 28982597416

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