#include <ZeroCopyRecordingRequestHandlerModel.hpp>

Inheritance diagram for dunedaq::datahandlinglibs::ZeroCopyRecordingRequestHandlerModel< ReadoutType, LatencyBufferType >:

[legend]

Collaboration diagram for dunedaq::datahandlinglibs::ZeroCopyRecordingRequestHandlerModel< ReadoutType, LatencyBufferType >:

[legend]

Public Types
using	inherited = DefaultRequestHandlerModel<ReadoutType, LatencyBufferType>

Public Types inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >
using	RDT = ReadoutType

using	LBT = LatencyBufferType

using	RequestResult

using	ResultCode

Public Types inherited from dunedaq::opmonlib::MonitorableObject
using	NodePtr = std::weak_ptr<MonitorableObject>

using	NewNodePtr = std::shared_ptr<MonitorableObject>

using	ElementId = std::string

Public Member Functions
	ZeroCopyRecordingRequestHandlerModel (std::shared_ptr< LatencyBufferType > &latency_buffer, std::unique_ptr< FrameErrorRegistry > &error_registry)

void	conf (const appmodel::DataHandlerModule *conf) override

void	record (const nlohmann::json &args) override

Public Member Functions inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >
	DefaultRequestHandlerModel (std::shared_ptr< LatencyBufferType > &latency_buffer, std::unique_ptr< FrameErrorRegistry > &error_registry)

void	scrap (const nlohmann::json &) override

void	start (const nlohmann::json &)

void	stop (const nlohmann::json &)

void	cleanup_check () override
	Check if cleanup is necessary and execute it if necessary.

virtual void	periodic_data_transmission () override
	Periodic data transmission - relevant for trigger in particular.

void	issue_request (dfmessages::DataRequest datarequest, bool is_retry=false) override
	Issue a data request to the request handler.

virtual dunedaq::daqdataformats::timestamp_t	get_cutoff_timestamp ()

virtual bool	supports_cutoff_timestamp ()

Public Member Functions inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >
	RequestHandlerConcept ()

virtual	~RequestHandlerConcept ()

	RequestHandlerConcept (const RequestHandlerConcept &)=delete
	RequestHandlerConcept is not copy-constructible.

RequestHandlerConcept &	operator= (const RequestHandlerConcept &)=delete
	RequestHandlerConcept is not copy-assginable.

	RequestHandlerConcept (RequestHandlerConcept &&)=delete
	RequestHandlerConcept is not move-constructible.

RequestHandlerConcept &	operator= (RequestHandlerConcept &&)=delete
	RequestHandlerConcept is not move-assignable.

Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
	MonitorableObject (const MonitorableObject &)=delete

MonitorableObject &	operator= (const MonitorableObject &)=delete

	MonitorableObject (MonitorableObject &&)=delete

MonitorableObject &	operator= (MonitorableObject &&)=delete

virtual	~MonitorableObject ()=default

auto	get_opmon_id () const noexcept

auto	get_opmon_level () const noexcept

Private Attributes
int	m_fd

int	m_oflag

Additional Inherited Members
Static Public Member Functions inherited from dunedaq::opmonlib::MonitorableObject
static bool	publishable_metric (OpMonLevel entry, OpMonLevel system) noexcept

Protected Types inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >
enum	ResultCode { kFound = 0 , kNotFound , kTooOld , kNotYet , kPartial , kPartiallyOld , kCleanup , kUnknown }

Protected Member Functions inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >
daqdataformats::FragmentHeader	create_fragment_header (const dfmessages::DataRequest &dr)

std::unique_ptr< daqdataformats::Fragment >	create_empty_fragment (const dfmessages::DataRequest &dr)

void	dump_to_buffer (const void data, std::size_t size, void buffer, uint32_t buffer_pos, const std::size_t &buffer_size)

void	periodic_cleanups ()

void	periodic_data_transmissions ()

void	cleanup ()

void	check_waiting_requests ()

std::vector< std::pair< void *, size_t > >	get_fragment_pieces (uint64_t start_win_ts, uint64_t end_win_ts, RequestResult &rres)

RequestResult	data_request (dfmessages::DataRequest dr) override

virtual void	generate_opmon_data () override

Protected Member Functions inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >
const std::string &	resultCodeAsString (ResultCode rc)

Protected Member Functions inherited from dunedaq::opmonlib::MonitorableObject
	MonitorableObject ()=default

void	register_node (ElementId name, NewNodePtr)

void	publish (google::protobuf::Message &&, CustomOrigin &&co={}, OpMonLevel l=to_level(EntryOpMonLevel::kDefault)) const noexcept

Protected Attributes inherited from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >
std::shared_ptr< LatencyBufferType > &	m_latency_buffer

BufferedFileWriter	m_buffered_writer

utilities::ReusableThread	m_recording_thread

utilities::ReusableThread	m_cleanup_thread

utilities::ReusableThread	m_periodic_transmission_thread

std::map< dfmessages::DataRequest, int >	m_request_counter

std::size_t	m_max_requested_elements

std::mutex	m_cv_mutex

std::condition_variable	m_cv

std::atomic< bool >	m_cleanup_requested = false

std::atomic< int >	m_requests_running = 0

std::vector< RequestElement >	m_waiting_requests

std::mutex	m_waiting_requests_lock

std::unique_ptr< boost::asio::thread_pool >	m_request_handler_thread_pool

size_t	m_num_request_handling_threads = 0

std::unique_ptr< FrameErrorRegistry > &	m_error_registry

std::chrono::time_point< std::chrono::high_resolution_clock >	m_t0

std::atomic< bool >	m_run_marker = false

std::thread	m_waiting_queue_thread

std::atomic< bool >	m_recording = false

std::atomic< uint64_t >	m_next_timestamp_to_record = std::numeric_limits<uint64_t>::max()

bool	m_configured

float	m_pop_limit_pct

float	m_pop_size_pct

unsigned	m_pop_limit_size

size_t	m_buffer_capacity

daqdataformats::SourceID	m_sourceid

uint16_t	m_detid

std::string	m_output_file

size_t	m_stream_buffer_size = 0

bool	m_recording_configured = false

bool	m_warn_on_timeout = true

bool	m_warn_about_empty_buffer = true

uint32_t	m_periodic_data_transmission_ms = 0

std::vector< std::string >	m_frag_out_conn_ids

std::atomic< int >	m_pop_counter

std::atomic< int >	m_num_buffer_cleanups { 0 }

std::atomic< int >	m_pop_reqs

std::atomic< int >	m_pops_count

std::atomic< int >	m_occupancy

std::atomic< int >	m_num_requests_found { 0 }

std::atomic< int >	m_num_requests_bad { 0 }

std::atomic< int >	m_num_requests_old_window { 0 }

std::atomic< int >	m_num_requests_delayed { 0 }

std::atomic< int >	m_num_requests_uncategorized { 0 }

std::atomic< int >	m_num_requests_timed_out { 0 }

std::atomic< int >	m_handled_requests { 0 }

std::atomic< int >	m_response_time_acc { 0 }

std::atomic< int >	m_response_time_min { std::numeric_limits<int>::max() }

std::atomic< int >	m_response_time_max { 0 }

std::atomic< int >	m_payloads_written { 0 }

std::atomic< int >	m_bytes_written { 0 }

std::atomic< uint64_t >	m_num_periodic_sent { 0 }

std::atomic< uint64_t >	m_num_periodic_send_failed { 0 }

int	m_fragment_send_timeout_ms

Protected Attributes inherited from dunedaq::datahandlinglibs::RequestHandlerConcept< ReadoutType, LatencyBufferType >
std::map< ResultCode, std::string >	ResultCodeStrings

Detailed Description

template<class ReadoutType, class LatencyBufferType>
class dunedaq::datahandlinglibs::ZeroCopyRecordingRequestHandlerModel< ReadoutType, LatencyBufferType >

Definition at line 20 of file ZeroCopyRecordingRequestHandlerModel.hpp.

Member Typedef Documentation

◆ inherited

template<class ReadoutType , class LatencyBufferType >

using dunedaq::datahandlinglibs::ZeroCopyRecordingRequestHandlerModel< ReadoutType, LatencyBufferType >::inherited = DefaultRequestHandlerModel<ReadoutType, LatencyBufferType>

Definition at line 24 of file ZeroCopyRecordingRequestHandlerModel.hpp.

Constructor & Destructor Documentation

◆ ZeroCopyRecordingRequestHandlerModel()

template<class ReadoutType , class LatencyBufferType >

dunedaq::datahandlinglibs::ZeroCopyRecordingRequestHandlerModel< ReadoutType, LatencyBufferType >::ZeroCopyRecordingRequestHandlerModel	(	std::shared_ptr< LatencyBufferType > &	latency_buffer,
		std::unique_ptr< FrameErrorRegistry > &	error_registry )

inlineexplicit

Definition at line 27 of file ZeroCopyRecordingRequestHandlerModel.hpp.

    : DefaultRequestHandlerModel<ReadoutType, LatencyBufferType>(latency_buffer, error_registry)
  {
    TLOG_DEBUG(TLVL_WORK_STEPS) << "ZeroCopyRecordingRequestHandlerModel created...";
  }

Member Function Documentation

◆ conf()

template<class ReadoutType , class LatencyBufferType >

void dunedaq::datahandlinglibs::ZeroCopyRecordingRequestHandlerModel< ReadoutType, LatencyBufferType >::conf ( const appmodel::DataHandlerModule * conf )

overridevirtual

Reimplemented from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >.

Definition at line 9 of file ZeroCopyRecordingRequestHandlerModel.hxx.

{
  auto data_rec_conf = conf->get_module_configuration()->get_request_handler()->get_data_recorder();
  
  if (data_rec_conf != nullptr) {
    if (!data_rec_conf->get_output_file().empty()) {
      inherited::m_sourceid.id = conf->get_source_id();
      inherited::m_sourceid.subsystem = ReadoutType::subsystem;
  
      // Check for alignment restrictions for filesystem block size. (XFS default: 4096) 
      if (inherited::m_latency_buffer->get_alignment_size() == 0 ||
          sizeof(ReadoutType) * inherited::m_latency_buffer->size() % 4096) {
        ers::error(ConfigurationError(ERS_HERE, inherited::m_sourceid, "Latency buffer is not 4kB aligned"));
      }
 
      // Check for sensible stream chunk size
      inherited::m_stream_buffer_size = data_rec_conf->get_streaming_buffer_size();
      if (inherited::m_stream_buffer_size % 4096 != 0) {
        ers::error(ConfigurationError(ERS_HERE, inherited::m_sourceid, "Streaming chunk size is not divisible by 4kB!"));
      }
  
      // Prepare filename with full path 
      std::string file_full_path = data_rec_conf->get_output_file() + inherited::m_sourceid.to_string() + std::string(".bin");
      inherited::m_output_file = file_full_path;
 
  
      // RS: This will need to go away with the SNB store handler!
      if (std::remove(file_full_path.c_str()) == 0) {
        TLOG(TLVL_WORK_STEPS) << "Removed existing output file from previous run: " << file_full_path;
      }
  
      m_oflag = O_CREAT | O_WRONLY;
      if (data_rec_conf->get_use_o_direct()) {
        m_oflag |= O_DIRECT;
      }
      m_fd = ::open(file_full_path.c_str(), m_oflag, 0644);
      if (m_fd == -1) {
        TLOG() << "Failed to open file!";
        throw ConfigurationError(ERS_HERE, inherited::m_sourceid, "Failed to open file!");
      }
      inherited::m_recording_configured = true;
 
    } else { // no output dir specified
      TLOG(TLVL_WORK_STEPS) << "No output path is specified in data recorder config. Recording feature is inactive.";
    } 
  } else {
    TLOG(TLVL_WORK_STEPS) << "No recording config object specified. Recording feature is inactive."; 
  }
  
  inherited::conf(conf);
}

◆ record()

template<class ReadoutType , class LatencyBufferType >

void dunedaq::datahandlinglibs::ZeroCopyRecordingRequestHandlerModel< ReadoutType, LatencyBufferType >::record ( const nlohmann::json & args )

overridevirtual

Reimplemented from dunedaq::datahandlinglibs::DefaultRequestHandlerModel< ReadoutType, LatencyBufferType >.

Definition at line 64 of file ZeroCopyRecordingRequestHandlerModel.hxx.

{
  if (inherited::m_recording.load()) {
    ers::error(
      CommandError(ERS_HERE, inherited::m_sourceid, "A recording is still running, no new recording was started!"));
    return;
  }
 
// FIXME: Recording parameters to be clarified!
  int recording_time_sec = 0;
  if (cmdargs.contains("duration")) {
    recording_time_sec = cmdargs["duration"];
  } else {
    ers::warning(
      CommandError(ERS_HERE, inherited::m_sourceid, "A recording command with missing duration field received!"));
  }
  if (recording_time_sec == 0) {
    ers::warning(
      CommandError(ERS_HERE, inherited::m_sourceid, "Recording for 0 seconds requested. Recording command is ignored!"));
    return;
  }
 
  inherited::m_recording_thread.set_work(
    [&](int duration) {
      size_t chunk_size = inherited::m_stream_buffer_size;
      size_t alignment_size = inherited::m_latency_buffer->get_alignment_size();
      TLOG() << "Start recording for " << duration << " second(s)" << std::endl;
      inherited::m_recording.exchange(true);
      auto start_of_recording = std::chrono::high_resolution_clock::now();
      auto current_time = start_of_recording;
      inherited::m_next_timestamp_to_record = 0;
 
      const char* current_write_pointer = nullptr;
      const char* start_of_buffer_pointer =
        reinterpret_cast<const char*>(inherited::m_latency_buffer->start_of_buffer()); // NOLINT
      const char* current_end_pointer;
      const char* end_of_buffer_pointer = reinterpret_cast<const char*>(inherited::m_latency_buffer->end_of_buffer()); // NOLINT
 
      size_t bytes_written = 0;
      size_t failed_writes = 0;
 
      while (std::chrono::duration_cast<std::chrono::seconds>(current_time - start_of_recording).count() < duration) {
        if (!inherited::m_cleanup_requested || (inherited::m_next_timestamp_to_record == 0)) {
          size_t considered_chunks_in_loop = 0;
 
          // Wait for potential running cleanup to finish first
          {
            std::unique_lock<std::mutex> lock(inherited::m_cv_mutex);
            inherited::m_cv.wait(lock, [&] { return !inherited::m_cleanup_requested; });
          }
          inherited::m_cv.notify_all();
 
          // Some frames have to be skipped to start copying from an aligned piece of memory
          // These frames cannot be written without O_DIRECT as this would mess up the alignment of the write pointer
          // into the target file
          if (inherited::m_next_timestamp_to_record == 0) {
            auto begin = inherited::m_latency_buffer->begin();
            if (begin == inherited::m_latency_buffer->end()) {
              // There are no elements in the buffer, update time and try again
              current_time = std::chrono::high_resolution_clock::now();
              continue;
            }
            inherited::m_next_timestamp_to_record = begin->get_timestamp();
            size_t skipped_frames = 0;
            while (reinterpret_cast<std::uintptr_t>(&(*begin)) % alignment_size) { // NOLINT
              ++begin;
              skipped_frames++;
              if (!begin.good()) {
                // We reached the end of the buffer without finding an aligned element
                // Reset the next timestamp to record and try again
                current_time = std::chrono::high_resolution_clock::now();
                inherited::m_next_timestamp_to_record = 0;
                continue;
              }
            }
            TLOG() << "Skipped " << skipped_frames << " frames";
            current_write_pointer = reinterpret_cast<const char*>(&(*begin)); // NOLINT
          }
 
          current_end_pointer = reinterpret_cast<const char*>(inherited::m_latency_buffer->back()); // NOLINT
 
          // Break the loop from time to time to update the timestamp and check if we should stop recording
          while (considered_chunks_in_loop < 100) {
            auto iptr = reinterpret_cast<std::uintptr_t>(current_write_pointer); // NOLINT
            if (iptr % alignment_size) {
              // This should never happen
              TLOG() << "Error: Write pointer is not aligned";
            }
            bool failed_write = false;
            if (current_write_pointer + chunk_size < current_end_pointer) {
              // We can write a whole chunk to file
              failed_write |= !::write(m_fd, current_write_pointer, chunk_size);
              if (!failed_write) {
                bytes_written += chunk_size;
              }
              current_write_pointer += chunk_size;
            } else if (current_end_pointer < current_write_pointer) {
              if (current_write_pointer + chunk_size < end_of_buffer_pointer) {
                // Write whole chunk to file
                failed_write |= !::write(m_fd, current_write_pointer, chunk_size);
                if (!failed_write) {
                  bytes_written += chunk_size;
                }
                current_write_pointer += chunk_size;
              } else {
                // Write the last bit of the buffer without using O_DIRECT as it possibly doesn't fulfill the
                // alignment requirement
                fcntl(m_fd, F_SETFL, O_CREAT | O_WRONLY);
                failed_write |= !::write(m_fd, current_write_pointer, end_of_buffer_pointer - current_write_pointer);
                fcntl(m_fd, F_SETFL, m_oflag);
                if (!failed_write) {
                  bytes_written += end_of_buffer_pointer - current_write_pointer;
                }
                current_write_pointer = start_of_buffer_pointer;
              }
            }
 
            if (current_write_pointer == end_of_buffer_pointer) {
              current_write_pointer = start_of_buffer_pointer;
            }
 
            if (failed_write) {
              ++failed_writes;
              ers::warning(CannotWriteToFile(ERS_HERE, inherited::m_output_file));
            }
            considered_chunks_in_loop++;
            // This expression is "a bit" complicated as it finds the last frame that was written to file completely
            inherited::m_next_timestamp_to_record =
              reinterpret_cast<const ReadoutType*>( // NOLINT
                start_of_buffer_pointer +
                (((current_write_pointer - start_of_buffer_pointer) / ReadoutType::fixed_payload_size) *
                 ReadoutType::fixed_payload_size))
                ->get_timestamp();
          }
        }
        current_time = std::chrono::high_resolution_clock::now();
      }
 
      // Complete writing the last frame to file
      if (current_write_pointer != nullptr) {
        const char* last_started_frame =
          start_of_buffer_pointer +
          (((current_write_pointer - start_of_buffer_pointer) / ReadoutType::fixed_payload_size) *
           ReadoutType::fixed_payload_size);
        if (last_started_frame != current_write_pointer) {
          fcntl(m_fd, F_SETFL, O_CREAT | O_WRONLY);
          if (!::write(m_fd,
                       current_write_pointer,
                       (last_started_frame + ReadoutType::fixed_payload_size) - current_write_pointer)) {
            ers::warning(CannotWriteToFile(ERS_HERE, inherited::m_output_file));
          } else {
            bytes_written += (last_started_frame + ReadoutType::fixed_payload_size) - current_write_pointer;
          }
        }
      }
      ::close(m_fd);
 
      inherited::m_next_timestamp_to_record = std::numeric_limits<uint64_t>::max(); // NOLINT (build/unsigned)
 
      TLOG() << "Stopped recording, wrote " << bytes_written << " bytes. Failed write count: " << failed_writes;
      inherited::m_recording.exchange(false);
    }, recording_time_sec);
}

Member Data Documentation

◆ m_fd

template<class ReadoutType , class LatencyBufferType >

int dunedaq::datahandlinglibs::ZeroCopyRecordingRequestHandlerModel< ReadoutType, LatencyBufferType >::m_fd

private

Definition at line 41 of file ZeroCopyRecordingRequestHandlerModel.hpp.

◆ m_oflag

template<class ReadoutType , class LatencyBufferType >

int dunedaq::datahandlinglibs::ZeroCopyRecordingRequestHandlerModel< ReadoutType, LatencyBufferType >::m_oflag

private

Definition at line 42 of file ZeroCopyRecordingRequestHandlerModel.hpp.

The documentation for this class was generated from the following files:

/github/workspace/dunedaq/sourcecode/datahandlinglibs/include/datahandlinglibs/models/ZeroCopyRecordingRequestHandlerModel.hpp
/github/workspace/dunedaq/sourcecode/datahandlinglibs/include/datahandlinglibs/models/detail/ZeroCopyRecordingRequestHandlerModel.hxx

Public Types

Public Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ inherited

Constructor & Destructor Documentation

◆ ZeroCopyRecordingRequestHandlerModel()

Member Function Documentation

◆ conf()

◆ record()

Member Data Documentation

◆ m_fd

◆ m_oflag