28496595030

Committed 01 Jul 2026 05:50AM UTC coverage: 50.727% (-1.6%) from 52.278%

Build # 28496595030

Build Type

Pull #83

github

Committed by

web-flow

Commit Message

Merge 8f1ff4df5 into 2efed6649

Pull Request Pull Request #83: refactor and improve code QoL

Coverage Stats

31872 of 80367 branches covered (39.66%)

Branch coverage included in aggregate %.

770 of 1591 new or added lines in 85 files covered. (48.4%)

5070 existing lines in 182 files now uncovered.

32742 of 47009 relevant lines covered (69.65%)

9887.52 hits per line

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

38.19

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

llnl / dftracer-utils / 28496595030

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