shm_pub_sub.hpp

Go to the documentation of this file.

 
#ifndef __SHM_PS_LIB_H__
#define __SHM_PS_LIB_H__
 
#include <iostream>
#include <limits>
#include <string>
#include <regex>
#include <stdexcept>
#include <mutex>
extern "C" {
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/time.h>
#include <pthread.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
}
#include "shm_base.hpp"
 
namespace irlab
{
 
namespace shm
{
 
// ****************************************************************************
// ****************************************************************************
template <typename T>
class Publisher
{
public:
  Publisher(std::string name = "", int buffer_num = 3, PERM perm = DEFAULT_PERM);
  ~Publisher() = default;
 
    // コピーは禁止
  Publisher(const Publisher&) = delete;
  Publisher& operator=(const Publisher&) = delete;
 
  // ムーブコンストラクタ：ポインタを奪い、元を nullptr に
  Publisher(Publisher&& other) noexcept = default;
 
  void publish(const T& data);
 
private:
  std::string shm_name;
  int shm_buf_num;
  PERM shm_perm;
  std::unique_ptr<SharedMemory> shared_memory;
  std::unique_ptr<RingBuffer> ring_buffer;
 
  size_t data_size;
};
 
// ****************************************************************************
// ****************************************************************************
template <typename T>
class Subscriber
{
public:
  Subscriber(std::string name = "");
  ~Subscriber() = default;
 
  // コピーは禁止
  Subscriber(const Subscriber&) = delete;
  Subscriber& operator=(const Subscriber&) = delete;
 
  // ムーブコンストラクタ：ポインタを奪い、元を nullptr に
  Subscriber(Subscriber&& other) noexcept = default;
  
  const T subscribe(bool *state);
  bool waitFor(uint64_t timeout_usec);
  void setDataExpiryTime_us(uint64_t time_us);
  
private:
  std::string shm_name;
  std::unique_ptr<SharedMemory> shared_memory;
  std::unique_ptr<RingBuffer> ring_buffer;
  int current_reading_buffer;
  uint64_t data_expiry_time_us;
};
 
// ****************************************************************************
// Function Definications
// 関数定義
// （テンプレートクラス内の関数の定義はコンパイル時に実体化するのでヘッダに書く）
// ****************************************************************************
 
template <typename T>
Publisher<T>::Publisher(std::string name, int buffer_num, PERM perm)
: shm_name(name)
, shm_buf_num(buffer_num)
, shm_perm(perm)
, shared_memory(nullptr)
, ring_buffer(nullptr)
, data_size(sizeof(T))
{
  if (!std::is_standard_layout<T>::value)
  {
    throw std::runtime_error("shm::Publisher: Be setted not POD class!");
  }
 
  if (name.empty())
  {
    throw std::runtime_error("shm::Publisher: Please set name!");
  }
 
  try {
    shared_memory = std::make_unique<SharedMemoryPosix>(shm_name, O_RDWR | O_CREAT, shm_perm);
  } catch (const std::runtime_error& e) {
    throw std::runtime_error("shm::Publisher: " + std::string(e.what()));
  }
  shared_memory->connect(RingBuffer::getSize(sizeof(T), shm_buf_num));
 
  if (shared_memory->isDisconnected())
  {
    throw std::runtime_error("shm::Publisher: Cannot get memory!");
  }
 
  ring_buffer = std::make_unique<RingBuffer>(shared_memory->getPtr(), sizeof(T), shm_buf_num);
}
 
 
template <typename T>
void
Publisher<T>::publish(const T& data)
{
  int oldest_buffer = ring_buffer->getOldestBufferNum();
  for (size_t i = 0; i < 10; i++)
  {
    if (ring_buffer->allocateBuffer(oldest_buffer))
    {
      break;
    }
    usleep(1000); // Wait for 1ms
    oldest_buffer = ring_buffer->getOldestBufferNum();
  }
 
  (reinterpret_cast<T *>(ring_buffer->getDataList()))[oldest_buffer] = data;
 
  struct timespec t;
  clock_gettime(CLOCK_MONOTONIC_RAW, &t);
  ring_buffer->setTimestamp_us(((uint64_t) t.tv_sec * 1000000L) + ((uint64_t) t.tv_nsec / 1000L), oldest_buffer);
  
  ring_buffer->signal();
}
 
 
template <typename T>
Subscriber<T>::Subscriber(std::string name)
: shm_name(name)
, shared_memory(nullptr)
, ring_buffer(nullptr)
, current_reading_buffer(0)
, data_expiry_time_us(2000000)
{
  if (!std::is_standard_layout<T>::value)
  {
    throw std::runtime_error("shm::Subscriber: Be setted not POD class!");
  }
 
  if (name.empty())
  {
    throw std::runtime_error("shm::Subscriber: Please set name!");
  }
 
  try {
    shared_memory = std::make_unique<SharedMemoryPosix>(shm_name, O_RDWR, static_cast<PERM>(0));
  } catch (const std::runtime_error& e) {
    throw std::runtime_error("shm::Subscriber: " + std::string(e.what()));
  }
}
 
 
template <typename T>
const T
Subscriber<T>::subscribe(bool *is_success)
{
  if (shared_memory->isDisconnected())
  {
    if (ring_buffer != nullptr)
    {
      ring_buffer.reset();
    }
    shared_memory->connect();
    if (shared_memory->isDisconnected())
    {
      *is_success = false;
      return T();
    }
    try {
      if (shared_memory->getPtr() == nullptr) {
        *is_success = false;
        return T();
      }
      ring_buffer = std::make_unique<RingBuffer>(shared_memory->getPtr());
    } catch (const std::bad_alloc& e)
    {
      *is_success = false;
      return T();
    }
    ring_buffer->setDataExpiryTime_us(data_expiry_time_us);
  }
  int newest_buffer = ring_buffer->getNewestBufferNum();
  if (newest_buffer < 0)
  {
    *is_success = false;
    return (reinterpret_cast<T*>(ring_buffer->getDataList()))[current_reading_buffer];
  }
  *is_success = true;
  current_reading_buffer = newest_buffer;
  return (reinterpret_cast<T*>(ring_buffer->getDataList()))[current_reading_buffer];
}
 
 
template <typename T>
bool
Subscriber<T>::waitFor(uint64_t timeout_usec)
{
  if (shared_memory->isDisconnected())
  {
    if (ring_buffer != nullptr)
    {
      ring_buffer.reset();
    }
    shared_memory->connect();
    if (shared_memory->isDisconnected())
    {
      return false;
    }
    ring_buffer = std::make_unique<RingBuffer>(shared_memory->getPtr());
    ring_buffer->setDataExpiryTime_us(data_expiry_time_us);
  }
 
  return ring_buffer->waitFor(timeout_usec);
}
 
 
template <typename T>
void
Subscriber<T>::setDataExpiryTime_us(uint64_t time_us)
{
  data_expiry_time_us = time_us;
  if (ring_buffer != nullptr)
  {
    ring_buffer->setDataExpiryTime_us(data_expiry_time_us);
  }
}
 
 
}
 
}
 
#endif /* __SHM_PS_LIB_H__ */