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

37.97

/src/dftracer/utils/utilities/reader/internal/tar_reader.cpp

#include <dftracer/utils/core/common/filesystem.h>
#include <dftracer/utils/core/common/logging.h>
#include <dftracer/utils/core/coro/task.h>
#include <dftracer/utils/utilities/indexer/internal/indexer_factory.h>
#include <dftracer/utils/utilities/reader/error.h>
#include <dftracer/utils/utilities/reader/internal/streams/tar_byte_stream.h>
#include <dftracer/utils/utilities/reader/internal/string_line_processor.h>
#include <dftracer/utils/utilities/reader/internal/tar_reader.h>

#include <algorithm>
#include <cinttypes>
#include <cstring>
#include <sstream>
#include <vector>

using namespace dftracer::utils::utilities::indexer::internal::tar;

namespace dftracer::utils::utilities::reader::internal {

namespace {

std::string normalize_idx_path(const std::string &path) {
    fs::path input(path);
    if (input.filename() == ".dftindex") {
        return input.string();
    }
    if (input.parent_path().filename() == ".dftindex") {
        return input.parent_path().string();
    }
    if (input.has_extension()) {
        return (input.parent_path() / ".dftindex").string();
    }
    return (input / ".dftindex").string();
}

}  // namespace

TarReader::TarReader(const std::string &tar_gz_path_,
                     const std::string &idx_path_, std::size_t index_ckpt_size)
    : tar_gz_path(tar_gz_path_),
      index_path(normalize_idx_path(idx_path_)),
      is_open(false),
      default_buffer_size(DEFAULT_TAR_READER_BUFFER_SIZE),
      logical_mapping_cached(false),
      cached_total_logical_bytes(0),
      cached_total_logical_lines(0) {
    try {
        printf("Creating TAR reader for gz: %s and index: %s\n",
               tar_gz_path.c_str(), index_path.c_str());
        indexer = std::make_shared<TarIndexer>(tar_gz_path, index_path,
                                               index_ckpt_size, false);
        is_open = true;

        DFTRACER_UTILS_LOG_DEBUG(
            "Successfully created TAR reader for gz: %s and index: %s",
            tar_gz_path.c_str(), index_path.c_str());
    } catch (const std::exception &e) {
        throw ReaderError(ReaderError::INITIALIZATION_ERROR,
                          "Failed to initialize TAR reader with indexer: " +
                              std::string(e.what()));
    }
}

TarReader::TarReader(std::shared_ptr<TarIndexer> indexer_)
    : default_buffer_size(DEFAULT_TAR_READER_BUFFER_SIZE),
      indexer(indexer_),
      logical_mapping_cached(false),
      cached_total_logical_bytes(0),
      cached_total_logical_lines(0) {
    if (indexer == nullptr) {
        throw ReaderError(ReaderError::INVALID_ARGUMENT,
                          "Invalid indexer provided");
    }
    is_open = true;
    tar_gz_path = indexer->get_tar_gz_path();
    index_path = indexer->get_index_path();
}

TarReader::~TarReader() = default;

TarReader::TarReader(TarReader &&other) noexcept
    : tar_gz_path(std::move(other.tar_gz_path)),
      index_path(std::move(other.index_path)),
      is_open(other.is_open),
      default_buffer_size(other.default_buffer_size),
      indexer(std::move(other.indexer)),
      logical_mapping_cached(other.logical_mapping_cached),
      cached_file_mapping(std::move(other.cached_file_mapping)),
      cached_total_logical_bytes(other.cached_total_logical_bytes),
      cached_total_logical_lines(other.cached_total_logical_lines) {
    other.is_open = false;
    other.logical_mapping_cached = false;
}

TarReader &TarReader::operator=(TarReader &&other) noexcept {
    if (this != &other) {
        tar_gz_path = std::move(other.tar_gz_path);
        index_path = std::move(other.index_path);
        is_open = other.is_open;
        default_buffer_size = other.default_buffer_size;
        indexer = std::move(other.indexer);
        logical_mapping_cached = other.logical_mapping_cached;
        cached_file_mapping = std::move(other.cached_file_mapping);
        cached_total_logical_bytes = other.cached_total_logical_bytes;
        cached_total_logical_lines = other.cached_total_logical_lines;

        other.is_open = false;
        other.logical_mapping_cached = false;
    }
    return *this;
}

// Metadata operations
std::size_t TarReader::get_max_bytes() const {
    build_logical_mapping();
    return cached_total_logical_bytes;
}

std::size_t TarReader::get_num_lines() const {
    build_logical_mapping();
    return cached_total_logical_lines;
}

std::string TarReader::get_format_name() const { return "TAR.GZ"; }

const std::string &TarReader::get_archive_path() const { return tar_gz_path; }

const std::string &TarReader::get_index_path() const { return index_path; }

void TarReader::set_buffer_size(std::size_t size) {
    default_buffer_size = size;
}

std::unique_ptr<ReaderStream> TarReader::stream(
    [[maybe_unused]] const StreamConfig &config) {
    // TODO: Implement TAR stream creation
    throw ReaderError(ReaderError::UNKNOWN_ERROR,
                      "TarReader::stream() not yet implemented");
}

// Archive structure operations
std::vector<TarReader::TarFileInfo> TarReader::list_files() const {
    build_logical_mapping();
    return cached_file_mapping;
}

// Reader interface - operates on concatenated logical view
coro::CoroTask<std::size_t> TarReader::read_async(std::size_t start_bytes,
                                                  std::size_t end_bytes,
                                                  char *buffer,
                                                  std::size_t buffer_size) {
    co_return read_logical(start_bytes, end_bytes, buffer, buffer_size);
}

coro::CoroTask<std::size_t> TarReader::read_line_bytes_async(
    std::size_t start_bytes, std::size_t end_bytes, char *buffer,
    std::size_t buffer_size) {
    build_logical_mapping();

    if (start_bytes >= cached_total_logical_bytes || buffer_size == 0) {
        co_return 0;
    }

    std::size_t actual_end = std::min(
        end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));

    // Read a larger chunk to ensure we get complete lines
    std::size_t read_size = std::min(actual_end - start_bytes, buffer_size * 2);
    std::vector<char> temp_buffer(read_size);

    std::size_t bytes_read = read_logical(start_bytes, start_bytes + read_size,
                                          temp_buffer.data(), read_size);

    if (bytes_read == 0) {
        co_return 0;
    }

    // Find the first line start (if start_bytes is not at line beginning)
    std::size_t line_start = 0;
    if (start_bytes > 0) {
        // Look back to find the beginning of the current line
        std::size_t lookback_start =
            (start_bytes >= 512) ? start_bytes - 512 : 0;
        std::vector<char> lookback_buffer(512);
        std::size_t lookback_read = read_logical(lookback_start, start_bytes,
                                                 lookback_buffer.data(), 512);

        if (lookback_read > 0) {
            // Find the last newline in the lookback buffer
            for (std::size_t i = lookback_read; i > 0; --i) {
                if (lookback_buffer[i - 1] == '\n') {
                    // The line starts after this newline, but we need to adjust
                    // for the position difference
                    break;
                }
            }
        }
    }

    // Find complete lines within our read data
    std::size_t output_pos = 0;
    std::size_t search_pos = line_start;

    while (search_pos < bytes_read && output_pos < buffer_size) {
        // Find the next newline
        std::size_t newline_pos = search_pos;
        while (newline_pos < bytes_read && temp_buffer[newline_pos] != '\n') {
            newline_pos++;
        }

        if (newline_pos < bytes_read) {
            // Include the newline character
            newline_pos++;
            std::size_t line_len = newline_pos - search_pos;

            if (output_pos + line_len <= buffer_size) {
                std::memcpy(buffer + output_pos,
                            temp_buffer.data() + search_pos, line_len);
                output_pos += line_len;
                search_pos = newline_pos;
            } else {
                break;  // Buffer full
            }
        } else {
            // Incomplete line at the end, don't include it
            break;
        }
    }

    co_return output_pos;
}

coro::CoroTask<std::string> TarReader::read_lines_async(std::size_t start_line,
                                                        std::size_t end_line) {
    build_logical_mapping();

    if (start_line < 1 || start_line > cached_total_logical_lines) {
        co_return "";
    }

    std::size_t actual_end_line = std::min(
        end_line, static_cast<std::size_t>(cached_total_logical_lines));

    // Find which files contain the requested lines
    std::string result;
    std::size_t current_global_line = 1;

    for (const auto &file_info : cached_file_mapping) {
        // Check if this file contains any of our target lines
        if (current_global_line + file_info.estimated_lines <= start_line) {
            current_global_line += file_info.estimated_lines;
            continue;
        }
        if (current_global_line > actual_end_line) {
            break;
        }

        // Read content from this file
        std::size_t file_start_line =
            std::max(start_line, current_global_line) - current_global_line + 1;
        std::size_t file_end_line =
            std::min(actual_end_line,
                     static_cast<std::size_t>(current_global_line +
                                              file_info.estimated_lines - 1)) -
            current_global_line + 1;

        std::string file_content =
            read_file_content(file_info, 0, file_info.file_size);
        result += extract_lines_from_content(file_content, file_start_line,
                                             file_end_line);

        current_global_line += file_info.estimated_lines;
    }

    co_return result;
}

coro::CoroTask<void> TarReader::read_lines_with_processor_async(
    std::size_t start_line, std::size_t end_line, LineProcessor &processor) {
    build_logical_mapping();

    if (start_line < 1 || start_line > cached_total_logical_lines) {
        co_return;
    }

    std::size_t actual_end_line = std::min(
        end_line, static_cast<std::size_t>(cached_total_logical_lines));

    // Find which files contain the requested lines
    std::size_t current_global_line = 1;

    processor.begin(start_line, actual_end_line);

    for (const auto &file_info : cached_file_mapping) {
        // Check if this file contains any of our target lines
        if (current_global_line + file_info.estimated_lines <= start_line) {
            current_global_line += file_info.estimated_lines;
            continue;
        }
        if (current_global_line > actual_end_line) {
            break;
        }

        // Read content from this file
        std::string file_content =
            read_file_content(file_info, 0, file_info.file_size);
        process_content_lines(file_content, processor, current_global_line,
                              start_line, actual_end_line);

        current_global_line += file_info.estimated_lines;
    }

    processor.end();
}

coro::CoroTask<void> TarReader::read_line_bytes_with_processor_async(
    std::size_t start_bytes, std::size_t end_bytes, LineProcessor &processor) {
    build_logical_mapping();

    if (start_bytes >= cached_total_logical_bytes) {
        co_return;
    }

    std::size_t actual_end = std::min(
        end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));

    processor.begin(start_bytes, actual_end);

    // Find which files contain the requested byte range
    for (const auto &file_info : cached_file_mapping) {
        if (file_info.logical_end_offset <= start_bytes) {
            continue;  // This file is before our range
        }
        if (file_info.logical_start_offset >= actual_end) {
            break;     // We've passed our range
        }

        // Calculate the portion of this file we need
        std::size_t file_start =
            (start_bytes > file_info.logical_start_offset)
                ? start_bytes - file_info.logical_start_offset
                : 0;
        std::size_t file_end = std::min(
            file_info.file_size, actual_end - file_info.logical_start_offset);

        if (file_start < file_end) {
            std::string file_content =
                read_file_content(file_info, file_start, file_end - file_start);
            co_await processor.process(file_content.c_str(),
                                       file_content.length());
        }
    }

    processor.end();
}

void TarReader::reset() {
    logical_mapping_cached = false;
    cached_file_mapping.clear();
    cached_total_logical_bytes = 0;
    cached_total_logical_lines = 0;
}

bool TarReader::is_valid() const { return is_open && indexer; }

// File-level operations - NOT YET IMPLEMENTED
std::string TarReader::read_file(const std::string &filename,
                                 std::size_t start_line, std::size_t end_line) {
    (void)filename;
    (void)start_line;
    (void)end_line;  // Suppress unused warnings
    return "";       // Placeholder
}

bool TarReader::find_file(const std::string &filename,
                          TarFileInfo &file_info) const {
    build_logical_mapping();

    auto it =
        std::find_if(cached_file_mapping.begin(), cached_file_mapping.end(),
                     [&filename](const TarFileInfo &info) {
                         return info.file_name == filename;
                     });

    if (it != cached_file_mapping.end()) {
        file_info = *it;
        return true;
    }

    return false;
}

// Internal helper methods implementation moved from TarReaderImplementor
void TarReader::build_logical_mapping() const {
    if (logical_mapping_cached) {
        return;
    }

    DFTRACER_UTILS_LOG_DEBUG(
        "%s", "Building logical content mapping for TAR.GZ file");

    try {
        // Get all TAR file entries from the indexer
        auto tar_files = indexer->list_files();

        cached_file_mapping.clear();
        cached_file_mapping.reserve(tar_files.size());

        std::uint64_t logical_offset = 0;
        std::uint64_t logical_line = 1;

        for (const auto &tar_file : tar_files) {
            TarFileInfo file_info;
            file_info.file_name = tar_file.file_name;
            file_info.file_size = tar_file.file_size;
            file_info.file_mtime = tar_file.file_mtime;
            file_info.typeflag = tar_file.typeflag;
            file_info.logical_start_offset = logical_offset;
            file_info.logical_start_line = logical_line;

            // Calculate end positions
            file_info.logical_end_offset = logical_offset + tar_file.file_size;

            // Estimate lines in this file (rough approximation)
            file_info.estimated_lines =
                tar_file.file_size > 0 ? (tar_file.file_size / 50) + 1 : 1;
            file_info.logical_end_line =
                logical_line + file_info.estimated_lines - 1;

            cached_file_mapping.push_back(file_info);

            logical_offset = file_info.logical_end_offset;
            logical_line = file_info.logical_end_line + 1;
        }

        cached_total_logical_bytes = logical_offset;
        cached_total_logical_lines = logical_line - 1;
        logical_mapping_cached = true;

        DFTRACER_UTILS_LOG_DEBUG("Built logical mapping: %zu files, %" PRIu64
                                 " bytes, %" PRIu64 " lines",
                                 cached_file_mapping.size(),
                                 cached_total_logical_bytes,
                                 cached_total_logical_lines);
    } catch (const std::exception &e) {
        throw ReaderError(
            ReaderError::READ_ERROR,
            "Failed to build TAR logical mapping: " + std::string(e.what()));
    }
}

std::size_t TarReader::read_logical(std::size_t start_bytes,
                                    std::size_t end_bytes, char *buffer,
                                    std::size_t buffer_size) const {
    build_logical_mapping();

    if (start_bytes >= cached_total_logical_bytes || buffer_size == 0) {
        return 0;
    }

    std::size_t actual_end = std::min(
        end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));
    [[maybe_unused]] std::size_t total_to_read = actual_end - start_bytes;
    [[maybe_unused]] std::size_t total_written = 0;

    DFTRACER_UTILS_LOG_DEBUG(
        "TAR logical read: start=%zu, end=%zu, buffer_size=%zu", start_bytes,
        actual_end, buffer_size);

    // Find which files contain the requested byte range
    std::size_t bytes_written = 0;

    for (const auto &file_info : cached_file_mapping) {
        if (file_info.logical_end_offset <= start_bytes) {
            continue;  // This file is before our range
        }
        if (file_info.logical_start_offset >= actual_end) {
            break;     // We've passed our range
        }
        if (bytes_written >= buffer_size) {
            break;     // Buffer is full
        }

        // Calculate the portion of this file we need
        std::size_t file_start =
            (start_bytes > file_info.logical_start_offset)
                ? start_bytes - file_info.logical_start_offset
                : 0;
        std::size_t file_end = std::min(
            file_info.file_size, actual_end - file_info.logical_start_offset);

        if (file_start < file_end) {
            std::size_t bytes_to_read =
                std::min(file_end - file_start, buffer_size - bytes_written);
            std::string file_content =
                read_file_content(file_info, file_start, bytes_to_read);

            std::memcpy(buffer + bytes_written, file_content.c_str(),
                        file_content.length());
            bytes_written += file_content.length();
        }
    }

    DFTRACER_UTILS_LOG_DEBUG("TAR logical read: returning %zu bytes",
                             bytes_written);
    return bytes_written;
}

// Helper method implementations
std::string TarReader::read_file_content(const TarFileInfo &file_info,
                                         std::size_t offset,
                                         std::size_t size) const {
    if (size == 0) {
        return "";
    }

    try {
        // Calculate the absolute offset in the logical stream where the file
        // data starts
        std::size_t file_data_start = file_info.logical_start_offset + offset;
        std::size_t file_data_end = file_data_start + size;

        // Ensure we don't read beyond the file boundaries
        std::size_t max_end =
            file_info.logical_start_offset + file_info.file_size;
        if (file_data_end > max_end) {
            file_data_end = max_end;
        }

        if (file_data_start >= file_data_end) {
            return "";
        }

        // Use a buffer to read the data
        std::size_t actual_size = file_data_end - file_data_start;
        std::vector<char> buffer(actual_size);

        // Create a TAR byte stream to read file content
        auto byte_stream = std::make_unique<TarByteStream>();
        byte_stream->initialize(tar_gz_path, file_data_start, file_data_end,
                                *indexer);

        std::size_t total_read = 0;
        while (total_read < actual_size) {
            std::size_t chunk_size = std::min(actual_size - total_read,
                                              static_cast<std::size_t>(8192));
            std::size_t bytes_read =
                byte_stream->read(buffer.data() + total_read, chunk_size);

            if (bytes_read == 0) {
                break;  // EOF or error
            }

            total_read += bytes_read;
        }

        return std::string(buffer.data(), total_read);

    } catch (const std::exception &e) {
        DFTRACER_UTILS_LOG_DEBUG("Error reading TAR file content: %s",
                                 e.what());
        return "";
    }
}

std::string TarReader::extract_lines_from_content(const std::string &content,
                                                  std::size_t start_line,
                                                  std::size_t end_line) const {
    if (content.empty()) {
        return "";
    }

    std::istringstream stream(content);
    std::string line;
    std::string result;
    std::size_t current_line = 1;

    while (std::getline(stream, line) && current_line <= end_line) {
        if (current_line >= start_line) {
            result += line + "\n";
        }
        current_line++;
    }

    return result;
}

void TarReader::process_content_lines(const std::string &content,
                                      LineProcessor &processor,
                                      std::size_t base_line,
                                      std::size_t start_line,
                                      std::size_t end_line) const {
    if (content.empty()) {
        return;
    }

    std::istringstream stream(content);
    std::string line;
    std::size_t current_line = base_line;

    while (std::getline(stream, line) && current_line <= end_line) {
        if (current_line >= start_line) {
            // Sync C API path — .get() intentional
            if (!processor.process(line.c_str(), line.length()).get()) {
                break;
            }
        }
        current_line++;
    }
}

}  // namespace dftracer::utils::utilities::reader::internal

1	#include <dftracer/utils/core/common/filesystem.h>
2	#include <dftracer/utils/core/common/logging.h>
3	#include <dftracer/utils/core/coro/task.h>
4	#include <dftracer/utils/utilities/indexer/internal/indexer_factory.h>
5	#include <dftracer/utils/utilities/reader/error.h>
6	#include <dftracer/utils/utilities/reader/internal/streams/tar_byte_stream.h>
7	#include <dftracer/utils/utilities/reader/internal/string_line_processor.h>
8	#include <dftracer/utils/utilities/reader/internal/tar_reader.h>
9
10	#include <algorithm>
11	#include <cinttypes>
12	#include <cstring>
13	#include <sstream>
14	#include <vector>
15
16	using namespace dftracer::utils::utilities::indexer::internal::tar;
17
18	namespace dftracer::utils::utilities::reader::internal {
19
20	namespace {
21
22	std::string normalize_idx_path(const std::string &path) {	1✔
23	fs::path input(path);	1✔
24	if (input.filename() == ".dftindex") {	1!
25	return input.string();	1!
26	}
27	if (input.parent_path().filename() == ".dftindex") {	×
28	return input.parent_path().string();	×
29	}
30	if (input.has_extension()) {	×
31	return (input.parent_path() / ".dftindex").string();	×
32	}
33	return (input / ".dftindex").string();	×
34	}	1✔
35
36	} // namespace
37
38	TarReader::TarReader(const std::string &tar_gz_path_,	4✔
39	const std::string &idx_path_, std::size_t index_ckpt_size)
40	: tar_gz_path(tar_gz_path_),	1!
41	index_path(normalize_idx_path(idx_path_)),	1!
42	is_open(false),	1✔
43	default_buffer_size(DEFAULT_TAR_READER_BUFFER_SIZE),	1✔
44	logical_mapping_cached(false),	1✔
45	cached_total_logical_bytes(0),	1✔
46	cached_total_logical_lines(0) {	3✔
47	try {
48	printf("Creating TAR reader for gz: %s and index: %s\n",	1!
49	tar_gz_path.c_str(), index_path.c_str());	1✔
50	indexer = std::make_shared<TarIndexer>(tar_gz_path, index_path,	2!
51	index_ckpt_size, false);	1✔
52	is_open = true;	1✔
53
54	DFTRACER_UTILS_LOG_DEBUG(	1!
55	"Successfully created TAR reader for gz: %s and index: %s",
56	tar_gz_path.c_str(), index_path.c_str());
57	} catch (const std::exception &e) {	1!
58	throw ReaderError(ReaderError::INITIALIZATION_ERROR,	×
59	"Failed to initialize TAR reader with indexer: " +	×
60	std::string(e.what()));	×
61	}	×
62	}	2✔
63
64	TarReader::TarReader(std::shared_ptr<TarIndexer> indexer_)	44✔
65	: default_buffer_size(DEFAULT_TAR_READER_BUFFER_SIZE),	11✔
66	indexer(indexer_),	11✔
67	logical_mapping_cached(false),	11✔
68	cached_total_logical_bytes(0),	11✔
69	cached_total_logical_lines(0) {	33✔
70	if (indexer == nullptr) {	11!
71	throw ReaderError(ReaderError::INVALID_ARGUMENT,	×
72	"Invalid indexer provided");	×
73	}
74	is_open = true;	11✔
75	tar_gz_path = indexer->get_tar_gz_path();	11!
76	index_path = indexer->get_index_path();	11!
77	}	22✔
78
79	TarReader::~TarReader() = default;	24✔
80
81	TarReader::TarReader(TarReader &&other) noexcept	×
82	: tar_gz_path(std::move(other.tar_gz_path)),	×
83	index_path(std::move(other.index_path)),	×
84	is_open(other.is_open),	×
85	default_buffer_size(other.default_buffer_size),	×
86	indexer(std::move(other.indexer)),	×
87	logical_mapping_cached(other.logical_mapping_cached),	×
88	cached_file_mapping(std::move(other.cached_file_mapping)),	×
89	cached_total_logical_bytes(other.cached_total_logical_bytes),	×
90	cached_total_logical_lines(other.cached_total_logical_lines) {	×
91	other.is_open = false;	×
92	other.logical_mapping_cached = false;	×
93	}	×
94
95	TarReader &TarReader::operator=(TarReader &&other) noexcept {	×
96	if (this != &other) {	×
97	tar_gz_path = std::move(other.tar_gz_path);	×
98	index_path = std::move(other.index_path);	×
99	is_open = other.is_open;	×
100	default_buffer_size = other.default_buffer_size;	×
101	indexer = std::move(other.indexer);	×
102	logical_mapping_cached = other.logical_mapping_cached;	×
103	cached_file_mapping = std::move(other.cached_file_mapping);	×
104	cached_total_logical_bytes = other.cached_total_logical_bytes;	×
105	cached_total_logical_lines = other.cached_total_logical_lines;	×
106
107	other.is_open = false;	×
108	other.logical_mapping_cached = false;	×
UNCOV 109	}	×
110	return *this;	×
111	}
112
113	// Metadata operations
114	std::size_t TarReader::get_max_bytes() const {	1✔
115	build_logical_mapping();	1✔
116	return cached_total_logical_bytes;	1✔
117	}
118
119	std::size_t TarReader::get_num_lines() const {	×
120	build_logical_mapping();	×
121	return cached_total_logical_lines;	×
122	}
123
124	std::string TarReader::get_format_name() const { return "TAR.GZ"; }	×
125
126	const std::string &TarReader::get_archive_path() const { return tar_gz_path; }	1✔
127
128	const std::string &TarReader::get_index_path() const { return index_path; }	1✔
129
130	void TarReader::set_buffer_size(std::size_t size) {	×
131	default_buffer_size = size;	×
132	}	×
133
134	std::unique_ptr<ReaderStream> TarReader::stream(	×
135	[[maybe_unused]] const StreamConfig &config) {
136	// TODO: Implement TAR stream creation
137	throw ReaderError(ReaderError::UNKNOWN_ERROR,	×
138	"TarReader::stream() not yet implemented");	×
UNCOV 139	}	×
140
141	// Archive structure operations
142	std::vector<TarReader::TarFileInfo> TarReader::list_files() const {	×
143	build_logical_mapping();	×
144	return cached_file_mapping;	×
145	}
146
147	// Reader interface - operates on concatenated logical view
148	coro::CoroTask<std::size_t> TarReader::read_async(std::size_t start_bytes,	765!
149	std::size_t end_bytes,
150	char *buffer,
151	std::size_t buffer_size) {	153!
152	co_return read_logical(start_bytes, end_bytes, buffer, buffer_size);	306!
UNCOV 153	}	×
154
155	coro::CoroTask<std::size_t> TarReader::read_line_bytes_async(	5!
156	std::size_t start_bytes, std::size_t end_bytes, char *buffer,
157	std::size_t buffer_size) {	1!
158	build_logical_mapping();	1!
159
160	if (start_bytes >= cached_total_logical_bytes \|\| buffer_size == 0) {	1!
161	co_return 0;	1!
162	}
163
164	std::size_t actual_end = std::min(	2!
165	end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));	1✔
166
167	// Read a larger chunk to ensure we get complete lines
168	std::size_t read_size = std::min(actual_end - start_bytes, buffer_size * 2);	1!
169	std::vector<char> temp_buffer(read_size);	1!
170
171	std::size_t bytes_read = read_logical(start_bytes, start_bytes + read_size,	2!
172	temp_buffer.data(), read_size);	1✔
173
174	if (bytes_read == 0) {	1!
UNCOV 175	co_return 0;	×
176	}
177
178	// Find the first line start (if start_bytes is not at line beginning)
179	std::size_t line_start = 0;	1✔
180	if (start_bytes > 0) {	1!
181	// Look back to find the beginning of the current line
UNCOV 182	std::size_t lookback_start =	×
UNCOV 183	(start_bytes >= 512) ? start_bytes - 512 : 0;	×
UNCOV 184	std::vector<char> lookback_buffer(512);	×
UNCOV 185	std::size_t lookback_read = read_logical(lookback_start, start_bytes,	×
UNCOV 186	lookback_buffer.data(), 512);	×
187
UNCOV 188	if (lookback_read > 0) {	×
189	// Find the last newline in the lookback buffer
UNCOV 190	for (std::size_t i = lookback_read; i > 0; --i) {	×
UNCOV 191	if (lookback_buffer[i - 1] == '\n') {	×
192	// The line starts after this newline, but we need to adjust
193	// for the position difference
UNCOV 194	break;	×
195	}
UNCOV 196	}	×
UNCOV 197	}	×
UNCOV 198	}	×
199
200	// Find complete lines within our read data
201	std::size_t output_pos = 0;	1✔
202	std::size_t search_pos = line_start;	1✔
203
204	while (search_pos < bytes_read && output_pos < buffer_size) {	17!
205	// Find the next newline
206	std::size_t newline_pos = search_pos;	17✔
207	while (newline_pos < bytes_read && temp_buffer[newline_pos] != '\n') {	1,036!
208	newline_pos++;	1,019✔
209	}
210
211	if (newline_pos < bytes_read) {	17!
212	// Include the newline character
213	newline_pos++;	17✔
214	std::size_t line_len = newline_pos - search_pos;	17✔
215
216	if (output_pos + line_len <= buffer_size) {	17✔
217	std::memcpy(buffer + output_pos,	32✔
218	temp_buffer.data() + search_pos, line_len);	16✔
219	output_pos += line_len;	16✔
220	search_pos = newline_pos;	16✔
221	} else {	16✔
222	break; // Buffer full	1✔
223	}
224	} else {	17✔
225	// Incomplete line at the end, don't include it
UNCOV 226	break;	×
227	}
228	}	17✔
229
230	co_return output_pos;	1!
231	}	1✔
232
233	coro::CoroTask<std::string> TarReader::read_lines_async(std::size_t start_line,	80!
234	std::size_t end_line) {	16!
235	build_logical_mapping();	16!
236
237	if (start_line < 1 \|\| start_line > cached_total_logical_lines) {	16!
238	co_return "";	16!
239	}
240
241	std::size_t actual_end_line = std::min(	32!
242	end_line, static_cast<std::size_t>(cached_total_logical_lines));	16✔
243
244	// Find which files contain the requested lines
245	std::string result;	16✔
246	std::size_t current_global_line = 1;	16✔
247
248	for (const auto &file_info : cached_file_mapping) {	50!
249	// Check if this file contains any of our target lines
250	if (current_global_line + file_info.estimated_lines <= start_line) {	34✔
251	current_global_line += file_info.estimated_lines;	2✔
252	continue;	2✔
253	}
254	if (current_global_line > actual_end_line) {	32✔
255	break;	16✔
256	}
257
258	// Read content from this file
259	std::size_t file_start_line =	16✔
260	std::max(start_line, current_global_line) - current_global_line + 1;	16!
261	std::size_t file_end_line =	16✔
262	std::min(actual_end_line,	16!
263	static_cast<std::size_t>(current_global_line +	48✔
264	file_info.estimated_lines - 1)) -	32✔
265	current_global_line + 1;	32✔
266
267	std::string file_content =	16✔
268	read_file_content(file_info, 0, file_info.file_size);	16!
269	result += extract_lines_from_content(file_content, file_start_line,	32!
270	file_end_line);	16✔
271
272	current_global_line += file_info.estimated_lines;	16✔
273	}	34✔
274
275	co_return result;	16!
276	}	16✔
277
278	coro::CoroTask<void> TarReader::read_lines_with_processor_async(	×
UNCOV 279	std::size_t start_line, std::size_t end_line, LineProcessor &processor) {	×
UNCOV 280	build_logical_mapping();	×
281
UNCOV 282	if (start_line < 1 \|\| start_line > cached_total_logical_lines) {	×
UNCOV 283	co_return;	×
284	}
285
UNCOV 286	std::size_t actual_end_line = std::min(	×
UNCOV 287	end_line, static_cast<std::size_t>(cached_total_logical_lines));	×
288
289	// Find which files contain the requested lines
UNCOV 290	std::size_t current_global_line = 1;	×
291
UNCOV 292	processor.begin(start_line, actual_end_line);	×
293
UNCOV 294	for (const auto &file_info : cached_file_mapping) {	×
295	// Check if this file contains any of our target lines
UNCOV 296	if (current_global_line + file_info.estimated_lines <= start_line) {	×
UNCOV 297	current_global_line += file_info.estimated_lines;	×
UNCOV 298	continue;	×
299	}
UNCOV 300	if (current_global_line > actual_end_line) {	×
UNCOV 301	break;	×
302	}
303
304	// Read content from this file
UNCOV 305	std::string file_content =	×
UNCOV 306	read_file_content(file_info, 0, file_info.file_size);	×
UNCOV 307	process_content_lines(file_content, processor, current_global_line,	×
UNCOV 308	start_line, actual_end_line);	×
309
UNCOV 310	current_global_line += file_info.estimated_lines;	×
UNCOV 311	}	×
312
UNCOV 313	processor.end();	×
314	}	×
315
316	coro::CoroTask<void> TarReader::read_line_bytes_with_processor_async(	×
UNCOV 317	std::size_t start_bytes, std::size_t end_bytes, LineProcessor &processor) {	×
UNCOV 318	build_logical_mapping();	×
319
UNCOV 320	if (start_bytes >= cached_total_logical_bytes) {	×
UNCOV 321	co_return;	×
322	}
323
UNCOV 324	std::size_t actual_end = std::min(	×
UNCOV 325	end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));	×
326
UNCOV 327	processor.begin(start_bytes, actual_end);	×
328
329	// Find which files contain the requested byte range
UNCOV 330	for (const auto &file_info : cached_file_mapping) {	×
UNCOV 331	if (file_info.logical_end_offset <= start_bytes) {	×
UNCOV 332	continue; // This file is before our range	×
333	}
UNCOV 334	if (file_info.logical_start_offset >= actual_end) {	×
UNCOV 335	break; // We've passed our range	×
336	}
337
338	// Calculate the portion of this file we need
UNCOV 339	std::size_t file_start =	×
UNCOV 340	(start_bytes > file_info.logical_start_offset)	×
UNCOV 341	? start_bytes - file_info.logical_start_offset	×
342	: 0;
UNCOV 343	std::size_t file_end = std::min(	×
UNCOV 344	file_info.file_size, actual_end - file_info.logical_start_offset);	×
345
UNCOV 346	if (file_start < file_end) {	×
UNCOV 347	std::string file_content =	×
UNCOV 348	read_file_content(file_info, file_start, file_end - file_start);	×
UNCOV 349	co_await processor.process(file_content.c_str(),	×
UNCOV 350	file_content.length());	×
UNCOV 351	}	×
UNCOV 352	}	×
353
UNCOV 354	processor.end();	×
355	}	×
356
357	void TarReader::reset() {	×
358	logical_mapping_cached = false;	×
359	cached_file_mapping.clear();	×
360	cached_total_logical_bytes = 0;	×
361	cached_total_logical_lines = 0;	×
362	}	×
363
364	bool TarReader::is_valid() const { return is_open && indexer; }	3!
365
366	// File-level operations - NOT YET IMPLEMENTED
367	std::string TarReader::read_file(const std::string &filename,	×
368	std::size_t start_line, std::size_t end_line) {
UNCOV 369	(void)filename;	×
370	(void)start_line;
371	(void)end_line; // Suppress unused warnings
372	return ""; // Placeholder	×
373	}
374
375	bool TarReader::find_file(const std::string &filename,	×
376	TarFileInfo &file_info) const {
377	build_logical_mapping();	×
378
379	auto it =
380	std::find_if(cached_file_mapping.begin(), cached_file_mapping.end(),	×
381	[&filename](const TarFileInfo &info) {	×
382	return info.file_name == filename;	×
383	});
384
385	if (it != cached_file_mapping.end()) {	×
386	file_info = *it;	×
387	return true;	×
388	}
389
390	return false;	×
UNCOV 391	}	×
392
393	// Internal helper methods implementation moved from TarReaderImplementor
394	void TarReader::build_logical_mapping() const {	172✔
395	if (logical_mapping_cached) {	172✔
396	return;	163✔
397	}
398
399	DFTRACER_UTILS_LOG_DEBUG(	9!
400	"%s", "Building logical content mapping for TAR.GZ file");
401
402	try {
403	// Get all TAR file entries from the indexer
404	auto tar_files = indexer->list_files();	9!
405
406	cached_file_mapping.clear();	9✔
407	cached_file_mapping.reserve(tar_files.size());	9!
408
409	std::uint64_t logical_offset = 0;	9✔
410	std::uint64_t logical_line = 1;	9✔
411
412	for (const auto &tar_file : tar_files) {	45✔
413	TarFileInfo file_info;	36✔
414	file_info.file_name = tar_file.file_name;	36!
415	file_info.file_size = tar_file.file_size;	36✔
416	file_info.file_mtime = tar_file.file_mtime;	36✔
417	file_info.typeflag = tar_file.typeflag;	36✔
418	file_info.logical_start_offset = logical_offset;	36✔
419	file_info.logical_start_line = logical_line;	36✔
420
421	// Calculate end positions
422	file_info.logical_end_offset = logical_offset + tar_file.file_size;	36✔
423
424	// Estimate lines in this file (rough approximation)
425	file_info.estimated_lines =	36✔
426	tar_file.file_size > 0 ? (tar_file.file_size / 50) + 1 : 1;	36!
427	file_info.logical_end_line =	36✔
428	logical_line + file_info.estimated_lines - 1;	36✔
429
430	cached_file_mapping.push_back(file_info);	36!
431
432	logical_offset = file_info.logical_end_offset;	36✔
433	logical_line = file_info.logical_end_line + 1;	36✔
434	}	36✔
435
436	cached_total_logical_bytes = logical_offset;	9✔
437	cached_total_logical_lines = logical_line - 1;	9✔
438	logical_mapping_cached = true;	9✔
439
440	DFTRACER_UTILS_LOG_DEBUG("Built logical mapping: %zu files, %" PRIu64	9!
441	" bytes, %" PRIu64 " lines",
442	cached_file_mapping.size(),
443	cached_total_logical_bytes,
444	cached_total_logical_lines);
445	} catch (const std::exception &e) {	9!
446	throw ReaderError(	×
447	ReaderError::READ_ERROR,
448	"Failed to build TAR logical mapping: " + std::string(e.what()));	×
449	}	×
450	}	172✔
451
452	std::size_t TarReader::read_logical(std::size_t start_bytes,	154✔
453	std::size_t end_bytes, char *buffer,
454	std::size_t buffer_size) const {
455	build_logical_mapping();	154✔
456
457	if (start_bytes >= cached_total_logical_bytes \|\| buffer_size == 0) {	154!
458	return 0;	2✔
459	}
460
461	std::size_t actual_end = std::min(	152✔
462	end_bytes, static_cast<std::size_t>(cached_total_logical_bytes));	152✔
463	[[maybe_unused]] std::size_t total_to_read = actual_end - start_bytes;	152✔
464	[[maybe_unused]] std::size_t total_written = 0;	152✔
465
466	DFTRACER_UTILS_LOG_DEBUG(	152!
467	"TAR logical read: start=%zu, end=%zu, buffer_size=%zu", start_bytes,
468	actual_end, buffer_size);
469
470	// Find which files contain the requested byte range
471	std::size_t bytes_written = 0;	152✔
472
473	for (const auto &file_info : cached_file_mapping) {	364✔
474	if (file_info.logical_end_offset <= start_bytes) {	361✔
475	continue; // This file is before our range	49✔
476	}
477	if (file_info.logical_start_offset >= actual_end) {	312✔
478	break; // We've passed our range	147✔
479	}
480	if (bytes_written >= buffer_size) {	165✔
481	break; // Buffer is full	2✔
482	}
483
484	// Calculate the portion of this file we need
485	std::size_t file_start =	163✔
486	(start_bytes > file_info.logical_start_offset)	163✔
487	? start_bytes - file_info.logical_start_offset	146✔
488	: 0;
489	std::size_t file_end = std::min(	163✔
490	file_info.file_size, actual_end - file_info.logical_start_offset);	163✔
491
492	if (file_start < file_end) {	163!
493	std::size_t bytes_to_read =	163✔
494	std::min(file_end - file_start, buffer_size - bytes_written);	163✔
495	std::string file_content =
496	read_file_content(file_info, file_start, bytes_to_read);	163✔
497
498	std::memcpy(buffer + bytes_written, file_content.c_str(),	326✔
499	file_content.length());	163✔
500	bytes_written += file_content.length();	163✔
501	}	163✔
502	}
503
504	DFTRACER_UTILS_LOG_DEBUG("TAR logical read: returning %zu bytes",	152!
505	bytes_written);
506	return bytes_written;	152✔
507	}	154✔
508
509	// Helper method implementations
510	std::string TarReader::read_file_content(const TarFileInfo &file_info,	179✔
511	std::size_t offset,
512	std::size_t size) const {
513	if (size == 0) {	179!
514	return "";	×
515	}
516
517	try {
518	// Calculate the absolute offset in the logical stream where the file
519	// data starts
520	std::size_t file_data_start = file_info.logical_start_offset + offset;	179✔
521	std::size_t file_data_end = file_data_start + size;	179✔
522
523	// Ensure we don't read beyond the file boundaries
524	std::size_t max_end =	179✔
525	file_info.logical_start_offset + file_info.file_size;	179✔
526	if (file_data_end > max_end) {	179!
527	file_data_end = max_end;	×
UNCOV 528	}	×
529
530	if (file_data_start >= file_data_end) {	179!
531	return "";	×
532	}
533
534	// Use a buffer to read the data
535	std::size_t actual_size = file_data_end - file_data_start;	179✔
536	std::vector<char> buffer(actual_size);	179!
537
538	// Create a TAR byte stream to read file content
539	auto byte_stream = std::make_unique<TarByteStream>();	179!
540	byte_stream->initialize(tar_gz_path, file_data_start, file_data_end,	358!
541	*indexer);	179✔
542
543	std::size_t total_read = 0;	179✔
544	while (total_read < actual_size) {	391✔
545	std::size_t chunk_size = std::min(actual_size - total_read,	424!
546	static_cast<std::size_t>(8192));	212✔
547	std::size_t bytes_read =	212✔
548	byte_stream->read(buffer.data() + total_read, chunk_size);	212!
549
550	if (bytes_read == 0) {	212!
551	break; // EOF or error	×
552	}
553
554	total_read += bytes_read;	212✔
555	}
556
557	return std::string(buffer.data(), total_read);	179!
558
559	} catch (const std::exception &e) {	179!
560	DFTRACER_UTILS_LOG_DEBUG("Error reading TAR file content: %s",	×
561	e.what());
562	return "";	×
563	}	×
564	}	179✔
565
566	std::string TarReader::extract_lines_from_content(const std::string &content,	16✔
567	std::size_t start_line,
568	std::size_t end_line) const {
569	if (content.empty()) {	16!
570	return "";	×
571	}
572
573	std::istringstream stream(content);	16✔
574	std::string line;	16✔
575	std::string result;	16✔
576	std::size_t current_line = 1;	16✔
577
578	while (std::getline(stream, line) && current_line <= end_line) {	325!
579	if (current_line >= start_line) {	309✔
580	result += line + "\n";	16!
581	}	16✔
582	current_line++;	309✔
583	}
584
585	return result;	16✔
586	}	16!
587
588	void TarReader::process_content_lines(const std::string &content,	×
589	LineProcessor &processor,
590	std::size_t base_line,
591	std::size_t start_line,
592	std::size_t end_line) const {
593	if (content.empty()) {	×
594	return;	×
595	}
596
597	std::istringstream stream(content);	×
598	std::string line;	×
599	std::size_t current_line = base_line;	×
600
601	while (std::getline(stream, line) && current_line <= end_line) {	×
602	if (current_line >= start_line) {	×
603	// Sync C API path — .get() intentional
604	if (!processor.process(line.c_str(), line.length()).get()) {	×
605	break;	×
606	}
UNCOV 607	}	×
608	current_line++;	×
609	}
610	}	×
611
612	} // namespace dftracer::utils::utilities::reader::internal

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