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

44.64

/src/dftracer/utils/binaries/dftracer_reader.cpp

#include <dftracer/utils/core/common/config.h>
#include <dftracer/utils/core/common/constants.h>
#include <dftracer/utils/core/common/filesystem.h>
#include <dftracer/utils/core/common/logging.h>
#include <dftracer/utils/core/coro/task.h>
#include <dftracer/utils/core/io/io.h>
#include <dftracer/utils/utilities/composites/composites.h>
#include <dftracer/utils/utilities/composites/dft/indexing/resolve_and_build.h>
#include <dftracer/utils/utilities/indexer/internal/indexer.h>
#include <dftracer/utils/utilities/indexer/internal/indexer_factory.h>
#include <dftracer/utils/utilities/reader/internal/reader.h>
#include <dftracer/utils/utilities/reader/internal/reader_factory.h>
#include <dftracer/utils/utilities/reader/internal/stream.h>
#include <dftracer/utils/utilities/reader/internal/stream_config.h>
#include <dftracer/utils/utilities/reader/internal/stream_type.h>
#include <fcntl.h>
#include <unistd.h>

#include <argparse/argparse.hpp>
#include <cstdio>
#include <cstdlib>
#include <cstring>

#include "common_cli.h"
using namespace dftracer::utils;
using namespace dftracer::utils::utilities::indexer::internal;
using namespace dftracer::utils::utilities::reader::internal;

static coro::CoroTask<int> run_reader(const std::string &gz_path,
                                      const std::string &index_path,
                                      std::size_t checkpoint_size,
                                      bool force_rebuild, bool check_rebuild,
                                      const std::string &read_mode,
                                      std::size_t read_buffer_size,
                                      int64_t start, int64_t end) {
    const std::string index_root = normalize_index_root(index_path);

    // Create indexer first
    std::shared_ptr<Indexer> indexer;
    try {
        // Check whether the root-local .dftindex store already exists.
        if (!fs::exists(index_root)) {
            if (check_rebuild) {
                DFTRACER_UTILS_LOG_ERROR(
                    "Index store '%s' does not exist, cannot check",
                    index_root.c_str());
                co_return 1;
            }
            DFTRACER_UTILS_LOG_DEBUG("Index store '%s' does not exist",
                                     index_root.c_str());
            DFTRACER_UTILS_LOG_DEBUG("%s", "Will create new index store");
            force_rebuild = true;
        }

        // Use IndexerFactory to create appropriate indexer
        indexer = IndexerFactory::create(gz_path, index_path, checkpoint_size,
                                         force_rebuild);

        if (check_rebuild) {
            if (!indexer->need_rebuild()) {
                DFTRACER_UTILS_LOG_DEBUG(
                    "%s", "Index is up to date, no rebuild needed");
                co_return 0;
            }
        }

        if (force_rebuild) {
            if (fs::exists(index_root)) {
                DFTRACER_UTILS_LOG_DEBUG("Removing existing index store: %s",
                                         index_root.c_str());
                fs::remove_all(index_root);
            }
            // Recreate the store after removing the old .dftindex root.
            indexer = IndexerFactory::create(gz_path, index_path,
                                             checkpoint_size, true);
            DFTRACER_UTILS_LOG_INFO("Building index store for file: %s",
                                    gz_path.c_str());
            co_await indexer->build_async();
        }
    } catch (const std::runtime_error &e) {
        DFTRACER_UTILS_LOG_ERROR("Indexer error: %s", e.what());
        co_return 1;
    }

    // read operations
    try {
        // Use ReaderFactory to create appropriate reader, sharing
        // ownership of indexer
        auto reader = ReaderFactory::create(indexer);

        if (read_mode.find("bytes") == std::string::npos) {
            std::size_t start_line =
                (start == -1) ? 1 : static_cast<std::size_t>(start);
            std::size_t end_line = static_cast<std::size_t>(end);
            if (end == -1) {
                end_line = reader->get_num_lines();
            }

            DFTRACER_UTILS_LOG_DEBUG("Reading lines from %zu to %zu",
                                     start_line, end_line);

            auto stream =
                reader->stream(StreamConfig()
                                   .stream_type(StreamType::MULTI_LINES)
                                   .range_type(RangeType::LINE_RANGE)
                                   .from(start_line)
                                   .to(end_line)
                                   .buffer_size(read_buffer_size));

#if DFTRACER_UTILS_LOGGER_DEBUG_ENABLED
            std::size_t line_count = 0;
#endif

            while (!stream->done()) {
                auto chunk = co_await stream->read_async();
                if (chunk.empty()) break;
                co_await io::write(STDOUT_FILENO, chunk.data(), chunk.size());
#if DFTRACER_UTILS_LOGGER_DEBUG_ENABLED
                line_count += std::count(chunk.begin(), chunk.end(), '\n');
#endif
            }

            DFTRACER_UTILS_LOG_DEBUG("Successfully read %zu lines from range",
                                     line_count);
        } else {
            std::size_t start_bytes_ =
                (start == -1) ? 0 : static_cast<std::size_t>(start);
            std::size_t end_bytes_ =
                end == -1 ? std::numeric_limits<std::size_t>::max()
                          : static_cast<size_t>(end);

            auto max_bytes = reader->get_max_bytes();
            if (end_bytes_ > max_bytes) {
                end_bytes_ = max_bytes;
            }
            DFTRACER_UTILS_LOG_DEBUG("%s",
                                     "Performing byte range read operation");
            DFTRACER_UTILS_LOG_DEBUG("Using read buffer size: %zu bytes",
                                     read_buffer_size);

            StreamType stream_type = (read_mode == "bytes")
                                         ? StreamType::BYTES
                                         : StreamType::MULTI_LINES_BYTES;

            auto stream = reader->stream(StreamConfig()
                                             .stream_type(stream_type)
                                             .range_type(RangeType::BYTE_RANGE)
                                             .from(start_bytes_)
                                             .to(end_bytes_)
                                             .buffer_size(read_buffer_size));

#if DFTRACER_UTILS_LOGGER_DEBUG_ENABLED == 1
            std::size_t total_bytes = 0;
#endif

            while (!stream->done()) {
                auto chunk = co_await stream->read_async();
                if (chunk.empty()) break;
                co_await io::write(STDOUT_FILENO, chunk.data(), chunk.size());
#if DFTRACER_UTILS_LOGGER_DEBUG_ENABLED == 1
                total_bytes += chunk.size();
#endif
            }

            DFTRACER_UTILS_LOG_DEBUG("Successfully read %zu bytes from range",
                                     total_bytes);
        }
        fsync(STDOUT_FILENO);
    } catch (const std::runtime_error &e) {
        DFTRACER_UTILS_LOG_ERROR("Reader error: %s", e.what());
        co_return 1;
    }

    co_return 0;
}

int main(int argc, char **argv) {
    dftracer::utils::logger::init();
    auto default_checkpoint_size_str =
        std::to_string(Indexer::DEFAULT_CHECKPOINT_SIZE) + " B (" +
        std::to_string(Indexer::DEFAULT_CHECKPOINT_SIZE / (1024 * 1024)) +
        " MB)";
    argparse::ArgumentParser program("dft_reader",
                                     DFTRACER_UTILS_PACKAGE_VERSION);
    program.add_description(
        "DFTracer utility for reading and indexing compressed files (GZIP, "
        "TAR.GZ)");
    program.add_argument("file")
        .help("Compressed file to process (GZIP, TAR.GZ)")
        .required();
    program.add_argument("-i", "--index")
        .help("Path to the .dftindex store to use")
        .default_value<std::string>("");
    program.add_argument("-s", "--start")
        .help("Start position in bytes")
        .default_value<std::int64_t>(-1)
        .scan<'d', std::int64_t>();
    program.add_argument("-e", "--end")
        .help("End position in bytes")
        .default_value<std::int64_t>(-1)
        .scan<'d', std::int64_t>();
    program.add_argument("-c", "--checkpoint-size")
        .help("Checkpoint size for indexing in bytes (default: " +
              default_checkpoint_size_str + ")")
        .scan<'d', std::size_t>()
        .default_value(
            static_cast<std::size_t>(Indexer::DEFAULT_CHECKPOINT_SIZE));
    program.add_argument("-f", "--force-rebuild")
        .help("Force rebuild the .dftindex store")
        .flag();
    program.add_argument("--check")
        .help("Check if the .dftindex store is valid")
        .flag();
    program.add_argument("--read-buffer-size")
        .help("Size of the read buffer in bytes (default: 1MB)")
        .default_value<std::size_t>(1 * 1024 * 1024)
        .scan<'d', std::size_t>();
    program.add_argument("--mode")
        .help("Set the reading mode (bytes, line_bytes, lines)")
        .default_value<std::string>("bytes")
        .choices("bytes", "line_bytes", "lines");
    program.add_argument("--index-dir")
        .help("Directory to store root-local .dftindex directories")
        .default_value<std::string>("");
    cli::add_log_level_arg(program);

    try {
        program.parse_args(argc, argv);
    } catch (const std::exception &err) {
        DFTRACER_UTILS_LOG_ERROR("Error occurred: %s", err.what());
        std::cerr << program;
        return 1;
    }
    cli::apply_log_level_arg(program);

    std::string gz_path = program.get<std::string>("file");
    std::string index_path = program.get<std::string>("--index");
    int64_t start = program.get<int64_t>("--start");
    int64_t end = program.get<int64_t>("--end");
    std::size_t checkpoint_size = program.get<std::size_t>("--checkpoint-size");
    bool force_rebuild = program.get<bool>("--force-rebuild");
    bool check_rebuild = program.get<bool>("--check");
    std::string read_mode = program.get<std::string>("--mode");
    std::size_t read_buffer_size =
        program.get<std::size_t>("--read-buffer-size");
    std::string index_dir = program.get<std::string>("--index-dir");

    DFTRACER_UTILS_LOG_DEBUG("Processing file: %s", gz_path.c_str());
    DFTRACER_UTILS_LOG_DEBUG("Start position: %lld", (long long)start);
    DFTRACER_UTILS_LOG_DEBUG("End position: %lld", (long long)end);
    DFTRACER_UTILS_LOG_DEBUG("Mode: %s", read_mode.c_str());
    DFTRACER_UTILS_LOG_DEBUG("Checkpoint size: %zu B (%zu MB)", checkpoint_size,
                             checkpoint_size / (1024 * 1024));
    DFTRACER_UTILS_LOG_DEBUG("Force rebuild: %s",
                             force_rebuild ? "true" : "false");

    if (checkpoint_size <= 0) {
        DFTRACER_UTILS_LOG_ERROR(
            "%s",
            "Checkpoint size must be positive (greater than 0 and in MB)");
        return 1;
    }

    int test_fd = ::open(gz_path.c_str(), O_RDONLY);
    if (test_fd < 0) {
        DFTRACER_UTILS_LOG_ERROR("File '%s' does not exist or cannot be opened",
                                 gz_path.c_str());
        return 1;
    }
    ::close(test_fd);

    const bool index_explicit = !index_path.empty();
    if (index_path.empty()) {
        index_path = utilities::composites::dft::internal::determine_index_path(
            gz_path, index_dir);
    }

#if DFTRACER_UTILS_LOGGER_DEBUG_ENABLED
    ArchiveFormat format = IndexerFactory::detect_format(gz_path);

    DFTRACER_UTILS_LOG_DEBUG("Detected format: %s",
                             format == ArchiveFormat::TAR_GZ ? "TAR.GZ"
                             : format == ArchiveFormat::GZIP ? "GZIP"
                                                             : "UNKNOWN");
#endif

    // Skip for an explicit --index path (helper derives its own) or when
    // --check/--force-rebuild already handle rebuilds in run_reader below.
    if (!index_explicit && !check_rebuild && !force_rebuild) {
        cli::PipelineArgs refresh_pipeline;
        refresh_pipeline.executor_threads =
            dftracer_utils_hardware_concurrency();
        refresh_pipeline.io_threads = dftracer_utils_hardware_concurrency();
        cli::ensure_indexes_fresh_blocking("DFTracer Reader Index Refresh",
                                           refresh_pipeline, "", {gz_path},
                                           index_dir);
    }

    return run_reader(gz_path, index_path, checkpoint_size, force_rebuild,
                      check_rebuild, read_mode, read_buffer_size, start, end)
        .get();
}

1	#include <dftracer/utils/core/common/config.h>
2	#include <dftracer/utils/core/common/constants.h>
3	#include <dftracer/utils/core/common/filesystem.h>
4	#include <dftracer/utils/core/common/logging.h>
5	#include <dftracer/utils/core/coro/task.h>
6	#include <dftracer/utils/core/io/io.h>
7	#include <dftracer/utils/utilities/composites/composites.h>
8	#include <dftracer/utils/utilities/composites/dft/indexing/resolve_and_build.h>
9	#include <dftracer/utils/utilities/indexer/internal/indexer.h>
10	#include <dftracer/utils/utilities/indexer/internal/indexer_factory.h>
11	#include <dftracer/utils/utilities/reader/internal/reader.h>
12	#include <dftracer/utils/utilities/reader/internal/reader_factory.h>
13	#include <dftracer/utils/utilities/reader/internal/stream.h>
14	#include <dftracer/utils/utilities/reader/internal/stream_config.h>
15	#include <dftracer/utils/utilities/reader/internal/stream_type.h>
16	#include <fcntl.h>
17	#include <unistd.h>
18
19	#include <argparse/argparse.hpp>
20	#include <cstdio>
21	#include <cstdlib>
22	#include <cstring>
23
24	#include "common_cli.h"
25	using namespace dftracer::utils;
26	using namespace dftracer::utils::utilities::indexer::internal;
27	using namespace dftracer::utils::utilities::reader::internal;
28
29	static coro::CoroTask<int> run_reader(const std::string &gz_path,	32!
30	const std::string &index_path,
31	std::size_t checkpoint_size,
32	bool force_rebuild, bool check_rebuild,
33	const std::string &read_mode,
34	std::size_t read_buffer_size,
35	int64_t start, int64_t end) {	4!
36	const std::string index_root = normalize_index_root(index_path);	4!
37
38	// Create indexer first
39	std::shared_ptr<Indexer> indexer;	4✔
40	try {
41	// Check whether the root-local .dftindex store already exists.
42	if (!fs::exists(index_root)) {	4!
UNCOV 43	if (check_rebuild) {	×
UNCOV 44	DFTRACER_UTILS_LOG_ERROR(	×
45	"Index store '%s' does not exist, cannot check",
46	index_root.c_str());
47	co_return 1;	4✔
48	}
UNCOV 49	DFTRACER_UTILS_LOG_DEBUG("Index store '%s' does not exist",	×
50	index_root.c_str());
UNCOV 51	DFTRACER_UTILS_LOG_DEBUG("%s", "Will create new index store");	×
UNCOV 52	force_rebuild = true;	×
UNCOV 53	}	×
54
55	// Use IndexerFactory to create appropriate indexer
56	indexer = IndexerFactory::create(gz_path, index_path, checkpoint_size,	8!
57	force_rebuild);	4✔
58
59	if (check_rebuild) {	4✔
60	if (!indexer->need_rebuild()) {	1!
61	DFTRACER_UTILS_LOG_DEBUG(	1!
62	"%s", "Index is up to date, no rebuild needed");
63	co_return 0;	1✔
64	}
UNCOV 65	}	×
66
67	if (force_rebuild) {	3!
UNCOV 68	if (fs::exists(index_root)) {	×
UNCOV 69	DFTRACER_UTILS_LOG_DEBUG("Removing existing index store: %s",	×
70	index_root.c_str());
UNCOV 71	fs::remove_all(index_root);	×
UNCOV 72	}	×
73	// Recreate the store after removing the old .dftindex root.
UNCOV 74	indexer = IndexerFactory::create(gz_path, index_path,	×
UNCOV 75	checkpoint_size, true);	×
UNCOV 76	DFTRACER_UTILS_LOG_INFO("Building index store for file: %s",	×
77	gz_path.c_str());
UNCOV 78	co_await indexer->build_async();	×
UNCOV 79	}	×
80	} catch (const std::runtime_error &e) {	3!
UNCOV 81	DFTRACER_UTILS_LOG_ERROR("Indexer error: %s", e.what());	×
UNCOV 82	co_return 1;	×
UNCOV 83	}	×
84
85	// read operations
86	try {
87	// Use ReaderFactory to create appropriate reader, sharing
88	// ownership of indexer
89	auto reader = ReaderFactory::create(indexer);	3!
90
91	if (read_mode.find("bytes") == std::string::npos) {	3!
92	std::size_t start_line =	6✔
93	(start == -1) ? 1 : static_cast<std::size_t>(start);	3✔
94	std::size_t end_line = static_cast<std::size_t>(end);	3✔
95	if (end == -1) {	3✔
96	end_line = reader->get_num_lines();	2!
97	}	2✔
98
99	DFTRACER_UTILS_LOG_DEBUG("Reading lines from %zu to %zu",	3!
100	start_line, end_line);
101
102	auto stream =	3✔
103	reader->stream(StreamConfig()	6!
104	.stream_type(StreamType::MULTI_LINES)	3!
105	.range_type(RangeType::LINE_RANGE)	3!
106	.from(start_line)	3!
107	.to(end_line)	3!
108	.buffer_size(read_buffer_size));	3!
109
110	#if DFTRACER_UTILS_LOGGER_DEBUG_ENABLED
111	std::size_t line_count = 0;	3✔
112	#endif
113
114	while (!stream->done()) {	6!
115	auto chunk = co_await stream->read_async();	24!
116	if (chunk.empty()) break;	6✔
117	co_await io::write(STDOUT_FILENO, chunk.data(), chunk.size());	3!
118	#if DFTRACER_UTILS_LOGGER_DEBUG_ENABLED
119	line_count += std::count(chunk.begin(), chunk.end(), '\n');	3!
120	#endif
121	}	6✔
122
123	DFTRACER_UTILS_LOG_DEBUG("Successfully read %zu lines from range",	21!
124	line_count);
125	} else {	9✔
UNCOV 126	std::size_t start_bytes_ =	×
UNCOV 127	(start == -1) ? 0 : static_cast<std::size_t>(start);	×
UNCOV 128	std::size_t end_bytes_ =	×
UNCOV 129	end == -1 ? std::numeric_limits<std::size_t>::max()	×
UNCOV 130	: static_cast<size_t>(end);	×
131
UNCOV 132	auto max_bytes = reader->get_max_bytes();	×
UNCOV 133	if (end_bytes_ > max_bytes) {	×
UNCOV 134	end_bytes_ = max_bytes;	×
UNCOV 135	}	×
UNCOV 136	DFTRACER_UTILS_LOG_DEBUG("%s",	×
137	"Performing byte range read operation");
UNCOV 138	DFTRACER_UTILS_LOG_DEBUG("Using read buffer size: %zu bytes",	×
139	read_buffer_size);
140
UNCOV 141	StreamType stream_type = (read_mode == "bytes")	×
142	? StreamType::BYTES
143	: StreamType::MULTI_LINES_BYTES;
144
UNCOV 145	auto stream = reader->stream(StreamConfig()	×
UNCOV 146	.stream_type(stream_type)	×
UNCOV 147	.range_type(RangeType::BYTE_RANGE)	×
UNCOV 148	.from(start_bytes_)	×
UNCOV 149	.to(end_bytes_)	×
UNCOV 150	.buffer_size(read_buffer_size));	×
151
152	#if DFTRACER_UTILS_LOGGER_DEBUG_ENABLED == 1
UNCOV 153	std::size_t total_bytes = 0;	×
154	#endif
155
UNCOV 156	while (!stream->done()) {	×
UNCOV 157	auto chunk = co_await stream->read_async();	×
UNCOV 158	if (chunk.empty()) break;	×
UNCOV 159	co_await io::write(STDOUT_FILENO, chunk.data(), chunk.size());	×
160	#if DFTRACER_UTILS_LOGGER_DEBUG_ENABLED == 1
UNCOV 161	total_bytes += chunk.size();	×
162	#endif
UNCOV 163	}	×
164
UNCOV 165	DFTRACER_UTILS_LOG_DEBUG("Successfully read %zu bytes from range",	×
166	total_bytes);
UNCOV 167	}	×
168	fsync(STDOUT_FILENO);	3!
169	} catch (const std::runtime_error &e) {	9!
UNCOV 170	DFTRACER_UTILS_LOG_ERROR("Reader error: %s", e.what());	×
UNCOV 171	co_return 1;	×
UNCOV 172	}	×
173
174	co_return 0;	3✔
175	}	22✔
176
177	int main(int argc, char **argv) {	5✔
178	dftracer::utils::logger::init();	5✔
179	auto default_checkpoint_size_str =
180	std::to_string(Indexer::DEFAULT_CHECKPOINT_SIZE) + " B (" +	15!
181	std::to_string(Indexer::DEFAULT_CHECKPOINT_SIZE / (1024 * 1024)) +	10!
182	" MB)";
183	argparse::ArgumentParser program("dft_reader",	10!
184	DFTRACER_UTILS_PACKAGE_VERSION);	5!
185	program.add_description(	5!
186	"DFTracer utility for reading and indexing compressed files (GZIP, "	5!
187	"TAR.GZ)");
188	program.add_argument("file")	10!
189	.help("Compressed file to process (GZIP, TAR.GZ)")	5!
190	.required();	5!
191	program.add_argument("-i", "--index")	10!
192	.help("Path to the .dftindex store to use")	5!
193	.default_value<std::string>("");	5!
194	program.add_argument("-s", "--start")	10!
195	.help("Start position in bytes")	5!
196	.default_value<std::int64_t>(-1)	5!
197	.scan<'d', std::int64_t>();	5!
198	program.add_argument("-e", "--end")	10!
199	.help("End position in bytes")	5!
200	.default_value<std::int64_t>(-1)	5!
201	.scan<'d', std::int64_t>();	5!
202	program.add_argument("-c", "--checkpoint-size")	10!
203	.help("Checkpoint size for indexing in bytes (default: " +	10!
204	default_checkpoint_size_str + ")")	5!
205	.scan<'d', std::size_t>()	5!
206	.default_value(	5!
207	static_cast<std::size_t>(Indexer::DEFAULT_CHECKPOINT_SIZE));	5✔
208	program.add_argument("-f", "--force-rebuild")	10!
209	.help("Force rebuild the .dftindex store")	5!
210	.flag();	5!
211	program.add_argument("--check")	10!
212	.help("Check if the .dftindex store is valid")	5!
213	.flag();	5!
214	program.add_argument("--read-buffer-size")	10!
215	.help("Size of the read buffer in bytes (default: 1MB)")	5!
216	.default_value<std::size_t>(1 * 1024 * 1024)	5!
217	.scan<'d', std::size_t>();	5!
218	program.add_argument("--mode")	10!
219	.help("Set the reading mode (bytes, line_bytes, lines)")	5!
220	.default_value<std::string>("bytes")	5!
221	.choices("bytes", "line_bytes", "lines");	5!
222	program.add_argument("--index-dir")	10!
223	.help("Directory to store root-local .dftindex directories")	5!
224	.default_value<std::string>("");	5!
225	cli::add_log_level_arg(program);	5!
226
227	try {
228	program.parse_args(argc, argv);	5!
229	} catch (const std::exception &err) {	5!
230	DFTRACER_UTILS_LOG_ERROR("Error occurred: %s", err.what());	×
231	std::cerr << program;	×
232	return 1;	×
233	}	×
234	cli::apply_log_level_arg(program);	5!
235
236	std::string gz_path = program.get<std::string>("file");	5!
237	std::string index_path = program.get<std::string>("--index");	5!
238	int64_t start = program.get<int64_t>("--start");	5!
239	int64_t end = program.get<int64_t>("--end");	5!
240	std::size_t checkpoint_size = program.get<std::size_t>("--checkpoint-size");	5!
241	bool force_rebuild = program.get<bool>("--force-rebuild");	5!
242	bool check_rebuild = program.get<bool>("--check");	5!
243	std::string read_mode = program.get<std::string>("--mode");	5!
244	std::size_t read_buffer_size =	5✔
245	program.get<std::size_t>("--read-buffer-size");	5!
246	std::string index_dir = program.get<std::string>("--index-dir");	5!
247
248	DFTRACER_UTILS_LOG_DEBUG("Processing file: %s", gz_path.c_str());	5!
249	DFTRACER_UTILS_LOG_DEBUG("Start position: %lld", (long long)start);	5!
250	DFTRACER_UTILS_LOG_DEBUG("End position: %lld", (long long)end);	5!
251	DFTRACER_UTILS_LOG_DEBUG("Mode: %s", read_mode.c_str());	5!
252	DFTRACER_UTILS_LOG_DEBUG("Checkpoint size: %zu B (%zu MB)", checkpoint_size,	5!
253	checkpoint_size / (1024 * 1024));
254	DFTRACER_UTILS_LOG_DEBUG("Force rebuild: %s",	5!
255	force_rebuild ? "true" : "false");
256
257	if (checkpoint_size <= 0) {	5!
258	DFTRACER_UTILS_LOG_ERROR(	×
259	"%s",
260	"Checkpoint size must be positive (greater than 0 and in MB)");
261	return 1;	×
262	}
263
264	int test_fd = ::open(gz_path.c_str(), O_RDONLY);	5!
265	if (test_fd < 0) {	5✔
266	DFTRACER_UTILS_LOG_ERROR("File '%s' does not exist or cannot be opened",	1!
267	gz_path.c_str());
268	return 1;	1✔
269	}
270	::close(test_fd);	4!
271
272	const bool index_explicit = !index_path.empty();	4✔
273	if (index_path.empty()) {	4!
274	index_path = utilities::composites::dft::internal::determine_index_path(	4!
275	gz_path, index_dir);
276	}	4✔
277
278	#if DFTRACER_UTILS_LOGGER_DEBUG_ENABLED
279	ArchiveFormat format = IndexerFactory::detect_format(gz_path);	4!
280
281	DFTRACER_UTILS_LOG_DEBUG("Detected format: %s",	4!
282	format == ArchiveFormat::TAR_GZ ? "TAR.GZ"
283	: format == ArchiveFormat::GZIP ? "GZIP"
284	: "UNKNOWN");
285	#endif
286
287	// Skip for an explicit --index path (helper derives its own) or when
288	// --check/--force-rebuild already handle rebuilds in run_reader below.
289	if (!index_explicit && !check_rebuild && !force_rebuild) {	4!
290	cli::PipelineArgs refresh_pipeline;	3✔
291	refresh_pipeline.executor_threads =	3✔
292	dftracer_utils_hardware_concurrency();	3!
293	refresh_pipeline.io_threads = dftracer_utils_hardware_concurrency();	3!
294	cli::ensure_indexes_fresh_blocking("DFTracer Reader Index Refresh",	6!
295	refresh_pipeline, "", {gz_path},	3!
296	index_dir);
297	}	3✔
298
299	return run_reader(gz_path, index_path, checkpoint_size, force_rebuild,	12!
300	check_rebuild, read_mode, read_buffer_size, start, end)	4✔
301	.get();	4!
302	}	5✔

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