23529483807

Committed 25 Mar 2026 07:17AM UTC coverage: 48.515% (-1.6%) from 50.098%

Build # 23529483807

Build Type

Pull #57

github

Committed by

web-flow

Commit Message

Merge 5b1e117ad into 38f9f3616

Pull Request Pull Request #57: feat(comparator): add pairwise traces comparator

Coverage Stats

18829 of 49412 branches covered (38.11%)

Branch coverage included in aggregate %.

1584 of 1933 new or added lines in 14 files covered. (81.95%)

3552 existing lines in 135 files now uncovered.

18474 of 27477 relevant lines covered (67.23%)

241072.53 hits per line

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

39.5

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

#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/indexer/internal/tar/queries/queries.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 {

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_),
      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(), idx_path.c_str());
        indexer = std::make_shared<TarIndexer>(tar_gz_path, idx_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(), idx_path.c_str());
    } catch (const std::exception &e) {
        throw std::runtime_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 std::runtime_error("Invalid indexer provided");
    }
    is_open = true;
    tar_gz_path = indexer->get_tar_gz_path();
    idx_path = indexer->get_idx_path();
}

TarReader::~TarReader() = default;

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

llnl / dftracer-utils / 23529483807

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