read_binary_stream.hpp

/*
 * This file is part of libtcspc
 * Copyright 2019-2026 Board of Regents of the University of Wisconsin System
 * SPDX-License-Identifier: MIT
 */
 
#pragma once
 
#include "arg_wrappers.hpp"
#include "bucket.hpp"
#include "common.hpp"
#include "core.hpp"
#include "errors.hpp"
#include "int_types.hpp"
#include "introspect.hpp"
#include "processor.hpp"
 
#include <algorithm>
#include <cerrno>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <fstream>
#include <ios>
#include <istream>
#include <limits>
#include <memory>
#include <optional>
#include <span>
#include <stdexcept>
#include <string>
#include <system_error>
#include <type_traits>
#include <utility>
#include <vector>
 
// When editing this file, maintain partial symmetry with
// write_binary_stream.hpp.
 
namespace tcspc {
 
namespace internal {
 
struct null_input_stream {
    static auto is_error() noexcept -> bool { return false; }
    static auto is_eof() noexcept -> bool { return true; }
    static auto is_good() noexcept -> bool { return false; }
    static auto tell() noexcept -> std::optional<std::uint64_t> { return 0; }
    static auto skip(std::uint64_t bytes) noexcept -> bool {
        return bytes == 0;
    }
    static auto read(std::span<std::byte> /* buffer */) noexcept
        -> std::uint64_t {
        return 0;
    }
};
 
// We turn off istream exceptions in the constructor.
// NOLINTBEGIN(bugprone-exception-escape)
template <typename IStream> class istream_input_stream {
    static_assert(std::is_base_of_v<std::istream, IStream>);
    IStream stream;
 
  public:
    explicit istream_input_stream(IStream stream) : stream(std::move(stream)) {
        this->stream.exceptions(std::ios::goodbit);
    }
 
    auto is_error() noexcept -> bool {
        auto const flags = stream.rdstate();
        return ((flags & std::ios::failbit) || (flags & std::ios::badbit)) &&
               not(flags & std::ios::eofbit);
    }
 
    auto is_eof() noexcept -> bool { return stream.eof(); }
 
    auto is_good() noexcept -> bool { return stream.good(); }
 
    auto tell() noexcept -> std::optional<std::uint64_t> {
        if (stream.fail())
            return std::nullopt; // Do not affect flags.
        std::int64_t const pos = stream.tellg();
        if (pos >= 0)
            return std::uint64_t(pos);
        stream.clear();
        return std::nullopt;
    }
 
    auto skip(std::uint64_t bytes) noexcept -> bool {
        if (stream.fail() ||
            bytes > std::uint64_t(std::numeric_limits<std::streamoff>::max()))
            return false;
        stream.seekg(std::streamoff(bytes), std::ios::cur);
        auto const ret = stream.good();
        stream.clear();
        return ret;
    }
 
    auto read(std::span<std::byte> buffer) noexcept -> std::uint64_t {
        // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast)
        stream.read(reinterpret_cast<char *>(buffer.data()),
                    static_cast<std::streamsize>(buffer.size()));
        return static_cast<std::uint64_t>(stream.gcount());
    }
};
// NOLINTEND(bugprone-exception-escape)
 
class cfile_input_stream {
    std::FILE *fp;
    bool should_close;
 
  public:
    explicit cfile_input_stream(std::FILE *stream, bool close_on_destruction)
        : fp(stream), should_close(close_on_destruction && fp != nullptr) {}
 
    cfile_input_stream(cfile_input_stream const &) = delete;
    auto operator=(cfile_input_stream const &) = delete;
 
    cfile_input_stream(cfile_input_stream &&other) noexcept
        : fp(std::exchange(other.fp, nullptr)),
          should_close(std::exchange(other.should_close, false)) {}
 
    auto operator=(cfile_input_stream &&rhs) noexcept -> cfile_input_stream & {
        if (should_close)
            (void)std::fclose(fp); // NOLINT(cppcoreguidelines-owning-memory)
        fp = std::exchange(rhs.fp, nullptr);
        should_close = std::exchange(rhs.should_close, false);
        return *this;
    }
 
    ~cfile_input_stream() {
        if (should_close)
            (void)std::fclose(fp); // NOLINT(cppcoreguidelines-owning-memory)
    }
 
    auto is_error() noexcept -> bool {
        return fp == nullptr || std::ferror(fp) != 0;
    }
 
    auto is_eof() noexcept -> bool {
        return fp != nullptr && std::feof(fp) != 0;
    }
 
    auto is_good() noexcept -> bool {
        return fp != nullptr && std::ferror(fp) == 0 && std::feof(fp) == 0;
    }
 
    auto tell() noexcept -> std::optional<std::uint64_t> {
        if (fp == nullptr)
            return std::nullopt;
        std::int64_t pos =
#ifdef _WIN32
            ::_ftelli64(fp);
#else
            std::ftell(fp);
#endif
        if (pos >= 0)
            return std::uint64_t(pos);
        return std::nullopt;
    }
 
    auto skip(std::uint64_t bytes) noexcept -> bool {
        if (fp == nullptr)
            return false;
#ifdef _WIN32
        if (bytes <= std::uint64_t(std::numeric_limits<__int64>::max()))
            return ::_fseeki64(fp, __int64(bytes), SEEK_CUR) == 0;
#else
        if (bytes <= std::numeric_limits<long>::max())
            return std::fseek(fp, long(bytes), SEEK_CUR) == 0;
#endif
        return false;
    }
 
    auto read(std::span<std::byte> buffer) noexcept -> std::uint64_t {
        if (fp == nullptr)
            return 0;
        return std::fread(buffer.data(), 1, buffer.size(), fp);
    }
};
 
template <typename InputStream>
inline void skip_stream_bytes(InputStream &stream, std::uint64_t bytes) {
    if (not stream.skip(bytes)) {
        // Try instead reading and discarding up to 'start', to support
        // non-seekable streams (e.g., pipes).
        std::uint64_t bytes_discarded = 0;
        // For now, use the read size that was found fastest when reading
        // /dev/zero on an Apple M1 Pro laptop. Could be tuned.
        static constexpr std::streamsize bufsize = 32768;
        std::vector<std::byte> buf(bufsize);
        std::span<std::byte> const bufspan(buf);
        while (bytes_discarded < bytes) {
            auto read_size =
                std::min<std::uint64_t>(bufsize, bytes - bytes_discarded);
            bytes_discarded += stream.read(bufspan.first(read_size));
            if (not stream.is_good())
                break;
        }
    }
}
 
// For benchmarking only
inline auto
unbuffered_binary_ifstream_input_stream(std::string const &filename,
                                        arg::start_offset<u64> start_offset) {
    std::ifstream stream;
 
    // The standard says that the following makes the stream "unbuffered", but
    // its definition of unbuffered specifies nothing about input streams. At
    // least with libc++, this is a huge pessimization:
    stream.rdbuf()->pubsetbuf(nullptr, 0);
 
    stream.open(filename, std::ios::binary);
    if (stream.fail())
        throw input_output_error("failed to open input file: " + filename);
    auto ret = internal::istream_input_stream(std::move(stream));
    skip_stream_bytes(ret, start_offset.value);
    return ret;
}
 
// For benchmarking only
inline auto binary_ifstream_input_stream(std::string const &filename,
                                         arg::start_offset<u64> start_offset) {
    std::ifstream stream;
    stream.open(filename, std::ios::binary);
    if (stream.fail())
        throw input_output_error("failed to open input file: " + filename);
    auto ret = internal::istream_input_stream(std::move(stream));
    skip_stream_bytes(ret, start_offset.value);
    return ret;
}
 
inline auto unbuffered_binary_cfile_input_stream(
    std::string const &filename,
    arg::start_offset<std::uint64_t> start_offset) {
#ifdef _WIN32 // Avoid requiring _CRT_SECURE_NO_WARNINGS.
    std::FILE *fp{};
    (void)fopen_s(&fp, filename.c_str(), "rb");
#else
    errno = 0; // ISO C does not require fopen to set errno on error.
    // NOLINTNEXTLINE(cppcoreguidelines-owning-memory)
    std::FILE *fp = std::fopen(filename.c_str(), "rb");
#endif
    if (fp == nullptr) {
        if (errno != 0)
            throw std::system_error(errno, std::generic_category());
        throw input_output_error("failed to open input file: " + filename);
    }
    if (std::setvbuf(fp, nullptr, _IONBF, 0) != 0)
        throw input_output_error(
            "failed to disable buffering for input file: " + filename);
    auto ret = internal::cfile_input_stream(fp, true);
    skip_stream_bytes(ret, start_offset.value);
    return ret;
}
 
// For benchmarking only
inline auto binary_cfile_input_stream(std::string const &filename,
                                      arg::start_offset<u64> start_offset) {
#ifdef _WIN32 // Avoid requiring _CRT_SECURE_NO_WARNINGS.
    std::FILE *fp{};
    (void)fopen_s(&fp, filename.c_str(), "rb");
#else
    errno = 0; // ISO C does not require fopen to set errno on error.
    // NOLINTNEXTLINE(cppcoreguidelines-owning-memory)
    std::FILE *fp = std::fopen(filename.c_str(), "rb");
#endif
    if (fp == nullptr) {
        if (errno != 0)
            throw std::system_error(errno, std::generic_category());
        throw input_output_error("failed to open input file: " + filename);
    }
    auto ret = internal::cfile_input_stream(fp, true);
    skip_stream_bytes(ret, start_offset.value);
    return ret;
}
 
} // namespace internal

inline auto null_input_stream() { return internal::null_input_stream(); }

inline auto binary_file_input_stream(
    std::string const &filename,
    arg::start_offset<u64> start_offset = arg::start_offset{u64(0)}) {
    // Prefer cfile over ifstream for performance; for cfile, unbuffered
    // performs better (given our own buffering). See benchmark.
    return internal::unbuffered_binary_cfile_input_stream(filename,
                                                          start_offset);
}

template <typename IStream> inline auto istream_input_stream(IStream stream) {
    static_assert(std::is_base_of_v<std::istream, IStream>);
    return internal::istream_input_stream(std::move(stream));
}

inline auto owning_cfile_input_stream(std::FILE *fp) {
    return internal::cfile_input_stream(fp, true);
}

inline auto borrowed_cfile_input_stream(std::FILE *fp) {
    return internal::cfile_input_stream(fp, false);
}
 
namespace internal {
 
template <typename InputStream, typename Event, typename Downstream>
class read_binary_stream {
    static_assert(
        std::is_trivial_v<Event>,
        "Event type must be trivial to work with read_binary_stream");
    static_assert(processor<Downstream, bucket<Event>, warning_event>);
 
    InputStream stream;
    std::uint64_t length;
 
    std::size_t read_granularity;
    std::shared_ptr<bucket_source<Event>> bsource;
 
    Downstream downstream;
 
    LIBTCSPC_NOINLINE auto first_read_size() -> std::uint64_t {
        auto ret = read_granularity;
        if (stream.is_good()) {
            // Align second and subsequent reads to read_granularity if current
            // offset is available. This may or may not improve read
            // performance (when the read_granularity is a multiple of the page
            // size or block size), but shouldn't hurt.
            std::optional<std::uint64_t> pos = stream.tell();
            if (pos.has_value())
                ret -= *pos % read_granularity;
        }
        return ret;
    }
 
    // Read some multiple (max: max_units) of the read granularity that fits in
    // dest, subject to first-read size and max length.
    auto read_units(std::span<std::byte> dest, std::size_t max_units,
                    std::uint64_t &total_bytes_read) -> std::uint64_t {
        auto bytes_to_read =
            std::min<std::uint64_t>(dest.size(), length - total_bytes_read);
        if (total_bytes_read == 0) {
            bytes_to_read =
                std::min<std::uint64_t>(bytes_to_read, first_read_size());
        }
        if (bytes_to_read > read_granularity) {
            bytes_to_read = read_granularity *
                            std::min<std::uint64_t>(
                                max_units, bytes_to_read / read_granularity);
        }
        auto const bytes_read = stream.read(dest.first(bytes_to_read));
        total_bytes_read += bytes_read;
        return bytes_read;
    }
 
  public:
    explicit read_binary_stream(
        InputStream stream, arg::max_length<std::uint64_t> max_length,
        std::shared_ptr<bucket_source<Event>> buffer_provider,
        arg::granularity<std::size_t> granularity, Downstream downstream)
        : stream(std::move(stream)), length(max_length.value),
          read_granularity(granularity.value),
          bsource(std::move(buffer_provider)),
          downstream(std::move(downstream)) {
        if (not bsource)
            throw std::invalid_argument(
                "read_binary_stream buffer_provider must not be null");
        if (read_granularity <= 0)
            throw std::invalid_argument(
                "read_binary_stream granularity must be positive");
    }
 
    [[nodiscard]] auto introspect_node() const -> processor_info {
        return processor_info(this, "read_binary_stream");
    }
 
    [[nodiscard]] auto introspect_graph() const -> processor_graph {
        return downstream.introspect_graph().push_entry_point(this);
    }
 
    void flush() {
        auto const bucket_size =
            sizeof(Event) >= read_granularity
                ? 1
                : (read_granularity - 1) / sizeof(Event) + 1;
        auto const bucket_size_bytes = bucket_size * sizeof(Event);
 
        std::uint64_t total_bytes_read = 0;
        bucket<Event> bkt;
        std::size_t remainder_nbytes = 0; // Always < sizeof(Event)
 
        while (total_bytes_read < length && stream.is_good()) {
            auto const bytes_left_in_bucket =
                bucket_size_bytes - remainder_nbytes;
            if (bytes_left_in_bucket >= read_granularity) {
                if (bkt.empty())
                    bkt = bsource->bucket_of_size(bucket_size);
                auto const bytes_read = read_units(
                    std::as_writable_bytes(std::span(bkt))
                        .subspan(remainder_nbytes),
                    std::numeric_limits<std::size_t>::max(), total_bytes_read);
                auto const available_nbytes = remainder_nbytes + bytes_read;
                auto const this_batch_size = available_nbytes / sizeof(Event);
                remainder_nbytes = available_nbytes % sizeof(Event);
                if (this_batch_size == 0)
                    continue; // Leave incomplete event in current bucket.
                bucket<Event> bkt2;
                if (remainder_nbytes > 0) {
                    bkt2 = bsource->bucket_of_size(bucket_size);
                    auto const remainder_span =
                        std::as_bytes(std::span(bkt))
                            .subspan(available_nbytes - remainder_nbytes);
                    std::copy(remainder_span.begin(), remainder_span.end(),
                              std::as_writable_bytes(std::span(bkt2)).begin());
                }
                bkt.shrink(0, this_batch_size);
                downstream.handle(std::move(bkt));
                bkt = std::move(bkt2);
            } else { // Top off single event.
                bucket<Event> bkt2;
                bkt2 = bsource->bucket_of_size(bucket_size);
                auto const bytes_read =
                    read_units(std::as_writable_bytes(std::span(bkt2)), 1,
                               total_bytes_read);
                if (bytes_read < bytes_left_in_bucket)
                    break;
                auto const top_off_span =
                    std::as_bytes(std::span(bkt2)).first(bytes_left_in_bucket);
                auto const remainder_span = std::as_bytes(std::span(bkt2))
                                                .first(bytes_read)
                                                .subspan(bytes_left_in_bucket);
                std::copy(top_off_span.begin(), top_off_span.end(),
                          std::as_writable_bytes(std::span(bkt))
                              .last(bytes_left_in_bucket)
                              .begin());
                std::copy(remainder_span.begin(), remainder_span.end(),
                          std::as_writable_bytes(std::span(bkt2)).begin());
                downstream.handle(std::move(bkt));
                bkt = std::move(bkt2);
                remainder_nbytes = remainder_span.size();
            }
        }
 
        if (stream.is_error())
            throw input_output_error("failed to read input");
        if (remainder_nbytes > 0) {
            downstream.handle(warning_event{
                "bytes fewer than record size remain at end of input"});
        }
        downstream.flush();
    }
};
 
} // namespace internal

template <typename Event, typename InputStream, typename Downstream>
auto read_binary_stream(InputStream stream,
                        arg::max_length<std::uint64_t> max_length,
                        std::shared_ptr<bucket_source<Event>> buffer_provider,
                        arg::granularity<std::size_t> granularity,
                        Downstream downstream) {
    // Support direct passing of C++ iostreams stream.
    if constexpr (std::is_base_of_v<std::istream, InputStream>) {
        auto wrapped = istream_input_stream(std::move(stream));
        return internal::read_binary_stream<decltype(wrapped), Event,
                                            Downstream>(
            std::move(wrapped), max_length, std::move(buffer_provider),
            granularity, std::move(downstream));
    } else {
        return internal::read_binary_stream<InputStream, Event, Downstream>(
            std::move(stream), max_length, std::move(buffer_provider),
            granularity, std::move(downstream));
    }
}
 
} // namespace tcspc