ddtMemoryAccessor.hpp

Go to the documentation of this file.

// @copyright
// (c) Copyright ESO 2020
// All Rights Reserved
// ESO (eso.org) is an Intergovernmental Organization, and therefore special
// legal conditions apply.
//
// @file ddtMemoryAccessor.hpp
// @brief Accessor for a shared memory.
//
// This class provides the functionalities to access created shared memories
// (especially read and write functionality).
//
// @author Matthias Grimm, CGI
// @since  2020/01/16
//
 
#ifndef DDTMEMORYACCESSOR_H_
#define DDTMEMORYACCESSOR_H_
 
#include <boost/circular_buffer.hpp>
#include <boost/interprocess/containers/string.hpp>
#include <boost/interprocess/containers/vector.hpp>
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/signals2/signal.hpp>
#include <fstream>
#include <future>
#include <iostream>
 
#include "ddt/ddtConstants.hpp"
#include "ddt/ddtCrc32.hpp"
#include "ddt/ddtLogger.hpp"
 
namespace ip = boost::interprocess;
 
typedef ip::managed_shared_memory::segment_manager segment_manager_t;
 
typedef ip::allocator<void, segment_manager_t> void_allocator;
 
typedef ip::allocator<uint8_t, segment_manager_t> uint8_allocator;
 
typedef ip::vector<uint8_t, uint8_allocator> uint8_vector;
 
typedef ip::allocator<uint16_t, segment_manager_t> uint16_allocator;
 
typedef ip::vector<uint16_t, uint16_allocator> uint16_vector;
 
typedef ip::allocator<char, segment_manager_t> char_allocator;
 
typedef ip::basic_string<char, std::char_traits<char>, char_allocator>
    char_string;
 
typedef boost::signals2::signal<void()> signal_t;
 
namespace ddt {
 
struct DataSampleShared {
  int32_t topic_id = 0;
 
  int32_t meta_data_length;
 
  uint8_vector meta_data;
 
  int32_t sample_id;
 
  uint8_vector data;
 
  DataSampleShared(const int32_t id, const int md_length,
                   const int vector_length, const void_allocator &void_alloc)
      : meta_data_length(md_length),
        meta_data(md_length, uint8_t(), void_alloc),
        sample_id(id),
        data(vector_length, uint8_t(), void_alloc) {}
};
 
struct DataPacketShared {
  char_string data_stream_identifier;
 
  uint32_t checksum;
 
  int32_t sample_length;
 
  int64_t writer_index = -1;
 
  uint64_t timestamp = 0;
 
  DataSampleShared sample;
 
  DataPacketShared(const char *const ds_id, const int32_t check,
                   const int vector_length, const void_allocator &void_alloc)
      : data_stream_identifier(ds_id, void_alloc),
        checksum(check),
        sample_length(vector_length),
        sample(0, META_DATA_LENGTH, vector_length, void_alloc) {}
};
 
struct DataSample {
  int32_t topic_id = 0;
 
  int32_t meta_data_length;
 
  std::vector<uint8_t> meta_data;
 
  int32_t sample_id;
 
  std::vector<uint8_t> data;
 
  DataSample(const int32_t id, const int md_length, const int vector_length)
      : meta_data_length(md_length),
        meta_data(md_length),
        sample_id(id),
        data(vector_length) {}
};
 
struct DataPacket {
  std::string data_stream_identifier;
 
  uint32_t checksum;
 
  int32_t sample_length;
 
  int64_t writer_index = -1;
 
  uint64_t timestamp = 0;
 
  DataSample sample;
 
  DataPacket(const char *const ds_id, const int32_t check,
             const int vector_length)
      : data_stream_identifier(ds_id),
        checksum(check),
        sample_length(vector_length),
        sample(0, META_DATA_LENGTH, vector_length) {}
};
 
class DdtMemoryAccessor {
 public:
  DdtMemoryAccessor();
 
  explicit DdtMemoryAccessor(const std::string &shm_id,
                             const std::string &data_stream_identifier,
                             DdtLogger *logger, const uint64_t time_window = 0,
                             const int32_t reading_interval = 10);
 
  virtual ~DdtMemoryAccessor();
 
  const uint32_t ComputeChecksum(DataSampleShared *const data_sample_shared);
 
  const uint32_t ComputeChecksum(DataSample *const data_sample);
 
  int32_t OpenSharedMemory();
 
  void CloseSharedMemory();
 
  void WriteData(const int32_t writer_index, const int32_t topic_id,
                 const int32_t sample_id, const uint8_t *datavec,
                 const int32_t datavec_size, const uint8_t *metadata_vec,
                 const int32_t metadatavec_size, const uint64_t timestamp);
 
  void StartReading();
 
  void StopReading();
 
  void SetSizeConstraints(const int32_t max_sample_size, const int32_t space);
 
  void Reattach();
 
  void NewData();
 
  void get_data_packet(std::string *stream_identifier, uint32_t *checksum,
                       int32_t *sample_length, int64_t *writer_idx,
                       uint64_t *timestamp, DataSample **sample);
 
  bool get_data_available();
 
  int32_t get_number_of_unread_elements();
 
  signal_t *DataAvailableSignal();
 
  void Reset();
 
  void set_pub_unreg(const bool STATE);
 
  void set_compute_checksum(const bool compute_crc);
 
  bool get_compute_checksum() const;
 
  bool get_is_initialized() const;
 
 private:
  signal_t data_available_signal;
 
  void ReadData();
 
  void SetDefaults();
 
  void Init(const std::string &mem_id, const std::string &stream_id,
            DdtLogger *ddt_logger, const uint64_t time_win,
            const int32_t interval);
 
  void PrintData();
 
  int32_t CreateNewShm();
 
  int32_t SearchCircBuffer();
 
  int32_t SearchWriterIndex();
 
  ip::managed_shared_memory *managed_shm;
 
  typedef ip::allocator<DataPacketShared,
                        ip::managed_shared_memory::segment_manager>
      cb_alloc;
 
  typedef boost::circular_buffer<DataPacketShared, cb_alloc> cb;
 
  std::string shm_id;
  std::string data_stream_identifier;
  uint64_t time_window;      // in [ms]
  int32_t reading_interval;  // in [ms]
  cb *circ_buffer;
  std::atomic<int64_t> *writer_index;
  int32_t local_index;
  int64_t reader_index;
 
  int32_t number_of_unread_elements;
  int32_t circ_buf_capacity;
  int32_t number_of_lost_packages;
 
  std::mutex circ_buffer_mutex;
  std::mutex packets_mutex;
 
  bool is_initialized;
 
  std::promise<void> exit_signal;
  std::future<void> future_object;
 
  std::atomic<bool> reading_active;
  std::atomic<bool> pub_unreg;
  std::atomic<bool> compute_checksum;
 
  int32_t max_data_sample_size;
  int additional_space;
 
  std::list<DataPacketShared *> packets;
 
  DdtLogger *logger;
};
 
}  // namespace ddt
 
#endif /* DDTMEMORYACCESSOR_H_ */