ddtMemoryAccessor.hpp

Go to the documentation of this file.

 
#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()> SignalT;
 
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();

  SignalT *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:
  SignalT 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_ */