store_provider.hpp

Go to the documentation of this file.

#pragma once
 
#include <ios>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <fc/io/cfile.hpp>
#include <boost/filesystem.hpp>
#include <fc/variant.hpp>
#include <sysio/trace_api/common.hpp>
#include <sysio/trace_api/metadata_log.hpp>
#include <sysio/trace_api/data_log.hpp>
#include <sysio/trace_api/compressed_file.hpp>
 
namespace sysio::trace_api {
   using namespace boost::filesystem;
 
   class path_does_not_exist : public std::runtime_error {
   public:
      explicit path_does_not_exist(const char* what_arg)
         :std::runtime_error(what_arg)
      {}
      explicit path_does_not_exist(const std::string& what_arg)
         :std::runtime_error(what_arg)
      {}
   };
 
   class old_slice_version : public std::runtime_error {
   public:
      explicit old_slice_version(const char* what_arg)
         :std::runtime_error(what_arg)
      {}
      explicit old_slice_version(const std::string& what_arg)
         :std::runtime_error(what_arg)
      {}
   };
 
   class incompatible_slice_files : public std::runtime_error {
   public:
      explicit incompatible_slice_files(const char* what_arg)
         :std::runtime_error(what_arg)
      {}
      explicit incompatible_slice_files(const std::string& what_arg)
         :std::runtime_error(what_arg)
      {}
   };
 
   class malformed_slice_file : public std::runtime_error {
   public:
      explicit malformed_slice_file(const char* what_arg)
         :std::runtime_error(what_arg)
      {}
      explicit malformed_slice_file(const std::string& what_arg)
         :std::runtime_error(what_arg)
      {}
   };
 
   template<typename DataEntry, typename File>
   static uint64_t append_store(const DataEntry &entry, File &file) {
      auto data = fc::raw::pack(entry);
      const auto offset = file.tellp();
      file.write(data.data(), data.size());
      file.flush();
      file.sync();
      return offset;
   }
 
   template<typename DataEntry, typename File>
   static DataEntry extract_store( File& file ) {
      DataEntry entry;
      auto ds = file.create_datastream();
      fc::raw::unpack(ds, entry);
      return entry;
   }
 
 
   class store_provider;
 
   class slice_directory {
   public:
      struct index_header {
         uint32_t version;
      };
 
      enum class open_state { read /*read from front to back*/, write /*write to end of file*/ };
      slice_directory(const boost::filesystem::path& slice_dir, uint32_t width, std::optional<uint32_t> minimum_irreversible_history_blocks,
                      std::optional<uint32_t> minimum_uncompressed_irreversible_history_blocks, size_t compression_seek_point_stride);
 
      uint32_t slice_number(uint32_t block_height) const {
         return block_height / _width;
      }
 
      bool find_or_create_index_slice(uint32_t slice_number, open_state state, fc::cfile& index_file) const;
 
      bool find_index_slice(uint32_t slice_number, open_state state, fc::cfile& index_file, bool open_file = true) const;
 
      bool find_or_create_trace_slice(uint32_t slice_number, open_state state, fc::cfile& trace_file) const;
 
      bool find_trace_slice(uint32_t slice_number, open_state state, fc::cfile& trace_file, bool open_file = true) const;
 
      std::optional<compressed_file> find_compressed_trace_slice(uint32_t slice_number, bool open_file = true) const;
 
      void find_or_create_slice_pair(uint32_t slice_number, open_state state, fc::cfile& trace, fc::cfile& index);
 
      bool find_or_create_trx_id_slice(uint32_t slice_number, open_state state, fc::cfile& trx_id_file) const;
 
      bool find_trx_id_slice(uint32_t slice_number, open_state state, fc::cfile& trx_id_file, bool open_file = true) const;
 
      void set_lib(uint32_t lib);
 
      void start_maintenance_thread( log_handler log );
 
      void stop_maintenance_thread();
 
      void run_maintenance_tasks(uint32_t lib, const log_handler& log);
 
   private:
      // returns true if slice is found, slice_file will always be set to the appropriate path for
      // the slice_prefix and slice_number, but will only be opened if found
      bool find_slice(const char* slice_prefix, uint32_t slice_number, fc::cfile& slice_file, bool open_file) const;
 
      // take an index file that is initialized to a file and open it and write its header
      void create_new_index_slice_file(fc::cfile& index_file) const;
 
      // take an open index slice file and verify its header is valid and prepare the file to be appended to (or read from)
      void validate_existing_index_slice_file(fc::cfile& index_file, open_state state) const;
 
      // helper for methods that process irreversible slice files
      template<typename F>
      void process_irreversible_slice_range(uint32_t lib, uint32_t upper_bound_block, std::optional<uint32_t>& lower_bound_slice, F&& f);
 
      const boost::filesystem::path _slice_dir;
      const uint32_t _width;
      const std::optional<uint32_t> _minimum_irreversible_history_blocks;
      std::optional<uint32_t> _last_cleaned_up_slice;
      const std::optional<uint32_t> _minimum_uncompressed_irreversible_history_blocks;
      std::optional<uint32_t> _last_compressed_slice;
      const size_t _compression_seek_point_stride;
 
      std::mutex _maintenance_mtx;
      std::condition_variable _maintenance_condition;
      std::thread _maintenance_thread;
      bool _maintenance_shutdown{false};
      uint32_t _best_known_lib{0};
   };
 
   class store_provider {
   public:
      using open_state = slice_directory::open_state;
 
      store_provider(const boost::filesystem::path& slice_dir, uint32_t stride_width, std::optional<uint32_t> minimum_irreversible_history_blocks,
            std::optional<uint32_t> minimum_uncompressed_irreversible_history_blocks, size_t compression_seek_point_stride);
 
      template<typename BlockTrace>
      void append(const BlockTrace& bt);
      void append_lib(uint32_t lib);
      void append_trx_ids(block_trxs_entry tt);
 
      get_block_t get_block(uint32_t block_height, const yield_function& yield= {});
 
      get_block_n get_trx_block_number(const chain::transaction_id_type& trx_id, std::optional<uint32_t> minimum_irreversible_history_blocks, const yield_function& yield= {});
 
      void start_maintenance_thread( log_handler log ) {
         _slice_directory.start_maintenance_thread( std::move(log) );
      }
      void stop_maintenance_thread() {
         _slice_directory.stop_maintenance_thread();
      }
 
      protected:
      template<typename Fn>
      uint64_t scan_metadata_log_from( uint32_t block_height, uint64_t offset, Fn&& fn, const yield_function& yield ) {
         // ignoring offset
         offset = 0;
         fc::cfile index;
         const uint32_t slice_number = _slice_directory.slice_number(block_height);
         const bool found = _slice_directory.find_index_slice(slice_number, open_state::read, index);
         if( !found ) {
            return 0;
         }
         const uint64_t end = file_size(index.get_file_path());
         offset = index.tellp();
         uint64_t last_read_offset = offset;
         while (offset < end) {
            yield();
            const auto metadata = extract_store<metadata_log_entry>(index);
            if(! fn(metadata)) {
               break;
            }
            last_read_offset = offset;
            offset = index.tellp();
         }
         return last_read_offset;
      }
 
      std::optional<data_log_entry> read_data_log( uint32_t block_height, uint64_t offset ) {
         const uint32_t slice_number = _slice_directory.slice_number(block_height);
 
         fc::cfile trace;
         if( !_slice_directory.find_trace_slice(slice_number, open_state::read, trace) ) {
            // attempt to read a compressed trace if one exists
            std::optional<compressed_file> ctrace = _slice_directory.find_compressed_trace_slice(slice_number);
            if (ctrace) {
               ctrace->seek(offset);
               return extract_store<data_log_entry>(*ctrace);
            }
 
            const std::string offset_str = boost::lexical_cast<std::string>(offset);
            const std::string bh_str = boost::lexical_cast<std::string>(block_height);
            throw malformed_slice_file("Requested offset: " + offset_str + " to retrieve block number: " + bh_str + " but this trace file is new, so there are no traces present.");
         }
         const uint64_t end = file_size(trace.get_file_path());
         if( offset >= end ) {
            const std::string offset_str = boost::lexical_cast<std::string>(offset);
            const std::string bh_str = boost::lexical_cast<std::string>(block_height);
            const std::string end_str = boost::lexical_cast<std::string>(end);
            throw malformed_slice_file("Requested offset: " + offset_str + " to retrieve block number: " + bh_str + " but this trace file only goes to offset: " + end_str);
         }
         trace.seek(offset);
         return extract_store<data_log_entry>(trace);
      }
 
      void initialize_new_index_slice_file(fc::cfile& index);
 
      void validate_existing_index_slice_file(fc::cfile& index, open_state state);
 
      slice_directory _slice_directory;
   };
 
}
 
FC_REFLECT(sysio::trace_api::slice_directory::index_header, (version))