29787659139

Committed 20 Jul 2026 11:34PM UTC coverage: 51.578% (-1.1%) from 52.66%

Build # 29787659139

Build Type

Pull #99

github

Committed by

web-flow

Commit Message

Merge ee805adeb into 06bc84ec9

Pull Request Pull Request #99: Support CM time_metric (NS/MS/SEC/US) across reader and viz

Coverage Stats

34832 of 86278 branches covered (40.37%)

Branch coverage included in aggregate %.

1034 of 1319 new or added lines in 30 files covered. (78.39%)

5193 existing lines in 197 files now uncovered.

35349 of 49791 relevant lines covered (70.99%)

9765.54 hits per line

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

63.51

/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/index_database_writer_context.h>
#include <dftracer/utils/utilities/indexer/internal/helpers.h>
#include <dftracer/utils/utilities/reader/trace_reader.h>

#include <cinttypes>
#include <limits>
#include <memory>
#include <unordered_map>
#include <vector>

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;

    // The reuse decision below is existence-only, so a changed .pfw would be
    // served from a stale index. Drop any (shared) index root whose sources
    // changed since indexing; the loop then rebuilds it as if absent.
    {
        std::unordered_map<std::string, std::vector<std::string>> by_root;
        for (const auto& entry : entries)
            by_root[internal::determine_index_path(entry.path.string(),
                                                   index_dir_)]
                .push_back(entry.path.string());
        for (auto& [root, paths] : by_root) {
            if (!fs::exists(root)) continue;
            bool stale = false;
            try {
                indexer::IndexDatabase db(root);
                stale = db.find_stale_files(paths).stale();
            } catch (const std::exception& e) {
                DFTRACER_UTILS_LOG_WARN(
                    "TraceIndex: stale check failed for %s: %s; rebuilding",
                    root.c_str(), e.what());
                stale = true;
            }
            if (stale) {
                DFTRACER_UTILS_LOG_WARN(
                    "TraceIndex: source changed since indexing; rebuilding %s",
                    root.c_str());
                std::error_code ec;
                fs::remove_all(root, ec);
            }
        }
    }

    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());

                    // Assign file identities up front, single-threaded, so the
                    // concurrent builders below don't race the shared next-id
                    // counter and collide two files onto one id. Hold each
                    // read-write handle open across the build so builders reuse
                    // it instead of hitting a read-only -> read-write upgrade.
                    std::vector<std::unique_ptr<indexer::IndexDatabase>>
                        held_dbs;
                    {
                        std::unordered_map<std::string,
                                           std::vector<std::size_t>>
                            by_index;
                        for (auto idx : *needs_build_ptr)
                            by_index[(*files_ptr)[idx].index_path].push_back(
                                idx);
                        for (auto& [ipath, idxs] : by_index) {
                            try {
                                auto db =
                                    std::make_unique<indexer::IndexDatabase>(
                                        ipath);
                                auto w = db->begin_write();
                                for (auto idx : idxs) {
                                    const auto& p = (*files_ptr)[idx].path;
                                    w->get_or_create_file_info(
                                        indexer::internal::get_logical_path(p),
                                        indexer::internal::calculate_file_hash(
                                            p));
                                }
                                w->commit();
                                held_dbs.push_back(std::move(db));
                            } catch (const std::exception& e) {
                                DFTRACER_UTILS_LOG_WARN(
                                    "TraceIndex: file-id pre-assign failed for "
                                    "%s: %s",
                                    ipath.c_str(), e.what());
                            }
                        }
                    }

                    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();
    }

    // One time unit per trace: resolve it once from the first file's CM.
    if (!files_.empty()) {
        try {
            utilities::reader::TraceReaderConfig cfg;
            cfg.file_path = files_.front().path;
            cfg.index_dir = index_dir_;
            utilities::reader::TraceReader reader(cfg);
            time_metric_ = co_await reader.read_time_metric();
        } catch (const std::exception& e) {
            DFTRACER_UTILS_LOG_WARN(
                "TraceIndex: failed to resolve time_metric: %s", e.what());
        }
        if (time_metric_ != TimeMetric::US) {
            DFTRACER_UTILS_LOG_INFO(
                "TraceIndex: trace time unit is %s (scaling viz to us)",
                std::string(time_metric_to_string(time_metric_)).c_str());
        }
    }

    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/index_database_writer_context.h>
17	#include <dftracer/utils/utilities/indexer/internal/helpers.h>
18	#include <dftracer/utils/utilities/reader/trace_reader.h>
19
20	#include <cinttypes>
21	#include <limits>
22	#include <memory>
23	#include <unordered_map>
24	#include <vector>
25
26	namespace dftracer::utils::server {
27
28	using namespace dftracer::utils::utilities::composites::dft;
29	using namespace dftracer::utils::utilities::composites::dft::indexing;
30	using namespace dftracer::utils::utilities::filesystem;
31	namespace indexer = dftracer::utils::utilities::indexer;
32
33	TraceIndex::TraceIndex(const std::string& directory,	42!
34	const std::string& index_dir, std::size_t max_concurrent)
35	: directory_(directory),	14✔
36	index_dir_(index_dir),	14!
37	max_concurrent_(max_concurrent == 0 ? 8 : max_concurrent) {}	28!
38
39	coro::CoroTask<void> TraceIndex::initialize() {	91!
40	PatternDirectoryScannerUtility scanner;	39!
41	PatternDirectoryScannerUtilityInput scan_input{	78!
42	directory_, {".pfw", ".pfw.gz"}, false};	39!
43	auto entries = co_await scanner.process(scan_input);	52!
44
45	files_.clear();	13✔
46	path_to_index_.clear();	13✔
47	files_.reserve(entries.size());	13!
48
49	global_min_ts_ = std::numeric_limits<std::uint64_t>::max();	13✔
50	global_max_ts_ = 0;	13✔
51
52	std::vector<std::size_t> needs_build;	13✔
53	std::vector<std::size_t> large_files;	13✔
54
55	// The reuse decision below is existence-only, so a changed .pfw would be
56	// served from a stale index. Drop any (shared) index root whose sources
57	// changed since indexing; the loop then rebuilds it as if absent.
58	{
59	std::unordered_map<std::string, std::vector<std::string>> by_root;	13✔
60	for (const auto& entry : entries)	27✔
61	by_root[internal::determine_index_path(entry.path.string(),	28!
62	index_dir_)]	14✔
63	.push_back(entry.path.string());	14!
64	for (auto& [root, paths] : by_root) {	25!
65	if (!fs::exists(root)) continue;	12!
66	bool stale = false;	1✔
67	try {
68	indexer::IndexDatabase db(root);	1!
69	stale = db.find_stale_files(paths).stale();	1!
70	} catch (const std::exception& e) {	1!
UNCOV 71	DFTRACER_UTILS_LOG_WARN(	×
72	"TraceIndex: stale check failed for %s: %s; rebuilding",
73	root.c_str(), e.what());
UNCOV 74	stale = true;	×
UNCOV 75	}	×
76	if (stale) {	1!
77	DFTRACER_UTILS_LOG_WARN(	1!
78	"TraceIndex: source changed since indexing; rebuilding %s",
79	root.c_str());
80	std::error_code ec;	1✔
81	fs::remove_all(root, ec);	1!
82	}	1✔
83	}	12✔
84	}	13✔
85
86	for (const auto& entry : entries) {	27✔
87	FileInfo info;	14✔
88	info.path = entry.path.string();	14!
89	info.index_path = internal::determine_index_path(info.path, index_dir_);	14!
90
91	std::error_code ec;	14✔
92	auto fsize = fs::file_size(info.path, ec);	14!
93	info.compressed_size = (!ec && fsize > 0) ? fsize : 0;	14!
94
95	std::size_t idx = files_.size();	14✔
96	path_to_index_[info.path] = idx;	14!
97
98	info.has_bloom_data = fs::exists(info.index_path);	14!
99	info.has_checkpoint_index = fs::exists(info.index_path);	14!
100	if (!info.has_bloom_data) {	14!
101	needs_build.push_back(idx);	14!
102	} else {	14✔
UNCOV 103	large_files.push_back(idx);	×
104	}
105
106	files_.push_back(std::move(info));	14!
107	}	14✔
108
109	if (!needs_build.empty() \|\| !large_files.empty()) {	13✔
110	auto pipeline_config =	36✔
111	PipelineConfig()	48!
112	.with_name("TraceIndex Init")	12!
113	.with_compute_threads(max_concurrent_)	12!
114	.with_watchdog(false)	12!
115	.with_global_timeout(std::chrono::seconds(0))	12!
116	.with_task_timeout(std::chrono::seconds(0))	12!
117	.with_io_backend(io::IoBackendType::THREADPOOL)	12!
118	.with_io_batch_size(1);	12!
119
120	Pipeline pipeline(pipeline_config);	12!
121
122	auto* files_ptr = &files_;	12✔
123	auto* needs_build_ptr = &needs_build;	12✔
124	auto* large_files_ptr = &large_files;	12✔
125	auto* global_min_ts_ptr = &global_min_ts_;	12✔
126	auto* global_max_ts_ptr = &global_max_ts_;	12✔
127	std::string index_dir = index_dir_;	12!
128	std::size_t max_concurrent = max_concurrent_;	12✔
129
130	auto init_task = make_task(	12!
131	[files_ptr, needs_build_ptr, large_files_ptr, global_min_ts_ptr,	132!
132	global_max_ts_ptr, index_dir,	24!
133	max_concurrent](CoroScope& ctx) -> coro::CoroTask<void> {	24!
134	if (!needs_build_ptr->empty()) {	60✔
135	DFTRACER_UTILS_LOG_INFO(	36!
136	"TraceIndex: building index for %zu file(s) ...",
137	needs_build_ptr->size());
138
139	// Assign file identities up front, single-threaded, so the
140	// concurrent builders below don't race the shared next-id
141	// counter and collide two files onto one id. Hold each
142	// read-write handle open across the build so builders reuse
143	// it instead of hitting a read-only -> read-write upgrade.
144	std::vector<std::unique_ptr<indexer::IndexDatabase>>	36✔
145	held_dbs;	36✔
146	{
147	std::unordered_map<std::string,	36✔
148	std::vector<std::size_t>>
149	by_index;	36✔
150	for (auto idx : *needs_build_ptr)	50✔
151	by_index[(*files_ptr)[idx].index_path].push_back(	14!
152	idx);	14✔
153	for (auto& [ipath, idxs] : by_index) {	48!
154	try {
155	auto db =	12✔
156	std::make_unique<indexer::IndexDatabase>(	12!
157	ipath);	12✔
158	auto w = db->begin_write();	12!
159	for (auto idx : idxs) {	26✔
160	const auto& p = (*files_ptr)[idx].path;	14✔
161	w->get_or_create_file_info(	28!
162	indexer::internal::get_logical_path(p),	14!
163	indexer::internal::calculate_file_hash(	14!
164	p));	14✔
165	}	14✔
166	w->commit();	12!
167	held_dbs.push_back(std::move(db));	12!
168	} catch (const std::exception& e) {	12!
UNCOV 169	DFTRACER_UTILS_LOG_WARN(	×
170	"TraceIndex: file-id pre-assign failed for "
171	"%s: %s",
172	ipath.c_str(), e.what());
UNCOV 173	}	×
174	}	12✔
175	}	36✔
176
177	auto file_chan =	36✔
178	coro::make_channel<std::size_t>(max_concurrent * 2);	36!
179
180	const auto* index_dir_ptr = &index_dir;	36✔
181	co_await ctx.scope([&file_chan, files_ptr, needs_build_ptr,	192!
182	index_dir_ptr,	36✔
183	max_concurrent](CoroScope& scope)	36✔
184	-> coro::CoroTask<void> {	12!
185	scope.spawn(	12!
186	[ch = file_chan->producer(), needs_build_ptr](	128!
187	CoroScope&) mutable -> coro::CoroTask<void> {	12!
188	auto guard = ch.guard();	12!
189	for (auto idx : *needs_build_ptr) {	82✔
190	if (!co_await ch.send(idx)) co_return;	68!
191	}	14✔
192	co_return;	12✔
193	});	96✔
194
195	for (std::size_t w = 0; w < max_concurrent; ++w) {	60✔
196	scope.spawn([ch = file_chan->consumer(), files_ptr,	291!
197	index_dir_ptr](CoroScope&)	48✔
198	-> coro::CoroTask<void> {	47✔
199	while (auto fi_opt = co_await ch.receive()) {	148!
200	std::size_t fi = *fi_opt;	42✔
201	auto* info = &(*files_ptr)[fi];	42✔
202
203	indexer::IndexBuilderUtility builder;	42!
204	auto config =	42✔
205	indexer::IndexBuildConfig::for_file(	126!
206	info->path)	42✔
207	.with_index_dir(*index_dir_ptr);	42✔
208	auto result =	42✔
209	co_await builder.process(config);	56!
210
211	if (result.success) {	14!
212	info->index_path =	14✔
213	internal::determine_index_path(	28!
214	info->path, *index_dir_ptr);	14✔
215	info->has_bloom_data = true;	14✔
216	info->has_checkpoint_index =	14✔
217	fs::exists(info->index_path);	14!
218	} else {	14✔
UNCOV 219	DFTRACER_UTILS_LOG_WARN(	×
220	"TraceIndex: failed to "
221	"index %s: %s",
222	info->path.c_str(),
223	result.error_message.c_str());
224	}
225	}	90!
226	co_return;	48✔
227	});	138✔
228	}	48✔
229	co_return;	24✔
UNCOV 230	});	×
231
232	for (auto idx : *needs_build_ptr) {	26✔
233	if ((*files_ptr)[idx].has_bloom_data) {	14!
234	large_files_ptr->push_back(idx);	14!
235	}	14✔
236	}	14✔
237	}	12!
238
239	if (!large_files_ptr->empty()) {	36!
240	auto meta_chan =	36✔
241	coro::make_channel<std::size_t>(max_concurrent * 2);	36!
242
243	co_await ctx.scope([&meta_chan, files_ptr, large_files_ptr,	144!
244	max_concurrent](CoroScope& scope)	36✔
245	-> coro::CoroTask<void> {	12!
246	scope.spawn(	12!
247	[ch = meta_chan->producer(), large_files_ptr](	128!
248	CoroScope&) mutable -> coro::CoroTask<void> {	12!
249	auto guard = ch.guard();	12!
250	for (auto idx : *large_files_ptr) {	82✔
251	if (!co_await ch.send(idx)) co_return;	68!
252	}	14✔
253	co_return;	12✔
254	});	96✔
255
256	for (std::size_t w = 0; w < max_concurrent; ++w) {	60✔
257	scope.spawn([ch = meta_chan->consumer(),	352!
258	files_ptr](CoroScope&)	48✔
259	-> coro::CoroTask<void> {	48!
260	while (auto fi_opt = co_await ch.receive()) {	200!
261	std::size_t fi = *fi_opt;	42✔
262	auto* info = &(*files_ptr)[fi];	42✔
263
264	if (info->has_bloom_data) {	42✔
265	try {
266	indexer::IndexDatabase idx_db(	28!
267	info->index_path);	14✔
268	auto logical = indexer::internal::	28!
269	get_logical_path(info->path);	14✔
270	int fid = idx_db.get_file_info_id(	28!
271	logical);	14✔
272	if (fid >= 0) {	14!
273	auto bounds =	14✔
274	idx_db.query_time_bounds(	14!
275	fid);	14✔
276	if (bounds.valid) {	14!
277	info->min_timestamp_us =	14✔
278	bounds.min_timestamp_us;	14✔
279	info->max_timestamp_us =	14✔
280	bounds.max_timestamp_us;	14✔
281	}	14✔
282	}	14✔
283	} catch (const std::exception& e) {	14!
UNCOV 284	DFTRACER_UTILS_LOG_WARN(	×
285	"TraceIndex: failed to "
286	"read time bounds from "
287	"%s: %s",
288	info->index_path.c_str(),
289	e.what());
UNCOV 290	}	×
291	}	14✔
292
293	auto meta_input =	42✔
294	MetadataCollectorUtilityInput::	84!
295	from_file(info->path)	42!
296	.with_index(info->index_path);	42!
297	auto metadata =	42✔
298	co_await MetadataCollectorUtility{}	98!
299	.process(meta_input);	42!
300	if (metadata.success) {	14!
301	info->uncompressed_size =	14✔
302	metadata.uncompressed_size;	14✔
303	info->num_checkpoints =	14✔
304	metadata.num_checkpoints;	14✔
305	info->checkpoint_size =	14✔
306	metadata.checkpoint_size;	14✔
307	info->compressed_size =	14✔
308	metadata.compressed_size;	14✔
309	info->num_lines = metadata.num_lines;	14✔
310	info->size_mb = metadata.size_mb;	14✔
311	}	14✔
312	}	90!
313	co_return;	48✔
314	});	204✔
315	}	48✔
316	co_return;	24✔
UNCOV 317	});	×
318
319	for (auto fi : *large_files_ptr) {	26✔
320	const auto& info = (*files_ptr)[fi];	14✔
321	if (info.min_timestamp_us > 0 &&	14!
322	info.min_timestamp_us < *global_min_ts_ptr)	14✔
323	*global_min_ts_ptr = info.min_timestamp_us;	12✔
324	if (info.max_timestamp_us > *global_max_ts_ptr)	14✔
325	*global_max_ts_ptr = info.max_timestamp_us;	13✔
326	}	14✔
327	}	12!
328
329	co_return;	12✔
330	},	48✔
331	"TraceIndexInit");	12!
332
333	pipeline.set_source(init_task);	12!
334	pipeline.set_destination(init_task);	12!
335	pipeline.execute();	12!
336	}	12✔
337
338	// One time unit per trace: resolve it once from the first file's CM.
339	if (!files_.empty()) {	37✔
340	try {
341	utilities::reader::TraceReaderConfig cfg;	36✔
342	cfg.file_path = files_.front().path;	36✔
343	cfg.index_dir = index_dir_;	12✔
344	utilities::reader::TraceReader reader(cfg);	36!
345	time_metric_ = co_await reader.read_time_metric();	48!
346	} catch (const std::exception& e) {	12!
NEW 347	DFTRACER_UTILS_LOG_WARN(	×
348	"TraceIndex: failed to resolve time_metric: %s", e.what());
NEW 349	}	×
350	if (time_metric_ != TimeMetric::US) {	12✔
351	DFTRACER_UTILS_LOG_INFO(	1!
352	"TraceIndex: trace time unit is %s (scaling viz to us)",
353	std::string(time_metric_to_string(time_metric_)).c_str());
354	}	1✔
355	}	12✔
356
357	DFTRACER_UTILS_LOG_INFO("TraceIndex: found %zu trace files in %s",	13!
358	files_.size(), directory_.c_str());
359	if (global_max_ts_ > 0) {	13✔
360	DFTRACER_UTILS_LOG_INFO("TraceIndex: global time range [%" PRIu64	12!
361	", %" PRIu64 "] us",
362	global_min_ts_, global_max_ts_);
363	}	12✔
364	}	163!
365
366	const TraceIndex::FileInfo* TraceIndex::find_file(	3✔
367	const std::string& path) const {
368	auto it = path_to_index_.find(path);	3✔
369	if (it == path_to_index_.end()) return nullptr;	3✔
370	return &files_[it->second];	2✔
371	}	3✔
372
373	const TraceIndex::FileInfo* TraceIndex::file_at(std::size_t index) const {	2✔
374	if (index >= files_.size()) return nullptr;	2✔
375	return &files_[index];	1✔
376	}	2✔
377
378	std::vector<const TraceIndex::FileInfo*> collect_candidate_files(	18✔
379	TraceIndex& index, const QueryParams& params) {
380	std::vector<const TraceIndex::FileInfo*> files;	18✔
381	auto file_param = params.get("file");	18!
382	if (!file_param.empty()) {	18!
383	auto* f = index.find_file(std::string(file_param));	×
384	if (f) files.push_back(f);	×
UNCOV 385	} else {	×
386	for (const auto& f : index.files()) {	40!
387	files.push_back(&f);	22!
388	}
389	}
390	return files;	18✔
391	}	18!
392
393	} // namespace dftracer::utils::server

llnl / dftracer-utils / 29787659139

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