Storage rework WIP

src/ifcparse/FileReader.cpp

@@ -0,0 +1,312 @@
+#include "FileReader.h"
+
+#include <algorithm>
+#include <cstdio>
+#include <filesystem>
+#include <fstream>
+#include <list>
+#include <stdexcept>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+#include <deque>
+
+// Boost iostreams mmap
+#include <boost/iostreams/device/mapped_file.hpp>
+
+
+
+namespace {
+
+#if defined(_WIN32)
+    inline void file_seek_abs(FILE* f, std::uint64_t off) {
+        if (_fseeki64(f, static_cast<long long>(off), SEEK_SET) != 0)
+            throw std::runtime_error("fseek failed");
+    }
+#else
+    inline void file_seek_abs(FILE* f, std::uint64_t off) {
+        if (fseeko(f, static_cast<off_t>(off), SEEK_SET) != 0)
+            throw std::runtime_error("fseeko failed");
+    }
+#endif
+
+} // namespace
+
+// ===================== Concrete backends =====================
+
+using namespace IfcParse;
+
+struct FullBufferImpl final : FileReader::Impl {
+    std::vector<char> buf_;
+    explicit FullBufferImpl(const std::string& fn) {
+        std::ifstream ifs(fn, std::ios::binary);
+        if (!ifs) throw std::runtime_error("Failed to open: " + fn);
+        ifs.seekg(0, std::ios::end);
+        const std::streamsize sz = ifs.tellg();
+        ifs.seekg(0, std::ios::beg);
+        buf_.resize(static_cast<size_t>(sz));
+        if (sz > 0 && !ifs.read(buf_.data(), sz)) {
+            throw std::runtime_error("Failed to read file into buffer");
+        }
+    }
+    size_t size() const override { return buf_.size(); }
+    char get(size_t pos) const override {
+        if (pos >= buf_.size()) throw std::out_of_range("get out of range");
+        return buf_[pos];
+    }
+};
+
+struct PagedFileImpl final : FileReader::Impl {
+    std::string fn_;
+    FILE* fp_ = nullptr;
+    size_t file_size_ = 0;
+    size_t page_size_ = 4096;
+
+    // LRU cache
+    size_t capacity_ = 8;
+    mutable std::list<size_t> lru_;  // most recent at front
+    struct Entry {
+        FileReader::Page page;
+        std::list<size_t>::iterator it;
+    };
+    mutable std::unordered_map<size_t, Entry> map_;
+
+    PagedFileImpl(const std::string& fn, size_t page_size, size_t cap)
+        : fn_(fn), page_size_(std::max<size_t>(512, page_size)), capacity_(std::max<size_t>(2, cap)) {
+        namespace fs = std::filesystem;
+        if (!fs::exists(fn_)) throw std::runtime_error("File not found: " + fn_);
+        file_size_ = static_cast<size_t>(fs::file_size(fn_));
+        fp_ = std::fopen(fn_.c_str(), "rb");
+        if (!fp_) throw std::runtime_error("Failed to fopen: " + fn_);
+    }
+
+    ~PagedFileImpl() override {
+        if (fp_) std::fclose(fp_);
+        fp_ = nullptr;
+    }
+
+    size_t size() const override { return file_size_; }
+
+    char get(size_t pos) const override {
+        if (pos >= file_size_) throw std::out_of_range("get out of range");
+        const size_t pidx = pos / page_size_;
+        const FileReader::Page& p = fetch_page_(pidx);
+        const size_t off = pos % page_size_;
+        if (off >= p.data.size()) throw std::out_of_range("offset beyond valid page bytes");
+        // Opportunistic read-ahead for sequential scans
+        // if (off + 1 == p.data.size()) (void)try_prefetch_(pidx + 1);
+        return p.data[off];
+    }
+
+private:
+    const FileReader::Page& fetch_page_(size_t idx) const {
+        auto it = map_.find(idx);
+        if (it != map_.end()) {
+            touch_(it);
+            return it->second.page;
+        }
+        // Load page from disk using persistent FILE*
+        FileReader::Page pg;
+        pg.data.resize(page_size_);
+        const size_t begin = idx * page_size_;
+        const size_t avail = std::min(page_size_, file_size_ - begin);
+
+        file_seek_abs(fp_, begin);
+        if (avail > 0) {
+            const size_t nread = std::fread(pg.data.data(), 1, avail, fp_);
+            if (nread != avail) throw std::runtime_error("Short fread on page");
+        }
+		pg.data.resize(avail); // trim to actual size
+
+        // Insert into LRU
+        if (map_.size() >= capacity_) evict_();
+        lru_.push_front(idx);
+        auto lit = lru_.begin();
+        auto [emplaced_it, ok] = map_.emplace(idx, Entry{ std::move(pg), lit });
+        (void)ok;
+        return emplaced_it->second.page;
+    }
+
+    /*
+    bool try_prefetch_(size_t idx) const {
+        if (idx * page_size_ >= file_size_) return false;
+        if (map_.find(idx) != map_.end()) return true;
+        if (map_.size() + 1 > capacity_) return false;
+        (void)fetch_page_(idx);
+        return true;
+    }
+    */
+
+    void touch_(typename std::unordered_map<size_t, Entry>::iterator it) const {
+        lru_.erase(it->second.it);
+        lru_.push_front(it->first);
+        it->second.it = lru_.begin();
+    }
+
+    void evict_() const {
+        if (lru_.empty()) return;
+        const size_t victim = lru_.back();
+        lru_.pop_back();
+        map_.erase(victim);
+    }
+};
+
+struct MMapImpl final : FileReader::Impl {
+    boost::iostreams::mapped_file_source map_;
+    size_t size_ = 0;
+
+    explicit MMapImpl(const std::string& fn) {
+        namespace fs = std::filesystem;
+        if (!fs::exists(fn)) throw std::runtime_error("File not found: " + fn);
+        size_ = static_cast<size_t>(fs::file_size(fn));
+        if (size_ == 0) return; // empty file: map_ stays closed
+        map_.open(fn);
+        if (!map_.is_open()) throw std::runtime_error("Failed to open mapped_file_source");
+        size_ = static_cast<size_t>(map_.size());
+    }
+
+    size_t size() const override { return size_; }
+
+    char get(size_t pos) const override {
+        if (pos >= size_) throw std::out_of_range("get out of range");
+        return map_.data()[pos];
+    }
+};
+
+/// User-pushed sequential backend with an arbitrary-length queue of future pages.
+/// We keep a deque of pages; when reads move forward, we drop fully-consumed
+/// pages from the front to release memory.
+struct PushedSequentialImpl final : std::enable_shared_from_this<PushedSequentialImpl>, FileReader::Impl {
+	// Deque of pages, front is earliest in file.
+    std::deque<FileReader::Page> pages_;
+    // total bytes in dropped pages
+	size_t discarded_page_bytes_ = 0;
+
+    size_t size() const override {
+        size_t n = discarded_page_bytes_;
+		for (auto& pg : pages_) n += pg.data.size();
+		return n;
+    }
+
+    // Drop fully-consumed pages so pos is guaranteed to be within the first page
+    void drop_consumed_up_to(size_t pos) {
+        while (!pages_.empty()) {
+            if (pos - discarded_page_bytes_ >= pages_.front().data.size()) {
+				discarded_page_bytes_ += pages_.front().data.size();
+                pages_.pop_front();
+            } else {
+                break;
+            }
+        }
+    }
+
+    char get(size_t pos) const override {
+        auto self = const_cast<PushedSequentialImpl*>(this);
+        if (this->shared_from_this().use_count() == 2) {
+            // only drop pages when there is only one active client.
+            // NB this->shared_from_this() increases count by 1
+            self->drop_consumed_up_to(pos);
+        }
+
+        const size_t avail_end = size();
+        if (pos >= avail_end) throw std::out_of_range("pushed backend: position not committed yet");
+
+		pos -= discarded_page_bytes_;
+
+		size_t page_start = 0;
+        for (const auto& pg : pages_) {
+            if (pos < page_start + pg.data.size()) {
+                const size_t off = pos - page_start;
+                return pg.data[off];
+            } else {
+				page_start += pg.data.size();
+            }
+        }
+
+        throw std::out_of_range("pushed backend: internal inconsistency");
+    }
+
+    void push_next_page(const std::string& data) override {
+        FileReader::Page p; p.data.assign(data.data(), data.data() + data.size());
+        pages_.push_back(std::move(p));
+    }
+};
+
+// ===================== FileReader public API =====================
+
+IfcParse::FileReader::FileReader(const std::string& fn)
+    : cursor_(0)
+{
+	impl_ = std::make_shared<FullBufferImpl>(fn);
+}
+
+IfcParse::FileReader::FileReader(const std::string& fn, const mmap_tag&)
+    : cursor_(0)
+{
+	impl_ = std::make_shared<MMapImpl>(fn);
+}
+
+IfcParse::FileReader::FileReader(const caller_fed_tag&)
+    : cursor_(0)
+{
+	impl_ = std::make_shared<PushedSequentialImpl>();
+}
+
+IfcParse::FileReader::FileReader(const std::string& content, const caller_fed_tag&)
+{
+    impl_ = std::make_shared<PushedSequentialImpl>();
+	impl_->push_next_page(content);
+}
+
+IfcParse::FileReader::FileReader(const std::string& fn, size_t page_size, size_t page_capacity)
+    : cursor_(0)
+{
+	impl_ = std::make_shared<PagedFileImpl>(fn, page_size, page_capacity);
+}
+
+FileReader FileReader::clone() const {
+    FileReader c(*this);
+    c.cursor_ = this->cursor_;
+    return c;
+}
+
+void FileReader::seek(size_t pos) {
+    if (pos > impl_->size()) throw std::out_of_range("seek out of range");
+    cursor_ = pos;
+}
+
+size_t FileReader::tell() const { return cursor_; }
+
+size_t FileReader::size() const { return impl_->size(); }
+
+char FileReader::peek() const {
+    if (cursor_ >= impl_->size()) throw std::out_of_range("peek at EOF");
+    return impl_->get(cursor_);
+}
+
+void FileReader::increment(size_t n) {
+    if (cursor_ + n > impl_->size()) throw std::out_of_range("increment past EOF");
+    cursor_ += n;
+}
+
+void IfcParse::FileReader::push_next_page(const std::string& data)
+{
+    impl_->push_next_page(data);
+}
+
+bool IfcParse::FileReader::eof() const
+{
+    return cursor_ >= impl_->size();
+}
+
+char IfcParse::FileReader::read()
+{
+	auto c = peek();
+	increment(1);
+	return c;
+}
+
+char IfcParse::FileReader::get(size_t offset) const
+{
+	return impl_->get(offset);
+}