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

62.23

/src/dftracer/utils/binaries/dftracer_split.cpp

#include <dftracer/utils/core/common/config.h>
#include <dftracer/utils/core/pipeline/pipeline.h>
#include <dftracer/utils/core/rocksdb/db_manager.h>
#include <dftracer/utils/core/task_graph/task_graph.h>
#include <dftracer/utils/core/tasks/coro_scope.h>
#include <dftracer/utils/core/tasks/task.h>
#include <dftracer/utils/core/utilities/utility_adapter.h>
#include <dftracer/utils/utilities/composites/composites.h>
#include <dftracer/utils/utilities/composites/dft/chunk_extractor_utility.h>
#include <dftracer/utils/utilities/fileio/types/types.h>
#include <dftracer/utils/utilities/indexer/index_builder_utility.h>
#include <dftracer/utils/utilities/indexer/internal/indexer.h>
#include <unistd.h>

#include <chrono>
#include <cinttypes>

#include "common_cli.h"

using namespace dftracer::utils;
using namespace dftracer::utils::task_graph;
using Metadata = utilities::composites::dft::MetadataCollectorUtilityOutput;
using ChunkManifest =
    utilities::composites::dft::internal::DFTracerChunkManifest;
using ExtractInput = utilities::composites::dft::ChunkExtractorUtilityInput;
using ExtractResult = utilities::composites::dft::ChunkExtractorUtilityOutput;

class SplitArgParse : public cli::ArgParse {
   public:
    cli::DirectoryArgs directory;
    cli::PipelineArgs pipeline;
    cli::IndexingArgs indexing;
    cli::WatchdogArgs watchdog;

    std::string app_name = "app";
    std::string output_dir = "./split";
    int chunk_size_mb = 4;
    bool compress = true;
    bool verify = false;

    explicit SplitArgParse(argparse::ArgumentParser& p) : ArgParse(p) {
        indexing.force_help =
            "Override existing files and force index recreation";
        schema(directory, pipeline, indexing, watchdog);
    }

   protected:
    void register_args() override {
        parser()
            .add_argument("-n", "--app-name")
            .help("Application name for output files")
            .default_value<std::string>("app");

        parser()
            .add_argument("-o", "--output")
            .help("Output directory for split files")
            .default_value<std::string>("./split");

        parser()
            .add_argument("-s", "--chunk-size")
            .help("Chunk size in MB")
            .scan<'d', int>()
            .default_value(4);

        parser()
            .add_argument("-c", "--compress")
            .help("Compress output files with gzip")
            .flag()
            .default_value(true);

        parser()
            .add_argument("--verify")
            .help("Verify output chunks match input by comparing event IDs")
            .flag();
    }

    void post_parse() override {
        app_name = parser().get<std::string>("--app-name");
        output_dir = parser().get<std::string>("--output");
        chunk_size_mb = parser().get<int>("--chunk-size");
        compress = parser().get<bool>("--compress");
        verify = parser().get<bool>("--verify");
    }
};

static coro::CoroTask<int> run_split(const SplitArgParse* cli) {
    const auto log_dir = fs::absolute(cli->directory.value).string();
    const auto output_dir = fs::absolute(cli->output_dir).string();
    const auto& app_name = cli->app_name;
    const auto chunk_size_mb = cli->chunk_size_mb;
    const auto force = cli->indexing.force;
    const auto compress = cli->compress;
    const auto verify = cli->verify;
    const auto checkpoint_size = cli->indexing.checkpoint_size;
    const auto executor_threads = cli->pipeline.executor_threads;
    auto index_dir = cli->indexing.index_dir;

    std::string temp_index_dir;
    if (index_dir.empty()) {
        temp_index_dir = fs::temp_directory_path() /
                         ("dftracer_idx_" + std::to_string(std::time(nullptr)) +
                          "_" + std::to_string(getpid()));
        fs::create_directories(temp_index_dir);
        index_dir = temp_index_dir;
        DFTRACER_UTILS_LOG_INFO("Created temporary index directory: %s",
                                index_dir.c_str());
    }

    std::printf("==========================================\n");
    std::printf("DFTracer Split (Explicit Pipeline)\n");
    std::printf("==========================================\n");
    std::printf("Arguments:\n");
    std::printf("  App name: %s\n", app_name.c_str());
    std::printf("  Override: %s\n", force ? "true" : "false");
    std::printf("  Compress: %s\n", compress ? "true" : "false");
    std::printf("  Data dir: %s\n", log_dir.c_str());
    std::printf("  Output dir: %s\n", output_dir.c_str());
    std::printf("  Chunk size: %d MB\n", chunk_size_mb);
    std::printf("  Executor threads: %zu\n", executor_threads);
    std::printf("==========================================\n\n");

    if (!fs::exists(output_dir)) {
        fs::create_directories(output_dir);
    }

    auto pipeline_config = cli::build_pipeline_config(
        "DFTracer Split", cli->pipeline, cli->watchdog);
    Pipeline pipeline(pipeline_config);

    auto start_time = std::chrono::high_resolution_clock::now();

    // Phase 1: Discover input files
    DFTRACER_UTILS_LOG_INFO("%s", "Discovering input files...");

    std::vector<std::string> input_files;
    for (const auto& entry : fs::directory_iterator(log_dir)) {
        if (entry.is_regular_file()) {
            std::string path = entry.path().string();
            if (path.ends_with(".pfw.gz") || path.ends_with(".pfw")) {
                input_files.push_back(path);
            }
        }
    }

    if (input_files.empty()) {
        DFTRACER_UTILS_LOG_ERROR("No .pfw or .pfw.gz files found in %s",
                                 log_dir.c_str());
        co_return 1;
    }

    DFTRACER_UTILS_LOG_INFO("Found %zu input files", input_files.size());

    if (force) {
        const std::string shared_index_path =
            utilities::composites::dft::internal::determine_index_path(
                input_files.front(), index_dir);
        if (fs::exists(shared_index_path)) {
            DFTRACER_UTILS_LOG_INFO("Clearing shared index store: %s",
                                    shared_index_path.c_str());
            fs::remove_all(shared_index_path);
        }
    }

    // Phase 2: Build TaskGraph for file processing
    auto graph = TaskGraph::builder(
        {.name = "DFTracerSplit", .max_concurrency = executor_threads});

    DFTRACER_UTILS_LOG_INFO("%s", "Creating batch index task...");

    auto* input_files_ptr = &input_files;
    auto batch_index_task = make_task(
        [input_files_ptr, checkpoint_size, index_dir,
         executor_threads](CoroScope& ctx) -> coro::CoroTask<void> {
            auto index_path =
                utilities::composites::dft::internal::determine_index_path(
                    input_files_ptr->front(), index_dir);
            dftracer::utils::rocksdb::RocksDBManager::instance().reset(
                index_path);

            auto batch_config =
                std::make_shared<utilities::indexer::IndexBuildBatchConfig>();
            batch_config->file_paths = *input_files_ptr;
            batch_config->index_dir = index_dir;
            batch_config->checkpoint_size = checkpoint_size;
            batch_config->parallelism = executor_threads;
            batch_config->use_batch_write = true;
            batch_config->rebuild_root_summaries = true;

            auto result =
                co_await utilities::indexer::IndexBatchBuilderUtility::process(
                    &ctx, std::move(batch_config));
            for (const auto& r : result.results) {
                if (!r.success && !r.error_message.empty()) {
                    DFTRACER_UTILS_LOG_ERROR("Auto-indexing failed for %s: %s",
                                             r.file_path.c_str(),
                                             r.error_message.c_str());
                }
            }
        },
        "BatchIndex");
    graph.add(batch_index_task);

    DFTRACER_UTILS_LOG_INFO("%s", "Creating file processing tasks...");

    auto file_metadata = graph.parallel<Metadata>(
        input_files.size(),
        [input_files_ptr, checkpoint_size, index_dir, verify](
            CoroScope&, std::size_t idx) -> coro::CoroTask<Metadata> {
            const auto& file_path = (*input_files_ptr)[idx];

            std::string index_path =
                utilities::composites::dft::internal::determine_index_path(
                    file_path, index_dir);

            auto meta_input =
                utilities::composites::dft::MetadataCollectorUtilityInput::
                    from_file(file_path)
                        .with_checkpoint_size(checkpoint_size)
                        .with_force_rebuild(false)
                        .with_index(index_path)
                        .with_compute_hash(verify);

            co_return co_await utilities::composites::dft::
                MetadataCollectorUtility{}
                    .process(meta_input);
        },
        {.name = "ProcessFile"});

    for (const auto& meta_task : file_metadata.tasks()) {
        meta_task->depends_on(batch_index_task);
    }

    DFTRACER_UTILS_LOG_INFO("%s", "Creating chunk mapping task...");

    auto manifests_group = graph.reduce<std::vector<ChunkManifest>>(
        file_metadata, split_every{input_files.size()},
        [chunk_size_mb](CoroScope&, std::vector<Metadata> all_metadata)
            -> coro::CoroTask<std::vector<ChunkManifest>> {
            DFTRACER_UTILS_LOG_INFO("Creating chunk mappings from %zu files...",
                                    all_metadata.size());

            utilities::composites::dft::ChunkManifestMapperUtility mapper;
            auto mapper_input =
                utilities::composites::dft::ChunkManifestMapperUtilityInput::
                    from_metadata(all_metadata)
                        .with_target_size(static_cast<double>(chunk_size_mb));

            auto manifests = co_await mapper.process(mapper_input);
            DFTRACER_UTILS_LOG_INFO("Created %zu chunks", manifests.size());
            co_return manifests;
        },
        {.name = "CreateManifests"});

    DFTRACER_UTILS_LOG_INFO("%s", "Creating extraction task...");

    using ExtractChunksOutput = std::vector<ExtractResult>;

    auto* app_name_ptr = &app_name;
    auto* output_dir_ptr = &output_dir;

    auto task_extract_chunks = make_task(
        [app_name_ptr, output_dir_ptr, compress, verify, executor_threads](
            CoroScope& scope, std::vector<ChunkManifest> manifests)
            -> coro::CoroTask<ExtractChunksOutput> {
            DFTRACER_UTILS_LOG_INFO("Extracting %zu chunks in parallel...",
                                    manifests.size());

            auto permits = coro::make_channel<bool>(executor_threads * 2);
            for (std::size_t i = 0; i < executor_threads * 2; ++i) {
                permits->try_send(true);
            }

            std::vector<coro::SpawnFuture<ExtractResult>> futures;
            futures.reserve(manifests.size());

            for (std::size_t i = 0; i < manifests.size(); ++i) {
                auto input = ExtractInput::from_manifest(
                                 static_cast<int>(i + 1), manifests[i])
                                 .with_output_dir(*output_dir_ptr)
                                 .with_app_name(*app_name_ptr)
                                 .with_compression(compress)
                                 .with_compute_hash(verify);

                futures.push_back(scope.spawn(
                    [input = std::move(input),
                     permits](CoroScope& s) -> coro::CoroTask<ExtractResult> {
                        co_await s.receive(permits);
                        try {
                            utilities::composites::dft::ChunkExtractorUtility
                                extractor;
                            auto result = co_await extractor.process(input);
                            permits->try_send(true);
                            co_return result;
                        } catch (...) {
                            permits->try_send(true);
                            throw;
                        }
                    }));
            }

            ExtractChunksOutput results;
            results.reserve(futures.size());
            for (auto& future : futures) {
                results.push_back(co_await future);
            }

            std::sort(results.begin(), results.end(),
                      [](const ExtractResult& a, const ExtractResult& b) {
                          return a.chunk_index < b.chunk_index;
                      });

            co_return results;
        },
        "ExtractChunks");

    task_extract_chunks->depends_on(manifests_group.task());
    graph.add(task_extract_chunks);

    // Phase 3: Optional verification
    std::shared_ptr<Task> final_task = task_extract_chunks;
    std::shared_ptr<Task> task_verify_chunks = nullptr;

    if (verify) {
        DFTRACER_UTILS_LOG_INFO("%s", "Configuring verification...");

        struct VerifyInput {
            ExtractChunksOutput chunks;
            std::vector<Metadata> all_metadata;
        };

        task_verify_chunks = make_task(
            [](CoroScope&, const VerifyInput& input)
                -> coro::CoroTask<
                    utilities::composites::ChunkVerificationUtilityOutput> {
                std::size_t output_hash = 0;
                for (const auto& chunk : input.chunks) {
                    output_hash += chunk.event_hash;
                }

                std::size_t input_hash = 0;
                for (const auto& meta : input.all_metadata) {
                    if (!meta.success) continue;
                    input_hash += meta.event_hash;
                }

                co_return utilities::composites::
                    ChunkVerificationUtilityOutput::success(
                        static_cast<std::uint64_t>(input_hash),
                        static_cast<std::uint64_t>(output_hash));
            },
            "VerifyChunks");

        task_verify_chunks->depends_on(task_extract_chunks);
        for (const auto& meta_task : file_metadata.tasks()) {
            task_verify_chunks->depends_on(meta_task);
        }

        task_verify_chunks->with_combiner(
            [](const std::vector<std::any>& inputs) -> std::any {
                auto chunks = std::any_cast<ExtractChunksOutput>(inputs[0]);

                std::vector<Metadata> all_metadata;
                all_metadata.reserve(inputs.size() - 1);
                for (std::size_t i = 1; i < inputs.size(); ++i) {
                    all_metadata.push_back(std::any_cast<Metadata>(inputs[i]));
                }

                VerifyInput vi{std::move(chunks), std::move(all_metadata)};
                return std::make_any<VerifyInput>(std::move(vi));
            });

        graph.add(task_verify_chunks);
        final_task = task_verify_chunks;
    }

    // Phase 4: Execute Pipeline
    DFTRACER_UTILS_LOG_INFO("%s", "Executing pipeline...");

    pipeline.set_source(batch_index_task);
    pipeline.set_destination(final_task);
    pipeline.execute();

    auto end_time = std::chrono::high_resolution_clock::now();
    std::chrono::duration<double, std::milli> duration = end_time - start_time;

    std::printf("\n");
    std::printf("==========================================\n");
    std::printf("Split Results\n");
    std::printf("==========================================\n");
    std::printf("  Execution time: %.2f seconds\n", duration.count() / 1000.0);
    std::printf("  Input: %zu files\n", input_files.size());

    int exit_code = 0;

    if (verify && task_verify_chunks) {
        auto verify_result =
            task_verify_chunks
                ->get<utilities::composites::ChunkVerificationUtilityOutput>();

        if (verify_result.input_hash == verify_result.output_hash) {
            std::printf(
                "  Verification: PASSED - all events present in output\n");
        } else {
            std::printf("  Verification: FAILED - event mismatch detected\n");
            exit_code = 1;
        }
        std::printf("    Input hash:  0x%016" PRIx64 "\n",
                    verify_result.input_hash);
        std::printf("    Output hash: 0x%016" PRIx64 "\n",
                    verify_result.output_hash);
    } else {
        auto extraction_results =
            task_extract_chunks->get<ExtractChunksOutput>();

        std::size_t successful_chunks = 0;
        std::size_t total_events = 0;

        for (const auto& result : extraction_results) {
            if (result.success) {
                successful_chunks++;
                total_events += result.events;
            } else {
                DFTRACER_UTILS_LOG_ERROR("Failed to create chunk %d",
                                         result.chunk_index);
            }
        }

        std::printf("  Output: %zu/%zu chunks, %zu events\n", successful_chunks,
                    extraction_results.size(), total_events);

        if (successful_chunks != extraction_results.size()) {
            exit_code = 1;
        }
    }

    std::printf("==========================================\n");

    if (!temp_index_dir.empty() && fs::exists(temp_index_dir)) {
        DFTRACER_UTILS_LOG_INFO("Cleaning up temporary index directory: %s",
                                temp_index_dir.c_str());
        fs::remove_all(temp_index_dir);
    }

    co_return exit_code;
}

int main(int argc, char** argv) {
    return cli::cli_main<SplitArgParse>(
        argc, argv, "dftracer_split",
        "Split DFTracer traces into equal-sized chunks using explicit pipeline "
        "with maximum parallelism",
        [](SplitArgParse& cli) { return run_split(&cli).get(); });
}

1	#include <dftracer/utils/core/common/config.h>
2	#include <dftracer/utils/core/pipeline/pipeline.h>
3	#include <dftracer/utils/core/rocksdb/db_manager.h>
4	#include <dftracer/utils/core/task_graph/task_graph.h>
5	#include <dftracer/utils/core/tasks/coro_scope.h>
6	#include <dftracer/utils/core/tasks/task.h>
7	#include <dftracer/utils/core/utilities/utility_adapter.h>
8	#include <dftracer/utils/utilities/composites/composites.h>
9	#include <dftracer/utils/utilities/composites/dft/chunk_extractor_utility.h>
10	#include <dftracer/utils/utilities/fileio/types/types.h>
11	#include <dftracer/utils/utilities/indexer/index_builder_utility.h>
12	#include <dftracer/utils/utilities/indexer/internal/indexer.h>
13	#include <unistd.h>
14
15	#include <chrono>
16	#include <cinttypes>
17
18	#include "common_cli.h"
19
20	using namespace dftracer::utils;
21	using namespace dftracer::utils::task_graph;
22	using Metadata = utilities::composites::dft::MetadataCollectorUtilityOutput;
23	using ChunkManifest =
24	utilities::composites::dft::internal::DFTracerChunkManifest;
25	using ExtractInput = utilities::composites::dft::ChunkExtractorUtilityInput;
26	using ExtractResult = utilities::composites::dft::ChunkExtractorUtilityOutput;
27
28	class SplitArgParse : public cli::ArgParse {
29	public:
30	cli::DirectoryArgs directory;
31	cli::PipelineArgs pipeline;
32	cli::IndexingArgs indexing;
33	cli::WatchdogArgs watchdog;
34
35	std::string app_name = "app";	5!
36	std::string output_dir = "./split";	5!
37	int chunk_size_mb = 4;	5✔
38	bool compress = true;	5✔
39	bool verify = false;	5✔
40
41	explicit SplitArgParse(argparse::ArgumentParser& p) : ArgParse(p) {	10!
42	indexing.force_help =	5!
43	"Override existing files and force index recreation";
44	schema(directory, pipeline, indexing, watchdog);	5!
45	}	10✔
46
47	protected:
48	void register_args() override {	5✔
49	parser()	10✔
50	.add_argument("-n", "--app-name")	5✔
51	.help("Application name for output files")	5✔
52	.default_value<std::string>("app");	5!
53
54	parser()	10✔
55	.add_argument("-o", "--output")	5✔
56	.help("Output directory for split files")	5✔
57	.default_value<std::string>("./split");	5!
58
59	parser()	10✔
60	.add_argument("-s", "--chunk-size")	5✔
61	.help("Chunk size in MB")	5✔
62	.scan<'d', int>()	5!
63	.default_value(4);	5!
64
65	parser()	10✔
66	.add_argument("-c", "--compress")	5✔
67	.help("Compress output files with gzip")	5✔
68	.flag()	5!
69	.default_value(true);	5!
70
71	parser()	5✔
72	.add_argument("--verify")	5✔
73	.help("Verify output chunks match input by comparing event IDs")	5✔
74	.flag();	5!
75	}	5✔
76
77	void post_parse() override {	5✔
78	app_name = parser().get<std::string>("--app-name");	5✔
79	output_dir = parser().get<std::string>("--output");	5✔
80	chunk_size_mb = parser().get<int>("--chunk-size");	5✔
81	compress = parser().get<bool>("--compress");	5✔
82	verify = parser().get<bool>("--verify");	5✔
83	}	5✔
84	};
85
86	static coro::CoroTask<int> run_split(const SplitArgParse* cli) {	35!
87	const auto log_dir = fs::absolute(cli->directory.value).string();	5!
88	const auto output_dir = fs::absolute(cli->output_dir).string();	5!
89	const auto& app_name = cli->app_name;	5✔
90	const auto chunk_size_mb = cli->chunk_size_mb;	5✔
91	const auto force = cli->indexing.force;	5✔
92	const auto compress = cli->compress;	5✔
93	const auto verify = cli->verify;	5✔
94	const auto checkpoint_size = cli->indexing.checkpoint_size;	5✔
95	const auto executor_threads = cli->pipeline.executor_threads;	5✔
96	auto index_dir = cli->indexing.index_dir;	5!
97
98	std::string temp_index_dir;	5✔
99	if (index_dir.empty()) {	5!
100	temp_index_dir = fs::temp_directory_path() /	10!
101	("dftracer_idx_" + std::to_string(std::time(nullptr)) +	10!
102	"_" + std::to_string(getpid()));	5!
103	fs::create_directories(temp_index_dir);	5!
104	index_dir = temp_index_dir;	5!
105	DFTRACER_UTILS_LOG_INFO("Created temporary index directory: %s",	5!
106	index_dir.c_str());
107	}	5✔
108
109	std::printf("==========================================\n");	5!
110	std::printf("DFTracer Split (Explicit Pipeline)\n");	5!
111	std::printf("==========================================\n");	5!
112	std::printf("Arguments:\n");	5!
113	std::printf(" App name: %s\n", app_name.c_str());	5!
114	std::printf(" Override: %s\n", force ? "true" : "false");	5!
115	std::printf(" Compress: %s\n", compress ? "true" : "false");	5!
116	std::printf(" Data dir: %s\n", log_dir.c_str());	5!
117	std::printf(" Output dir: %s\n", output_dir.c_str());	5!
118	std::printf(" Chunk size: %d MB\n", chunk_size_mb);	5!
119	std::printf(" Executor threads: %zu\n", executor_threads);	5!
120	std::printf("==========================================\n\n");	5!
121
122	if (!fs::exists(output_dir)) {	5!
123	fs::create_directories(output_dir);	5!
124	}	5✔
125
126	auto pipeline_config = cli::build_pipeline_config(	5!
127	"DFTracer Split", cli->pipeline, cli->watchdog);	5!
128	Pipeline pipeline(pipeline_config);	5!
129
130	auto start_time = std::chrono::high_resolution_clock::now();	5✔
131
132	// Phase 1: Discover input files
133	DFTRACER_UTILS_LOG_INFO("%s", "Discovering input files...");	5!
134
135	std::vector<std::string> input_files;	5✔
136	for (const auto& entry : fs::directory_iterator(log_dir)) {	15!
137	if (entry.is_regular_file()) {	10!
138	std::string path = entry.path().string();	5!
139	if (path.ends_with(".pfw.gz") \|\| path.ends_with(".pfw")) {	5!
140	input_files.push_back(path);	5!
141	}	5✔
142	}	5✔
143	}	10✔
144
145	if (input_files.empty()) {	5✔
146	DFTRACER_UTILS_LOG_ERROR("No .pfw or .pfw.gz files found in %s",	1!
147	log_dir.c_str());
148	co_return 1;	6!
149	}
150
151	DFTRACER_UTILS_LOG_INFO("Found %zu input files", input_files.size());	4!
152
153	if (force) {	4!
154	const std::string shared_index_path =	4✔
155	utilities::composites::dft::internal::determine_index_path(	4!
156	input_files.front(), index_dir);	4✔
157	if (fs::exists(shared_index_path)) {	4!
UNCOV 158	DFTRACER_UTILS_LOG_INFO("Clearing shared index store: %s",	×
159	shared_index_path.c_str());
UNCOV 160	fs::remove_all(shared_index_path);	×
UNCOV 161	}	×
162	}	4✔
163
164	// Phase 2: Build TaskGraph for file processing
165	auto graph = TaskGraph::builder(	4!
166	{.name = "DFTracerSplit", .max_concurrency = executor_threads});	4!
167
168	DFTRACER_UTILS_LOG_INFO("%s", "Creating batch index task...");	4!
169
170	auto* input_files_ptr = &input_files;	4✔
171	auto batch_index_task = make_task(	4!
172	[input_files_ptr, checkpoint_size, index_dir,	32!
173	executor_threads](CoroScope& ctx) -> coro::CoroTask<void> {	8!
174	auto index_path =	12✔
175	utilities::composites::dft::internal::determine_index_path(	12!
176	input_files_ptr->front(), index_dir);	12✔
177	dftracer::utils::rocksdb::RocksDBManager::instance().reset(	12✔
178	index_path);
179
180	auto batch_config =	12✔
181	std::make_shared<utilities::indexer::IndexBuildBatchConfig>();	12!
182	batch_config->file_paths = *input_files_ptr;	12✔
183	batch_config->index_dir = index_dir;	4✔
184	batch_config->checkpoint_size = checkpoint_size;	12✔
185	batch_config->parallelism = executor_threads;	12✔
186	batch_config->use_batch_write = true;	12✔
187	batch_config->rebuild_root_summaries = true;	12✔
188
189	auto result =	12✔
190	co_await utilities::indexer::IndexBatchBuilderUtility::process(	32!
191	&ctx, std::move(batch_config));	12✔
192	for (const auto& r : result.results) {	9✔
193	if (!r.success && !r.error_message.empty()) {	5!
UNCOV 194	DFTRACER_UTILS_LOG_ERROR("Auto-indexing failed for %s: %s",	×
195	r.file_path.c_str(),
196	r.error_message.c_str());
UNCOV 197	}	×
198	}	5✔
199	},	44!
200	"BatchIndex");	4!
201	graph.add(batch_index_task);	4!
202
203	DFTRACER_UTILS_LOG_INFO("%s", "Creating file processing tasks...");	4!
204
205	auto file_metadata = graph.parallel<Metadata>(	8!
206	input_files.size(),	4✔
207	[input_files_ptr, checkpoint_size, index_dir, verify](	40!
208	CoroScope&, std::size_t idx) -> coro::CoroTask<Metadata> {	6✔
209	const auto& file_path = (*input_files_ptr)[idx];	15✔
210
211	std::string index_path =	15✔
212	utilities::composites::dft::internal::determine_index_path(	15!
213	file_path, index_dir);	15✔
214
215	auto meta_input =	15✔
216	utilities::composites::dft::MetadataCollectorUtilityInput::	30!
217	from_file(file_path)	15!
218	.with_checkpoint_size(checkpoint_size)	15✔
219	.with_force_rebuild(false)	5!
220	.with_index(index_path)	5✔
221	.with_compute_hash(verify);	5!
222
223	co_return co_await utilities::composites::dft::	41!
224	MetadataCollectorUtility{}
225	.process(meta_input);	15!
226	},	55!
227	{.name = "ProcessFile"});	4!
228
229	for (const auto& meta_task : file_metadata.tasks()) {	9!
230	meta_task->depends_on(batch_index_task);	5!
231	}	5✔
232
233	DFTRACER_UTILS_LOG_INFO("%s", "Creating chunk mapping task...");	4!
234
235	auto manifests_group = graph.reduce<std::vector<ChunkManifest>>(	8!
236	file_metadata, split_every{input_files.size()},	4!
237	[chunk_size_mb](CoroScope&, std::vector<Metadata> all_metadata)	32!
238	-> coro::CoroTask<std::vector<ChunkManifest>> {	4!
239	DFTRACER_UTILS_LOG_INFO("Creating chunk mappings from %zu files...",	12!
240	all_metadata.size());
241
242	utilities::composites::dft::ChunkManifestMapperUtility mapper;	12!
243	auto mapper_input =	12✔
244	utilities::composites::dft::ChunkManifestMapperUtilityInput::	24!
245	from_metadata(all_metadata)	12!
246	.with_target_size(static_cast<double>(chunk_size_mb));	12✔
247
248	auto manifests = co_await mapper.process(mapper_input);	20!
249	DFTRACER_UTILS_LOG_INFO("Created %zu chunks", manifests.size());	4!
250	co_return manifests;	4!
251	},	28!
252	{.name = "CreateManifests"});	4!
253
254	DFTRACER_UTILS_LOG_INFO("%s", "Creating extraction task...");	4!
255
256	using ExtractChunksOutput = std::vector<ExtractResult>;
257
258	auto* app_name_ptr = &app_name;	4✔
259	auto* output_dir_ptr = &output_dir;	4✔
260
261	auto task_extract_chunks = make_task(	8!
262	[app_name_ptr, output_dir_ptr, compress, verify, executor_threads](	40!
263	CoroScope& scope, std::vector<ChunkManifest> manifests)
264	-> coro::CoroTask<ExtractChunksOutput> {	4!
265	DFTRACER_UTILS_LOG_INFO("Extracting %zu chunks in parallel...",	4!
266	manifests.size());
267
268	auto permits = coro::make_channel<bool>(executor_threads * 2);	4!
269	for (std::size_t i = 0; i < executor_threads * 2; ++i) {	28✔
270	permits->try_send(true);	24!
271	}	24✔
272
273	std::vector<coro::SpawnFuture<ExtractResult>> futures;	4✔
274	futures.reserve(manifests.size());	4!
275
276	for (std::size_t i = 0; i < manifests.size(); ++i) {	8✔
277	auto input = ExtractInput::from_manifest(	8!
278	static_cast<int>(i + 1), manifests[i])	4!
279	.with_output_dir(*output_dir_ptr)	4!
280	.with_app_name(*app_name_ptr)	4!
281	.with_compression(compress)	4!
282	.with_compute_hash(verify);	4!
283
284	futures.push_back(scope.spawn(	8!
285	[input = std::move(input),	40!
286	permits](CoroScope& s) -> coro::CoroTask<ExtractResult> {	8!
287	co_await s.receive(permits);	8!
288	try {
289	utilities::composites::dft::ChunkExtractorUtility	12✔
290	extractor;	12!
291	auto result = co_await extractor.process(input);	16!
292	permits->try_send(true);	4!
293	co_return result;	4!
294	} catch (...) {	4!
UNCOV 295	permits->try_send(true);	×
UNCOV 296	throw;	×
UNCOV 297	}	×
298	}));	8✔
299	}	4✔
300
301	ExtractChunksOutput results;	4✔
302	results.reserve(futures.size());	4!
303	for (auto& future : futures) {	24✔
304	results.push_back(co_await future);	20!
305	}	4✔
306
307	std::sort(results.begin(), results.end(),	4!
UNCOV 308	[](const ExtractResult& a, const ExtractResult& b) {	×
UNCOV 309	return a.chunk_index < b.chunk_index;	×
310	});
311
312	co_return results;	4!
313	},	20✔
314	"ExtractChunks");	4!
315
316	task_extract_chunks->depends_on(manifests_group.task());	4!
317	graph.add(task_extract_chunks);	4!
318
319	// Phase 3: Optional verification
320	std::shared_ptr<Task> final_task = task_extract_chunks;	4✔
321	std::shared_ptr<Task> task_verify_chunks = nullptr;	4✔
322
323	if (verify) {	4✔
324	DFTRACER_UTILS_LOG_INFO("%s", "Configuring verification...");	1!
325
326	struct VerifyInput {	2✔
327	ExtractChunksOutput chunks;
328	std::vector<Metadata> all_metadata;
329	};
330
331	task_verify_chunks = make_task(	2!
332	[](CoroScope&, const VerifyInput& input)	6!
333	-> coro::CoroTask<
334	utilities::composites::ChunkVerificationUtilityOutput> {	1!
335	std::size_t output_hash = 0;	1✔
336	for (const auto& chunk : input.chunks) {	2✔
337	output_hash += chunk.event_hash;	1✔
338	}	1✔
339
340	std::size_t input_hash = 0;	1✔
341	for (const auto& meta : input.all_metadata) {	2✔
342	if (!meta.success) continue;	1!
343	input_hash += meta.event_hash;	1✔
344	}	1!
345
346	co_return utilities::composites::	3!
347	ChunkVerificationUtilityOutput::success(
348	static_cast<std::uint64_t>(input_hash),	1✔
349	static_cast<std::uint64_t>(output_hash));	1✔
350	},	1✔
351	"VerifyChunks");	1!
352
353	task_verify_chunks->depends_on(task_extract_chunks);	1!
354	for (const auto& meta_task : file_metadata.tasks()) {	2!
355	task_verify_chunks->depends_on(meta_task);	1!
356	}	1✔
357
358	task_verify_chunks->with_combiner(	2!
359	[](const std::vector<std::any>& inputs) -> std::any {	2✔
360	auto chunks = std::any_cast<ExtractChunksOutput>(inputs[0]);	1✔
361
362	std::vector<Metadata> all_metadata;	1✔
363	all_metadata.reserve(inputs.size() - 1);	1!
364	for (std::size_t i = 1; i < inputs.size(); ++i) {	2✔
365	all_metadata.push_back(std::any_cast<Metadata>(inputs[i]));	1!
366	}	1✔
367
368	VerifyInput vi{std::move(chunks), std::move(all_metadata)};	1✔
369	return std::make_any<VerifyInput>(std::move(vi));	1!
370	});	1✔
371
372	graph.add(task_verify_chunks);	1!
373	final_task = task_verify_chunks;	1✔
374	}	1✔
375
376	// Phase 4: Execute Pipeline
377	DFTRACER_UTILS_LOG_INFO("%s", "Executing pipeline...");	4!
378
379	pipeline.set_source(batch_index_task);	4!
380	pipeline.set_destination(final_task);	4!
381	pipeline.execute();	4!
382
383	auto end_time = std::chrono::high_resolution_clock::now();	4✔
384	std::chrono::duration<double, std::milli> duration = end_time - start_time;	4!
385
386	std::printf("\n");	4!
387	std::printf("==========================================\n");	4!
388	std::printf("Split Results\n");	4!
389	std::printf("==========================================\n");	4!
390	std::printf(" Execution time: %.2f seconds\n", duration.count() / 1000.0);	4!
391	std::printf(" Input: %zu files\n", input_files.size());	4!
392
393	int exit_code = 0;	4✔
394
395	if (verify && task_verify_chunks) {	4!
396	auto verify_result =	1✔
397	task_verify_chunks	1✔
398	->get<utilities::composites::ChunkVerificationUtilityOutput>();	1!
399
400	if (verify_result.input_hash == verify_result.output_hash) {	1!
401	std::printf(	1!
402	" Verification: PASSED - all events present in output\n");
403	} else {	1✔
UNCOV 404	std::printf(" Verification: FAILED - event mismatch detected\n");	×
UNCOV 405	exit_code = 1;	×
406	}
407	std::printf(" Input hash: 0x%016" PRIx64 "\n",	1!
408	verify_result.input_hash);	1✔
409	std::printf(" Output hash: 0x%016" PRIx64 "\n",	1!
410	verify_result.output_hash);	1✔
411	} else {	1✔
412	auto extraction_results =	3✔
413	task_extract_chunks->get<ExtractChunksOutput>();	3!
414
415	std::size_t successful_chunks = 0;	3✔
416	std::size_t total_events = 0;	3✔
417
418	for (const auto& result : extraction_results) {	6✔
419	if (result.success) {	3!
420	successful_chunks++;	3✔
421	total_events += result.events;	3✔
422	} else {	3✔
UNCOV 423	DFTRACER_UTILS_LOG_ERROR("Failed to create chunk %d",	×
424	result.chunk_index);
425	}
426	}	3✔
427
428	std::printf(" Output: %zu/%zu chunks, %zu events\n", successful_chunks,	6!
429	extraction_results.size(), total_events);	3✔
430
431	if (successful_chunks != extraction_results.size()) {	3!
UNCOV 432	exit_code = 1;	×
UNCOV 433	}	×
434	}	3✔
435
436	std::printf("==========================================\n");	4!
437
438	if (!temp_index_dir.empty() && fs::exists(temp_index_dir)) {	8!
439	DFTRACER_UTILS_LOG_INFO("Cleaning up temporary index directory: %s",	4!
440	temp_index_dir.c_str());
441	fs::remove_all(temp_index_dir);	4!
442	}	4✔
443
444	co_return exit_code;	4!
445	}	5✔
446
447	int main(int argc, char** argv) {	5✔
448	return cli::cli_main<SplitArgParse>(	5✔
449	argc, argv, "dftracer_split",	5✔
450	"Split DFTracer traces into equal-sized chunks using explicit pipeline "
451	"with maximum parallelism",
452	[](SplitArgParse& cli) { return run_split(&cli).get(); });	5!
453	}

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