ringfilecache.H

Go to the documentation of this file.

/*
                          Aleph_w
 
  Data structures & Algorithms
  version 2.0.0b
  https://github.com/lrleon/Aleph-w
 
  This file is part of Aleph-w library
 
  Copyright (c) 2002-2026 Leandro Rabindranath Leon
 
  Permission is hereby granted, free of charge, to any person obtaining a copy
  of this software and associated documentation files (the "Software"), to deal
  in the Software without restriction, including without limitation the rights
  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  copies of the Software, and to permit persons to whom the Software is
  furnished to do so, subject to the following conditions:
 
  The above copyright notice and this permission notice shall be included in all
  copies or substantial portions of the Software.
 
  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  SOFTWARE.
*/
 
 
# ifndef RINGFILECACHE_H
# define RINGFILECACHE_H
 
# include <sys/time.h>
# include <stdio.h>
# include <cassert>
# include <string.h>
# include <stdexcept>
# include <memory>
# include <fstream>
# include <iostream>
# include <string>
# include <sstream>
# include <type_traits>
 
# include <ah-errors.H>
# include <tpl_array.H>
 
inline constexpr size_t Ring_Max_Name_Size = 4096;
 
template <typename T>
class RingFileCache
{
public:
  static_assert(std::is_trivially_copyable<T>::value,
                "RingFileCache requires trivially copyable types");
  static_assert(std::is_default_constructible<T>::value,
                "T must be default constructible");
 
  // The cache parameters are stored in a binary file with this structure
  struct Pars
  {
    size_t dim; // capacity in entries of cache
    size_t n; // Number of entries stored in cache
    size_t head; // next integral position ready for reading
    size_t tail; // next integral position ready for writing
    size_t size_cache_file = 0; // num of chars of std::string (including '\0')
    char cache_file_name[0]; // be careful with this field. DON'T USE IT!
 
    [[nodiscard]] std::string to_string() const
    {
      std::ostringstream s;
      s << dim << " " << n << " " << tail << " " << head << " "
          << size_cache_file;
      return s.str();
    }
 
    friend std::ostream &operator <<(std::ostream & out, const Pars & pars)
    {
      return out << pars.to_string();
    }
  };
 
private:
  bool initialized = false;
  std::string pars_file_name;
  std::string cache_file_name;
 
  std::fstream pars_stream; // the stream for the pars (read/write mode)
  std::fstream cache_stream; // the stream for the cache (read/write mode)
 
  size_t dim = 0; // the maximum number of entries
  size_t n = 0; // the current number of stored entries
  size_t head = 0; // next integral position ready for reading
  size_t tail = 0; // next integral position ready for writing
 
  void test_and_set_tail_pointer()
  {
    cache_stream.seekp(tail * sizeof(T));
  }
 
  bool is_valid_offset(const size_t offset) const noexcept
  {
    return offset < n;
  }
 
  bool is_valid_position(const size_t pos) const noexcept
  {
    if (head <= tail)
      return pos >= head and pos < tail;
    // Wraparound case: valid if pos >= head OR pos < tail
    return pos >= head or pos < tail;
  }
 
  void test_and_set_head_pointer(const size_t num_entries = 0)
  {
    const size_t pos = (head + num_entries) % dim;
    cache_stream.seekg(pos * sizeof(T));
  }
 
  T read_entry()
  {
    T ret;
    cache_stream.read(reinterpret_cast<char *>(&ret), sizeof ret);
 
    return ret;
  }
 
  T read_entry(const size_t pos)
  {
    ah_range_error_if(pos >= n)
      << "read_entry(" << pos << "): cache has " << n << " items";
 
    const size_t real_pos = (head + pos) % dim;
 
    T ret;
    cache_stream.seekg(real_pos * sizeof(ret));
    cache_stream.read(reinterpret_cast<char *>(&ret), sizeof ret);
    return ret;
  }
 
  void write_entry(const T & item)
  {
    cache_stream.write(reinterpret_cast<const char *>(&item), sizeof item);
  }
 
  void validate_absolute_position(const size_t pos) const
  {
    if (pos < dim)
      return;
 
    ah_out_of_range_error() << "position " << pos << " is greater than dim " << dim;
  }
 
  T read_absolute(const size_t pos)
  {
    validate_absolute_position(pos);
    T ret;
    cache_stream.seekg(pos * sizeof(T));
    cache_stream.read(reinterpret_cast<char *>(&ret), sizeof ret);
    return ret;
  }
 
  void write_absolute(const size_t pos, const T & item)
  {
    validate_absolute_position(pos);
    cache_stream.seekp(pos * sizeof(T));
    cache_stream.write(reinterpret_cast<const char *>(&item), sizeof item);
  }
 
public:
  std::string to_string() const
  {
    const size_t tg = const_cast<std::fstream &>(cache_stream).tellg() / sizeof(T);
    const size_t tp = const_cast<std::fstream &>(cache_stream).tellp() / sizeof(T);
    std::ostringstream s;
    s << "Cache pars" << std::endl
        << "capacity = " << capacity() << std::endl
        << "size     = " << size() << std::endl
        << "sizeof T = " << sizeof(T) << std::endl
        << "head pos = " << head_pos() << std::endl
        << "tail pos = " << tail_pos() << std::endl
        << "tellg/T  = " << tg << std::endl
        << "tellp/T  = " << tp << std::endl;
    return s.str();
  }
 
  friend std::ostream &operator <<(std::ostream & out, const RingFileCache & cache)
  {
    return out << cache.to_string();
  }
 
  static void create(const std::string & pars_file_name,
                     const std::string & cache_file_name,
                     const size_t num_entries)
  {
    std::ostringstream s;
 
    ah_domain_error_if(cache_file_name.size() + 1 > Ring_Max_Name_Size)
      << "cache_file_name too long (max " << (Ring_Max_Name_Size - 1) << " chars)";
 
    std::fstream pars_stream(pars_file_name, std::ios::binary | std::ios::out);
    ah_domain_error_if(not pars_stream.is_open() or not pars_stream.good())
      << "cannot open " << pars_file_name;
 
    std::fstream cache_stream(cache_file_name, std::ios::binary | std::ios::out);
    ah_domain_error_if(not cache_stream.is_open() or not cache_stream.good())
      << "cannot open " << cache_file_name;
 
    const size_t pars_size = sizeof(Pars) + cache_file_name.size() + 1;
    std::shared_ptr<Pars> pars_ptr(static_cast<Pars *>(malloc(pars_size)), free);
 
    ah_bad_alloc_if(pars_ptr == nullptr);
 
    pars_ptr->dim = num_entries;
    pars_ptr->n = pars_ptr->head = pars_ptr->tail = 0;
    pars_ptr->size_cache_file = cache_file_name.size() + 1;
    strncpy(pars_ptr->cache_file_name, cache_file_name.c_str(),
            pars_ptr->size_cache_file);
    pars_ptr->cache_file_name[pars_ptr->size_cache_file - 1] = '\0';
    pars_stream.write(reinterpret_cast<const char *>(pars_ptr.get()), pars_size);
 
    T init{};
    for (size_t i = 0; i < num_entries; ++i)
      cache_stream.write(reinterpret_cast<const char *>(&init), sizeof(init));
  }
 
private:
  static std::string state(std::fstream & ss)
  {
    std::ostringstream s;
    const auto state = ss.rdstate();
    if (state & std::ios::goodbit)
      s << "    goodbit = true " << std::endl;
    else
      s << "    goodbit = false " << std::endl;
    if (state & std::ios::badbit)
      s << "    badbit  = true " << std::endl;
    else
      s << "    badbit  = false " << std::endl;
    if (state & std::ios::failbit)
      s << "    failbit = true " << std::endl;
    else
      s << "    failbit = false " << std::endl;
    if (state & std::ios::eofbit)
      s << "    eofbit  = true " << std::endl;
    else
      s << "    eofbit  = false " << std::endl;
    return s.str();
  }
 
  // Helper method for initializing the cache from the pars file
  void read_pars(const std::string & pars_file_name)
  {
    pars_stream.open(pars_file_name, std::ios::binary | std::ios::out | std::ios::in);
 
    ah_domain_error_if((not pars_stream.is_open()) or (not pars_stream.good()))
      << "cannot open " << pars_file_name;
 
    // set the cache internal state from the saved state in pars file
    Pars pars;
    pars_stream.read(reinterpret_cast<char *>(&pars), sizeof pars);
    dim = pars.dim;
    n = pars.n;
    head = pars.head;
    tail = pars.tail;
 
    // read the cache file name, which should be saved at the end of
    // pars file
    const size_t name_size = std::min(Ring_Max_Name_Size, pars.size_cache_file);
    const std::shared_ptr<char> name(static_cast<char *>(malloc(name_size)), free);
    pars_stream.read(name.get(), name_size);
    cache_file_name = name.get();
 
    cache_stream.open(cache_file_name, std::ios::binary | std::ios::out | std::ios::in);
    ah_domain_error_if(not cache_stream.is_open() or not cache_stream.good())
      << "cannot open " << cache_file_name;
 
    pars_stream.seekp(0);
 
    cache_stream.seekg(head * sizeof(T)); // set read position
    cache_stream.seekp(tail * sizeof(T)); // set write position
    initialized = true;
 
    assert(pars_stream.good());
    assert(cache_stream.good());
  }
 
public:
  class Pointer
  {
    friend class RingFileCache;
 
    RingFileCache *cache_ptr = nullptr;
    size_t pos = 0; // this an absolute position in the file
 
    void validate_position(const size_t pos) const
    {
      if (pos < cache_ptr->dim)
        return;
 
      std::ostringstream s;
      s << "position " << pos << " is outside of maximum " << cache_ptr->dim;
      ah_out_of_range_error_if(pos >= cache_ptr->dim) << s.str();
    }
 
    void increase_pos(const long delta = 1) noexcept
    {
      if (delta < 0)
        {
          decrease_pos(-delta);
          return;
        }
 
      const size_t dim = cache_ptr->dim;
      pos = (pos + static_cast<size_t>(delta)) % dim;
    }
 
    void decrease_pos(const long delta = 1) noexcept
    {
      if (delta < 0)
        {
          increase_pos(-delta);
          return;
        }
 
      const size_t dim = cache_ptr->dim;
      if (const size_t decr = static_cast<size_t>(delta) % dim; pos >= decr)
        pos -= decr;
      else
        pos = dim - (decr - pos);
    }
 
  public:
    Pointer() = default;
 
    Pointer(const RingFileCache & cache, const size_t __pos = 0)
      : cache_ptr(&const_cast<RingFileCache &>(cache)),
        pos((cache.head_pos() + __pos) % cache.dim) {}
 
    Pointer operator ++() noexcept
    {
      increase_pos(1);
      return *this;
    }
 
    Pointer operator ++(int) noexcept
    {
      Pointer ret = *this;
      increase_pos(1);
      return ret;
    }
 
    Pointer operator --() noexcept
    {
      decrease_pos(1);
      return *this;
    }
 
    Pointer operator --(int) noexcept
    {
      Pointer ret = *this;
      decrease_pos(1);
      return ret;
    }
 
    Pointer &operator +=(const long val) noexcept
    {
      increase_pos(val);
      return *this;
    }
 
    Pointer &operator -=(const long val) noexcept
    {
      decrease_pos(val);
      return *this;
    }
 
    Pointer operator +(const long val) const noexcept
    {
      Pointer ret = *this;
      ret += val;
      return ret;
    }
 
    Pointer operator -(const long val) const noexcept
    {
      Pointer ret = *this;
      ret -= val;
      return ret;
    }
 
    [[nodiscard]] size_t get_pos_respect_to_head() const noexcept
    {
      auto head = cache_ptr->head;
      if (pos >= head)
        return pos - head;
      return cache_ptr->dim - head + pos;
    }
 
    [[nodiscard]] size_t get_pos() const noexcept
    {
      return get_pos_respect_to_head();
    }
  };
 
  T read(const Pointer & ptr)
  {
    ah_domain_error_if(ptr.cache_ptr != this)
        << "RingFileCache::read(const Pointer&): invalid ptr";
 
    return read_absolute(ptr.pos);
  }
 
  void write(const Pointer & ptr, const T & item)
  {
    ah_domain_error_if(ptr.cache_ptr != this)
        << "RingFileCache::write(const Pointer&): invalid ptr";
 
    return write_absolute(ptr.pos, item);
  }
 
  bool is_initialized() const { return initialized; }
 
  size_t size() const noexcept { return n; }
 
  size_t capacity() const noexcept { return dim; }
 
  size_t avail() const noexcept { return dim - n; }
 
  size_t head_pos() const noexcept { return head; }
 
  size_t tail_pos() const noexcept { return tail; }
 
  bool is_empty() const noexcept { return n == 0; }
 
  bool is_full() const noexcept { return n == dim; }
 
  RingFileCache(const std::string & pars_fname) : pars_file_name(pars_fname)
  {
    read_pars(pars_file_name);
  }
 
  RingFileCache() = default;
 
  static bool test(const std::string & pars_fname)
  {
    std::ifstream pars_stream_test(pars_fname, std::ios::binary);
    if (not pars_stream_test.good())
      return false;
 
    Pars pars;
    pars_stream_test.read(reinterpret_cast<char *>(&pars), sizeof pars);
 
    // read the cache file name, which should be saved at the end of
    // pars file
    const size_t name_size = std::min(Ring_Max_Name_Size, pars.size_cache_file);
    std::shared_ptr<char> name(static_cast<char *>(malloc(name_size)), free);
    pars_stream_test.read(name.get(), name_size);
    const char *cache_file_name = name.get();
 
    if (std::ifstream cache_stream(cache_file_name, std::ios::binary); not cache_stream.good())
      return false;
 
    return true;
  }
 
  void init(const std::string & pars_fname)
  {
    ah_domain_error_if(pars_stream.is_open())
      << "this cache has already an opened pars file";
 
    pars_file_name = pars_fname;
    read_pars(pars_file_name);
  }
 
  bool put(const T & item)
  {
    if (n == dim) // cache full?
      return false;
 
    //    assert(cache_stream.good());
 
    test_and_set_tail_pointer();
 
    assert(cache_stream.tellp() == tail*sizeof(T));
 
    write_entry(item);
    if (++tail == dim) // eof border reached?
      {
        tail = 0;
        cache_stream.seekp(0);
        assert(cache_stream.good());
      }
    ++n;
 
    assert(cache_stream.good());
    assert(cache_stream.tellp() == tail*sizeof(T));
 
    return true;
  }
 
  bool read(T entries[], const size_t m = 1)
  {
    if (m > n)
      return false;
 
    test_and_set_head_pointer();
 
    assert(cache_stream.good());
    // assert(cache_stream.tellg() == head*sizeof(T));
 
    const size_t n_avail_until_eof = dim - head;
    if (m <= n_avail_until_eof)
      cache_stream.read(reinterpret_cast<char *>(entries), m * sizeof(T));
    else
      {
        cache_stream.read(reinterpret_cast<char *>(entries),
                          n_avail_until_eof * sizeof(T));
        cache_stream.seekg(0);
        cache_stream.read(reinterpret_cast<char *>(&entries[n_avail_until_eof]),
                          (m - n_avail_until_eof) * sizeof(T));
        assert(cache_stream.gcount() == (m - n_avail_until_eof)*sizeof(T));
      }
 
    assert(cache_stream.good());
 
    return true;
  }
 
  T read_first()
  {
    ah_underflow_error_if(n == 0) << "read_first(): cache is empty";
 
    test_and_set_head_pointer();
    T ret_val;
    read(&ret_val);
    return ret_val;
  }
 
  T read_last()
  {
    ah_underflow_error_if(n == 0) << "read_last(): cache is empty";
 
    if (tail == 0)
      cache_stream.seekg((dim - 1) * sizeof(T));
    else
      cache_stream.seekg((tail - 1) * sizeof(T));
 
    return read_entry();
  }
 
  T youngest() { return read_last(); }
 
  T oldest() { return read_first(); }
 
  T oldest(size_t i)
  {
    ah_overflow_error_if(i >= n)
      << "oldest(" << i << ") but the cache has " << n << " entries";
 
    return read_entry(i);
  }
 
  Array<T> read_all()
  {
    if (n == 0)
      return Array<T>();
 
    Array<T> ret(n);
    ret.putn(n);
    assert(ret.capacity() >= n);
    read(&ret.base(), n);
    return ret;
  }
 
  Array<T> read_from(const size_t pos, const size_t m)
  {
    Array<T> ret;
    Iterator it(*this, pos);
    for (size_t i = 0; i < m and it.has_curr(); ++i, it.next_ne())
      ret.append(it.get_curr());
 
    return ret;
  }
 
  Array<T> read_from(const Pointer & ptr, const size_t m)
  {
    Array<T> ret;
    Iterator it(*this, ptr.get_pos_respect_to_head());
    for (size_t i = 0; i < m and it.has_curr(); ++i, it.next_ne())
      ret.append(it.get_curr_ne());
 
    return ret;
  }
 
  bool get(const size_t m = 1) noexcept
  {
    if (m > n)
      return false;
 
    head = (head + m) % dim;
    n -= m;
 
    return true;
  }
 
  void empty() noexcept { get(n); }
 
  void flush()
  {
    assert(pars_stream.tellp() == 0);
    Pars pars;
    pars.dim = dim;
    pars.n = n;
    pars.head = head;
    pars.tail = tail;
    pars.size_cache_file = cache_file_name.size() + 1;
    pars_stream.write(reinterpret_cast<const char *>(&pars), sizeof(pars));
    pars_stream.flush();
    pars_stream.seekp(0);
 
    cache_stream.flush();
  }
 
  void close()
  {
    if (not initialized)
      return;
    flush();
    cache_stream.close();
    pars_stream.close();
    initialized = false;
  }
 
  ~RingFileCache() { close(); }
 
  void resize(const size_t sz)
  {
    ah_domain_error_if(sz < dim)
      << "RingFileCache::resize(): file truncation is not implemented (yet?)";
 
    if (sz == dim)
      return; // no-op
 
    const size_t old_dim = dim;
 
    // Handle wraparound case: data spans [head, old_dim) and [0, tail)
    // Move the wrapped portion [0, tail) to [old_dim, old_dim + tail)
    if (n > 0 && tail <= head && tail > 0)
      {
        const size_t wrapped_count = tail;
        auto wrapped_data = std::make_unique<T[]>(wrapped_count);
 
        cache_stream.seekg(0);
        cache_stream.read(reinterpret_cast<char *>(wrapped_data.get()),
                          wrapped_count * sizeof(T));
 
        cache_stream.seekp(old_dim * sizeof(T));
        cache_stream.write(reinterpret_cast<const char *>(wrapped_data.get()),
                           wrapped_count * sizeof(T));
 
        tail = old_dim + wrapped_count;
      }
 
    // Zero-initialize remaining slots from current tail (or old_dim) to sz
    T init{};
    const size_t fill_start = (tail > old_dim) ? tail : old_dim;
    for (size_t pos = fill_start; pos < sz; ++pos)
      {
        cache_stream.seekp(pos * sizeof(T));
        write_entry(init);
      }
 
    dim = sz;
    // tail is already correct after wraparound fix, or needs update for non-wrapped case
    if (n > 0 && tail <= old_dim)
      tail = (head + n) % dim;
 
    test_and_set_tail_pointer();
    flush();
  }
 
  // Iterator for read-only traversal (oldest to youngest).
  // Not reentrant: do not execute other operations on the cache object
  // while iterating.
  class Iterator
  {
    RingFileCache<T> *cache_ptr = nullptr;
    T curr{};
    size_t pos = 0;
    size_t curr_pos = 0;
 
    void set_curr_pointer()
    {
      cache_ptr->cache_stream.seekg(curr_pos * sizeof(T));
    }
 
    void increase_pos()
    {
      ++pos;
      if (++curr_pos == cache_ptr->dim)
        curr_pos = 0;
    }
 
    void load_curr()
    {
      set_curr_pointer();
      curr = cache_ptr->read_entry();
    }
 
  public:
    [[nodiscard]] size_t get_pos() const noexcept { return pos; }
 
    [[nodiscard]] bool has_curr() const { return pos < cache_ptr->n; }
 
    Iterator(const RingFileCache<T> & cache, const size_t offset = 0)
      : cache_ptr(&const_cast<RingFileCache<T> &>(cache)),
        pos(offset), curr_pos((cache_ptr->head + offset) % cache.dim)
    {
      if (cache_ptr->is_empty())
        return;
 
      if (not cache_ptr->is_valid_offset(pos))
        return;
 
      load_curr();
    }
 
    T get_curr_ne() const noexcept { return curr; }
 
    T get_curr() const
    {
      ah_overflow_error_if(not has_curr())
        << "RingFileCache::Iterator::get_curr(): iterator exhausted";
      return get_curr_ne();
    }
 
    void next_ne() noexcept
    {
      increase_pos();
      if (has_curr())
        load_curr();
    }
 
    void next()
    {
      ah_overflow_error_if(not has_curr())
        << "RingFileCache::Iterator::next(): iterator exhausted";
      next_ne();
    }
  };
 
  Iterator get_it() { return Iterator(*this); }
};
 
# endif // RINGFILECACHE_H