FileDataStoreImpl.hpp

Go to the documentation of this file.

 
#ifndef DFMODULES_INCLUDE_DFMODULES_FILEDATASTOREIMPL_HPP_
#define DFMODULES_INCLUDE_DFMODULES_FILEDATASTOREIMPL_HPP_
 
#include "dfmodules/DataStore.hpp"
#include "dfmodules/opmon/DataStore.pb.h"
 
#include "appmodel/DataStoreConf.hpp"
#include "appmodel/FilenameParams.hpp"
#include "confmodel/DetectorConfig.hpp"
#include "confmodel/Session.hpp"
 
#include "appfwk/DAQModule.hpp"
#include "daqdataformats/Types.hpp"
#include "logging/Logging.hpp" // NOTE: if ISSUES ARE DECLARED BEFORE include logging/Logging.hpp, TLOG_DEBUG<<issue wont work.
 
#include "boost/date_time/posix_time/posix_time.hpp"
#include "boost/lexical_cast.hpp"
 
#include <concepts>
#include <cstdlib>
#include <functional>
#include <limits>
#include <memory>
#include <string>
#include <sys/statvfs.h>
#include <utility>
#include <vector>
 
namespace dunedaq {
 
// Disable coverage checking LCOV_EXCL_START
ERS_DECLARE_ISSUE_BASE(dfmodules,
                       FileDataStoreImplBadConfiguration,
                       appfwk::GeneralDAQModuleIssue,
                       "Construction of the FileDataStoreImpl base class failed due to faulty configuration",
                       ((std::string)name), )
 
ERS_DECLARE_ISSUE_BASE(dfmodules,
                       InvalidOperationMode,
                       appfwk::GeneralDAQModuleIssue,
                       "Selected operation mode \"" << selected_operation
                                                    << "\" is NOT supported. Please update the configuration file.",
                       ((std::string)name),
                       ((std::string)selected_operation))
 
ERS_DECLARE_ISSUE_BASE(dfmodules,
                       FileOperationProblem,
                       appfwk::GeneralDAQModuleIssue,
                       "A problem was encountered when opening or closing file \"" << filename << "\"",
                       ((std::string)name),
                       ((std::string)filename))
 
ERS_DECLARE_ISSUE_BASE(dfmodules,
                       InvalidOutputPath,
                       appfwk::GeneralDAQModuleIssue,
                       "The specified output destination, \"" << output_path
                                                              << "\", is not a valid file system path on this server.",
                       ((std::string)name),
                       ((std::string)output_path))
 
ERS_DECLARE_ISSUE_BASE(dfmodules,
                       InsufficientDiskSpace,
                       appfwk::GeneralDAQModuleIssue,
                       "There is insufficient free space on the disk associated with output file path \""
                         << path << "\". There are " << free_bytes << " bytes free, and the "
                         << "required minimum is " << needed_bytes << " bytes based on " << criteria << ".",
                       ((std::string)name),
                       ((std::string)path)((size_t)free_bytes)((size_t)needed_bytes)((std::string)criteria))
 
ERS_DECLARE_ISSUE_BASE(dfmodules,
                       TimeSliceAlreadyExists,
                       appfwk::GeneralDAQModuleIssue,
                       "The TimeSlice record for timeslice #" << timeslice_number << " already exists.",
                       ((std::string)name),
                       ((daqdataformats::timeslice_number_t)timeslice_number))
 
// Re-enable coverage checking LCOV_EXCL_STOP
namespace dfmodules {
 
// A class which satisfies the FileHandleConcept should implement format-specific writes of objects
template<typename T>
concept FileHandleConcept = requires(T file_handle,
                                     const T const_file_handle,
                                     const daqdataformats::TriggerRecord& tr,
                                     const daqdataformats::TimeSlice& ts) {
  { file_handle.write(tr) } -> std::same_as<void>;
  { file_handle.write(ts) } -> std::same_as<void>;
  { file_handle.timeslice_already_exists(ts) } -> std::convertible_to<bool>;
  { const_file_handle.get_file_name() } -> std::convertible_to<std::string>;
  { const_file_handle.get_file_name_extension() } -> std::convertible_to<std::string>; // Don't include the "."
  { const_file_handle.get_recorded_size() } -> std::convertible_to<size_t>;
  { const_file_handle.get_uncompressed_raw_data_size() } -> std::convertible_to<size_t>;
  { const_file_handle.get_total_file_size() } -> std::convertible_to<size_t>;
};

template<FileHandleConcept FileHandleClass> // E.g., SummaryTextDataWriter
class FileDataStoreImpl : public DataStore
{
 
public:
  enum
  {
    TLVL_BASIC = 2
  };
 
  static constexpr size_t s_unset_record_number{ std::numeric_limits<size_t>::max() };

  explicit FileDataStoreImpl(std::string const& name,
                             std::shared_ptr<appfwk::ConfigurationManager> mcfg,
                             std::string const& writer_name)
    : DataStore(name)
    , m_file_handle{ nullptr }
    , m_run_number{ 0 }
    , m_file_index{ 0 }
    , m_writer_identifier{ writer_name }
    , m_config_params{ mcfg ? mcfg->get_dal<appmodel::DataStoreConf>(name) : nullptr }
    , m_session{ mcfg ? mcfg->get_session() : nullptr }
    , m_compression_level{ m_config_params ? m_config_params->get_compression_level() : static_cast<unsigned>(0) }
    // NOLINT(build/unsigned)
    , m_operational_environment{ m_session ? m_session->get_detector_configuration()->get_op_env() : "unavailable" }
    , m_offline_data_stream{ m_session ? m_session->get_detector_configuration()->get_offline_data_stream()
                                       : "unavailable" }
    , m_run_is_for_test_purposes{ false }
    , m_basic_name_of_open_file{ "" }
    , m_recorded_size{ 0 }
    , m_uncompressed_raw_data_size{ 0 }
    , m_previous_file_size{ 0 }
    , m_total_file_size{ 0 }
    , m_current_record_number{ s_unset_record_number }
    , m_new_bytes{ 0 }
    , m_new_objects{ 0 }
    , m_operation_mode{ m_config_params ? m_config_params->get_mode() : "unavailable" }
    , m_path{ m_config_params ? m_config_params->get_directory_path() : "unavailable" }
    , m_max_file_size{ m_config_params ? m_config_params->get_max_file_size() : std::numeric_limits<size_t>::max() }
    , m_disable_unique_suffix{ m_config_params ? m_config_params->get_disable_unique_filename_suffix() : false }
    , m_free_space_safety_factor_for_write{ m_config_params ? m_config_params->get_free_space_safety_factor()
                                                            : std::numeric_limits<float>::max() }
  {
    TLOG_DEBUG(TLVL_BASIC) << get_name();
 
    if (!m_config_params || !m_session) {
      throw FileDataStoreImplBadConfiguration(ERS_HERE, get_name());
    }
 
    if (m_operation_mode != "one-event-per-file" && m_operation_mode != "all-per-file") {
 
      throw InvalidOperationMode(ERS_HERE, get_name(), m_operation_mode);
    }
 
    // 05-Apr-2022, KAB: added warning message when the output destination
    // is not a valid directory.
    struct statvfs vfs_results;
    int retval = statvfs(m_path.c_str(), &vfs_results);
    if (retval != 0) {
      ers::warning(InvalidOutputPath(ERS_HERE, get_name(), m_path));
    }
  }
 
  virtual void open_new_file(const std::string& unique_filename) = 0;
 
  // Getter functions which can be used by the implementation of open_new_file
 
  std::string get_application_name() const noexcept { return m_writer_identifier; }
 
  unsigned get_compression_level() const noexcept
  { // NOLINT(build/unsigned)
    return m_compression_level;
  }
 
  const appmodel::DataStoreConf& get_configuration() const noexcept { return *m_config_params; }
 
  auto& get_file_handle() { return m_file_handle; }
 
  size_t get_file_index() const noexcept { return m_file_index.load(); }
 
  const std::string& get_offline_data_stream() const noexcept { return m_offline_data_stream; }
 
  const std::string& get_operational_environment() const noexcept { return m_operational_environment; }
 
  bool get_run_is_for_test_purposes() const noexcept { return m_run_is_for_test_purposes; }
 
  daqdataformats::run_number_t get_run_number() const noexcept { return m_run_number; }
 
  const confmodel::Session& get_session() const noexcept { return *m_session; }

  void write(const daqdataformats::TriggerRecord& tr) override
  {
    size_t tr_size = tr.get_total_size_bytes();
 
    throw_if_insufficient_space_for_object(tr_size, "trigger record");
 
    increment_file_index_if_needed(tr_size, tr.get_header_ref().get_trigger_number(), m_current_record_number);
 
    m_current_record_number = tr.get_header_ref().get_trigger_number();
 
    std::string full_filename = get_file_name(tr.get_header_ref().get_run_number());
 
    try {
      open_file_if_needed(full_filename);
    } catch (std::exception const& excpt) {
      throw FileOperationProblem(ERS_HERE, get_name(), full_filename, excpt);
    } catch (...) { // NOLINT(runtime/exceptions)
      // NOLINT here because we *ARE* re-throwing the exception!
      throw FileOperationProblem(ERS_HERE, get_name(), full_filename);
    }
 
    // write the record
    m_file_handle->write(tr);
    m_recorded_size = m_file_handle->get_recorded_size();
    m_uncompressed_raw_data_size = m_file_handle->get_uncompressed_raw_data_size();
    m_total_file_size = m_file_handle->get_total_file_size();
 
    m_new_bytes += m_total_file_size - m_previous_file_size;
    ++m_new_objects;
    m_previous_file_size.store(m_total_file_size.load());
  }

  void write(const daqdataformats::TimeSlice& ts) override
  {
    size_t ts_size = ts.get_total_size_bytes();
    throw_if_insufficient_space_for_object(ts_size, "time slice");
 
    increment_file_index_if_needed(ts_size, ts.get_header().timeslice_number, m_current_record_number);
 
    m_current_record_number = ts.get_header().timeslice_number;
 
    std::string full_filename = get_file_name(ts.get_header().run_number);
 
    try {
      open_file_if_needed(full_filename);
    } catch (std::exception const& excpt) {
      throw FileOperationProblem(ERS_HERE, get_name(), full_filename, excpt);
    } catch (...) { // NOLINT(runtime/exceptions)
      // NOLINT here because we *ARE* re-throwing the exception!
      throw FileOperationProblem(ERS_HERE, get_name(), full_filename);
    }
 
    // write the record
    try {
 
      if (m_file_handle->timeslice_already_exists(ts)) {
        throw TimeSliceAlreadyExists(ERS_HERE, get_name(), ts.get_header().timeslice_number);
      }
 
      m_file_handle->write(ts);
      m_recorded_size = m_file_handle->get_recorded_size();
      m_uncompressed_raw_data_size = m_file_handle->get_uncompressed_raw_data_size();
      m_total_file_size = m_file_handle->get_total_file_size();
 
    } catch (TimeSliceAlreadyExists const& excpt) {
      std::string msg = "writing a time slice to file " + m_file_handle->get_file_name();
      throw IgnorableDataStoreProblem(ERS_HERE, get_name(), msg, excpt);
    }
 
    m_new_bytes += m_total_file_size - m_previous_file_size;
    ++m_new_objects;
    m_previous_file_size.store(m_total_file_size.load());
  }

  void prepare_for_run(daqdataformats::run_number_t run_number, bool run_is_for_test_purposes) override
  {
    m_run_number = run_number;
    m_run_is_for_test_purposes = run_is_for_test_purposes;
 
    struct statvfs vfs_results;
    TLOG_DEBUG(TLVL_BASIC) << get_name() << ": Preparing to get the statvfs results for path: \"" << m_path << "\"";
 
    int retval = statvfs(m_path.c_str(), &vfs_results);
    TLOG_DEBUG(TLVL_BASIC) << get_name() << ": statvfs return code is " << retval;
    if (retval != 0) {
      throw InvalidOutputPath(ERS_HERE, get_name(), m_path);
    }
 
    size_t free_space = vfs_results.f_bsize * vfs_results.f_bavail;
    TLOG_DEBUG(TLVL_BASIC) << get_name() << ": Free space on disk with path \"" << m_path << "\" is " << free_space
                           << " bytes. This will be compared with the maximum size of a single file ("
                           << m_max_file_size << ") as a simple test to see if there is enough free space.";
    if (free_space < m_max_file_size) {
      throw InsufficientDiskSpace(
        ERS_HERE, get_name(), m_path, free_space, m_max_file_size, "the configured maximum size of a single file");
    }
 
    m_file_index = 0;
    m_recorded_size = 0;
    m_uncompressed_raw_data_size = 0;
    m_current_record_number = s_unset_record_number;
  }

  void finish_with_run(daqdataformats::run_number_t /*run_number*/) override
  {
    if (m_file_handle) {
      std::string open_filename = m_file_handle->get_file_name();
      try {
        m_file_handle.reset();
        m_run_number = 0;
      } catch (std::exception const& excpt) {
        m_run_number = 0;
        throw FileOperationProblem(ERS_HERE, get_name(), open_filename, excpt);
      } catch (...) { // NOLINT(runtime/exceptions)
        m_run_number = 0;
        // NOLINT here because we *ARE* re-throwing the exception!
        throw FileOperationProblem(ERS_HERE, get_name(), open_filename);
      }
    }
  }
 
  FileDataStoreImpl(const FileDataStoreImpl&) = delete;
  FileDataStoreImpl& operator=(const FileDataStoreImpl&) = delete;
  FileDataStoreImpl(FileDataStoreImpl&&) = delete;
  FileDataStoreImpl& operator=(FileDataStoreImpl&&) = delete;
 
protected:
  void generate_opmon_data() override
  {
    opmon::FileDataStoreImplInfo info;
 
    info.set_new_bytes_output(m_new_bytes.exchange(0));
    info.set_new_written_object(m_new_objects.exchange(0));
    info.set_bytes_in_file(m_total_file_size.load());
    info.set_written_files(m_file_index.load());
    publish(std::move(info), { { "path", m_path } });
  }
 
private:
  // Translates various available parameters (directory path, file extension, etc.) into the appropriate filename.
  std::string get_file_name(daqdataformats::run_number_t run_number) const;
 
  // Throws a RetryableDataStoreProblem if (size of the object)*(free space safety factor) exceeds available space in
  // m_path
  void throw_if_insufficient_space_for_object(size_t obj_size, const std::string& obj_name) const;
 
  // Available space in the path whose name is passed to get_free_space
  size_t get_free_space(const std::string& the_path) const;
 
  // Check if a new file should be opened for the record and increment m_file_index if so
  void increment_file_index_if_needed(const size_t size_of_object_to_write,
                                      const size_t object_record_number,
                                      const size_t current_record_number)
  {
    float compression_factor{ 1.0 };
 
    if (m_compression_level != 0 && m_recorded_size != 0) {
      // Without compression, the uncompressed raw data size is approximately the total file size, so it
      // serves as an approximation of what would have been written without compression
      compression_factor =
        static_cast<float>(m_file_handle->get_uncompressed_raw_data_size()) / m_file_handle->get_total_file_size();
    }
 
    float size_of_next_write = size_of_object_to_write / compression_factor;
 
    if ((m_total_file_size + size_of_next_write) > m_max_file_size && m_recorded_size > 0) {
      ++m_file_index;
      m_recorded_size = 0;
      m_uncompressed_raw_data_size = 0;
      m_previous_file_size.store(0);
      return;
    }
 
    // JCF, 06-27-2026: TODO: probably need to reset m_recorded_size, etc., as is done right above
    if (m_operation_mode == "one-event-per-file" && current_record_number != s_unset_record_number &&
        current_record_number != object_record_number) {
      ++m_file_index;
      return;
    }
  }
 
  void open_file_if_needed(const std::string& file_name)
  {
 
    if (!m_file_handle || m_basic_name_of_open_file.compare(file_name)) {
 
      // close an existing open file
      if (m_file_handle) {
        std::string open_filename = m_file_handle->get_file_name();
        try {
          m_file_handle.reset();
        } catch (std::exception const& excpt) {
          throw FileOperationProblem(ERS_HERE, get_name(), open_filename, excpt);
        } catch (...) { // NOLINT(runtime/exceptions)
          // NOLINT here because we *ARE* re-throwing the exception!
          throw FileOperationProblem(ERS_HERE, get_name(), open_filename);
        }
      }
 
      // 04-Feb-2021, KAB: adding unique substrings to the filename
      // 05-Feb-2026, KAB: moved this block of code *after* the block that closes an
      // existing open file. When the two blocks were executed in the opposite order,
      // it was possible that the closing of the currently-open file could take a
      // non-trivial amount of time, and then the timestamp in the filename (determined
      // here) and the timestamp in the creation_timestamp HDF5 file Attribute
      // (determined inside the HDF5RawDataFile constructor) could disagree.
      std::string unique_filename = file_name;
      if (!m_disable_unique_suffix) {
        time_t now{ time(nullptr) };
        std::string file_creation_timestamp = boost::posix_time::to_iso_string(boost::posix_time::from_time_t(now));
        // timestamp substring
        size_t ufn_len = unique_filename.length();
        size_t extension_length =
          m_file_handle->get_file_name_extension().size() + 1; // + 1 for the "." before the extension
        if (ufn_len > extension_length + 1) { // this gives us some confidence that we have at least one character
                                              // preceding the extension (e.g., x.hdf5)
          std::string timestamp_substring = "_" + file_creation_timestamp;
          TLOG_DEBUG(TLVL_BASIC) << get_name() << ": timestamp substring for filename: " << timestamp_substring;
          unique_filename.insert(ufn_len - extension_length, timestamp_substring);
        }
      }
 
      // opening file for the first time OR something changed in the name or the way of opening the file
      TLOG_DEBUG(TLVL_BASIC) << get_name() << ": going to open file " << unique_filename;
      m_basic_name_of_open_file = file_name;
 
      open_new_file(unique_filename);
 
    } else {
      TLOG_DEBUG(TLVL_BASIC) << get_name() << ": Pointer file to  " << m_basic_name_of_open_file
                             << " was already opened";
    }
  }
 
  std::unique_ptr<FileHandleClass> m_file_handle;
 
  daqdataformats::run_number_t m_run_number;
 
  // Total number of generated files
  std::atomic<size_t> m_file_index;
  const std::string m_writer_identifier;
 
  const appmodel::DataStoreConf* m_config_params;
 
  const confmodel::Session* m_session;
 
  unsigned m_compression_level; // NOLINT(build/unsigned)
 
  const std::string m_operational_environment;
  const std::string m_offline_data_stream;
  bool m_run_is_for_test_purposes;
 
  std::string m_basic_name_of_open_file;
 
  // Size of data being written, excluding metadata
  std::atomic<size_t> m_recorded_size;
 
  // Theoretical, "uncompressed" size of data being written, excluding metadata
  std::atomic<size_t> m_uncompressed_raw_data_size;
 
  // Used for tracking the "delta" of the current write
  std::atomic<size_t> m_previous_file_size = 0;
 
  // Total size of the file, including raw data, metadata, and free space
  std::atomic<size_t> m_total_file_size;
 
  // Record number for the record that is currently being written out
  // This is only useful for long-readout windows, in which there may
  // be multiple calls to write()
  size_t m_current_record_number;
 
  // incremental written data
  std::atomic<uint64_t> m_new_bytes;   // NOLINT(build/unsigned)
  std::atomic<uint64_t> m_new_objects; // NOLINT(build/unsigned)
 
  const std::string m_operation_mode;
  const std::string m_path;
  const size_t m_max_file_size;
  const bool m_disable_unique_suffix;
  float m_free_space_safety_factor_for_write;
};
 
} // namespace dfmodules
} // namespace dunedaq
 
#include "detail/FileDataStoreImpl.hxx"
 
#endif // DFMODULES_INCLUDE_DFMODULES_FILEDATASTOREIMPL_HPP_