ddtMemoryManager.hpp

Go to the documentation of this file.

 
#ifndef DDTMEMORYMANAGER_HPP_
#define DDTMEMORYMANAGER_HPP_
 
#include <boost/circular_buffer.hpp>
#include <boost/interprocess/containers/string.hpp>
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wmaybe-uninitialized"
#include <boost/interprocess/managed_shared_memory.hpp>
/* ignore this boost internal warning from the file above */
#pragma GCC diagnostic pop
#include <boost/tokenizer.hpp>
#include <csignal>
#include <fstream>
#include <iostream>
 
#include "ddt/ddtConstants.hpp"
#include "ddt/ddtErrorCodes.hpp"
#include "ddt/ddtLogger.hpp"
#include "ddt/ddtMemoryAccessor.hpp"
 
namespace ip = boost::interprocess;
 
namespace ddt {

struct Shm {
  ip::managed_shared_memory *managed_shm;

  int max_data_sample_size;

  int number_of_samples;

  DdtMemoryAccessor *memory_accessor;
};

class DdtMemoryManager {
 public:
  DdtMemoryManager(DdtLogger *logger, const std::string broker_port);

  virtual ~DdtMemoryManager();

  int32_t CreateSharedMemory(const std::string data_stream_identifier,
                             const int32_t max_data_sample_size,
                             const int32_t number_of_samples);

  bool DeleteSharedMemory(const std::string data_stream_identifier);

  bool CheckIfShmExists(const std::string data_stream_identifier);

  int32_t get_max_data_sample_size(const std::string data_stream_identifier);

  int32_t get_number_of_samples(const std::string data_stream_identifier);

  DdtMemoryAccessor *get_memory_accessor(
      const std::string data_stream_identifier);

  std::string get_shm_id(const std::string data_stream_identifier) const;

  std::string get_shm_full_path(const std::string data_stream_identifier) const;

  int32_t GetMaxPossibleBufferSize(const int32_t max_data_sample_size);
 
 protected:
  bool CheckSpaceAvailability(const int32_t max_data_sample_size,
                              const int buffer_size);

  const uint64_t GetAvailableSpace();
 
 private:
  void Init(DdtLogger *ddt_logger, const std::string broker_port);

  int32_t CreateNewSharedMemory(const std::string data_stream_identifier,
                                const int32_t max_data_sample_size,
                                const int32_t number_of_samples,
                                const std::string shm_id);

  int32_t CreateCircularBuffer(const std::string data_stream_identifier,
                               const int32_t max_data_sample_size,
                               const int32_t number_of_samples);

  std::map<std::string, Shm> shm_map;
 
  std::mutex shm_mutex;
  std::string port;
  DdtLogger *logger;
 
  const std::string SHM_PATH = "/dev/shm";
  const int AVAILABLE_POSITION = 3;
  const int BYTES_PER_KBYTE = 1024;
  static const int CHAR_LEN = 128;
  const int PERM_TYPE = 0664;
  const int NO_SUCH_SHM = 7;
};
 
}  // namespace ddt
 
#endif /* DDTMEMORYMANAGER_HPP_ */