BufferedFileWriter.hpp

Go to the documentation of this file.

#ifndef DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_UTILS_BUFFEREDFILEWRITER_HPP_
#define DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_UTILS_BUFFEREDFILEWRITER_HPP_
 
#include "datahandlinglibs/DataHandlingIssues.hpp"
#include "datahandlinglibs/ReadoutLogging.hpp"
 
#include "logging/Logging.hpp"
 
#include <boost/align/aligned_allocator.hpp>
#include <boost/iostreams/device/file_descriptor.hpp>
#include <boost/iostreams/filter/lzma.hpp>
#include <boost/iostreams/filter/zlib.hpp>
#include <boost/iostreams/filter/zstd.hpp>
#include <boost/iostreams/filtering_stream.hpp>
#include <boost/iostreams/stream.hpp>
#include <boost/iostreams/stream_buffer.hpp>
 
#include <fcntl.h>
#include <fstream>
#include <iostream>
#include <limits>
#include <string>
#include <unistd.h>
 
using dunedaq::datahandlinglibs::logging::TLVL_WORK_STEPS;
 
namespace dunedaq {
namespace datahandlinglibs {
template<size_t Alignment = 4096>
class BufferedFileWriter
{
  using io_sink_t = boost::iostreams::file_descriptor_sink;
  using aligned_allocator_t = boost::alignment::aligned_allocator<io_sink_t::char_type, Alignment>;
  using filtering_ostream_t =
    boost::iostreams::filtering_stream<boost::iostreams::output, char, std::char_traits<char>, aligned_allocator_t>;
 
public:
  BufferedFileWriter(std::string filename,
                     size_t buffer_size,
                     std::string compression_algorithm = "None",
                     bool use_o_direct = true)
  {
    open(filename, buffer_size, compression_algorithm, use_o_direct);
  }

  BufferedFileWriter() {}

  ~BufferedFileWriter()
  {
    if (m_is_open)
      close();
  }
 
  BufferedFileWriter(const BufferedFileWriter&) = delete;            
  BufferedFileWriter& operator=(const BufferedFileWriter&) = delete; 
  BufferedFileWriter(BufferedFileWriter&&) = delete;                 
  BufferedFileWriter& operator=(BufferedFileWriter&&) = delete;      

  void open(std::string filename,
            size_t buffer_size,
            std::string compression_algorithm = "None",
            bool use_o_direct = true)
  {
    m_use_o_direct = use_o_direct;
    if (m_is_open) {
      close();
    }
 
    m_filename = filename;
    m_buffer_size = buffer_size;
    m_compression_algorithm = compression_algorithm;
    auto oflag = O_CREAT | O_WRONLY;
    if (m_use_o_direct) {
      oflag = oflag | O_DIRECT;
    }
 
    m_fd = ::open(m_filename.c_str(), oflag, 0644);
    if (m_fd == -1) {
      throw BufferedReaderWriterCannotOpenFile(ERS_HERE, m_filename);
    }
 
    m_sink = io_sink_t(m_fd, boost::iostreams::file_descriptor_flags::close_handle);
    if (m_compression_algorithm == "zstd") {
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Using zstd compression" << std::endl;
      m_output_stream.push(boost::iostreams::zstd_compressor(boost::iostreams::zstd::best_speed));
    } else if (m_compression_algorithm == "lzma") {
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Using lzma compression" << std::endl;
      m_output_stream.push(boost::iostreams::lzma_compressor(boost::iostreams::lzma::best_speed));
    } else if (m_compression_algorithm == "zlib") {
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Using zlib compression" << std::endl;
      m_output_stream.push(boost::iostreams::zlib_compressor(boost::iostreams::zlib::best_speed));
    } else if (m_compression_algorithm == "None") {
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Running without compression" << std::endl;
    } else {
      throw BufferedReaderWriterConfigurationError(ERS_HERE,
                                                   "Non-recognized compression algorithm: " + m_compression_algorithm);
    }
 
    m_output_stream.push(m_sink, m_buffer_size);
    m_is_open = true;
  }

  bool is_open() const { return m_is_open; }

  bool write(const char* memory, const size_t size)
  {
    if (!m_is_open)
      return false;
    m_output_stream.write(memory, size); // NOLINT
    return !m_output_stream.bad();
  }

  void close()
  {
    // Set the file descriptor to not use O_DIRECT. This is necessary because the write size has to be aligned for
    // O_DIRECT to succeed. This is not guaranteed for the data that remains in the buffer.
    fcntl(m_fd, F_SETFL, O_CREAT | O_WRONLY);
    m_output_stream.reset();
    m_is_open = false;
  }

  void flush()
  {
    // Set the file descriptor to not use O_DIRECT. This is necessary because the write size has to be aligned for
    // O_DIRECT to succeed. This is not guaranteed for the data that remains in the buffer.
    fcntl(m_fd, F_SETFL, O_CREAT | O_WRONLY);
    // This does not flush the compressor as it is not flushable
    m_output_stream.flush();
    // Activate O_DIRECT again
    auto oflag = O_CREAT | O_WRONLY;
    if (m_use_o_direct) {
      oflag = oflag | O_DIRECT;
    }
    fcntl(m_fd, F_SETFL, oflag);
  }
 
private:
  // Config parameters
  std::string m_filename;
  size_t m_buffer_size;
  std::string m_compression_algorithm;
 
  // Internals
  int m_fd;
  io_sink_t m_sink;
  filtering_ostream_t m_output_stream;
  bool m_is_open = false;
  bool m_use_o_direct = true;
};
 
} // namespace datahandlinglibs
} // namespace dunedaq
 
#endif // DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_UTILS_BUFFEREDFILEWRITER_HPP_