Classes
struct	Header

Public Member Functions
size_t	size () throw ()

bool	open (const std::string &nodename, size_t nword, bool recreate=false)

bool	init ()

bool	close ()

bool	unlink ()

bool	lock () throw ()

bool	unlock () throw ()

bool	wait () throw ()

bool	wait (int time) throw ()

bool	notify () throw ()

void	clear ()

bool	isOpened ()

const std::string	getPath () const throw ()

Header *	getHeader () throw ()

int *	getBuffer () throw ()

bool	isWritable (int nword) throw ()

bool	isReadable (int nword) throw ()

unsigned int	write (const int *buf, unsigned int nword, bool fouce, unsigned int serial=0)

unsigned int	write (const int *buf, unsigned int nword, bool fouce, unsigned int serial, bool unlocked)

unsigned int	read (int buf, bool fouce, bool unlocked, Header hdr=NULL)

Private Attributes
std::string	m_path

SharedMemory	m_memory

MMutex	m_mutex

MCond	m_cond

Header *	m_header

int *	m_buf

unsigned int	m_nword

Detailed Description

Definition at line 21 of file SharedEventBuffer.h.

Constructor & Destructor Documentation

◆ SharedEventBuffer()

SharedEventBuffer ( )

inline

Definition at line 37 of file SharedEventBuffer.h.

    {
      m_buf = NULL;
      m_nword = 0;
    }

◆ ~SharedEventBuffer()

~SharedEventBuffer ( )

inline

Definition at line 42 of file SharedEventBuffer.h.

    {
      if (m_buf != NULL) m_memory.close();
    }

Member Function Documentation

◆ clear()

void clear ( )

Definition at line 64 of file SharedEventBuffer.cc.

{
  if (m_buf == NULL) return;
  m_mutex.lock();
  memset(m_header, 0, sizeof(Header));
  for (unsigned long long i = 0; i < m_nword; i++) {
    m_buf[i] = 0;
  }
  m_mutex.unlock();
}

◆ close()

bool close ( )

Definition at line 75 of file SharedEventBuffer.cc.

{
  m_memory.close();
  return true;
}

◆ getBuffer()

int * getBuffer	(	)
throw	(
	)

inline

Definition at line 65 of file SharedEventBuffer.h.

65{ return m_buf; }

◆ getHeader()

Header * getHeader	(	)
throw	(
	)

inline

Definition at line 64 of file SharedEventBuffer.h.

64{ return m_header; }

◆ getPath()

const std::string getPath	(	)	const
throw	(
	)

inline

Definition at line 63 of file SharedEventBuffer.h.

63{ return m_path; }

◆ init()

bool init ( )

Definition at line 51 of file SharedEventBuffer.cc.

{
  if (m_buf == NULL) return false;
  m_mutex.init();
  m_cond.init();
  memset(m_header, 0, sizeof(Header));
  memset(m_buf, 0, sizeof(int) * m_nword);
  //for (unsigned long long i = 0; i < m_nword; i++) {
  //  m_buf[i] = 0;
  //}
  return true;
}

◆ isOpened()

bool isOpened ( )

inline

Definition at line 60 of file SharedEventBuffer.h.

60{ return m_memory.isOpened(); }

◆ isReadable()

bool isReadable	(	int	nword	)
throw	(
	)

Definition at line 127 of file SharedEventBuffer.cc.

{
  if (m_buf == NULL) return false;
  m_mutex.lock();
  bool readable = m_header->nword_in - m_header->nword_out >= m_nword - (nword + 1);
  m_mutex.unlock();
  return readable;
 
}

◆ isWritable()

bool isWritable	(	int	nword	)
throw	(
	)

Definition at line 118 of file SharedEventBuffer.cc.

{
  if (m_buf == NULL) return false;
  m_mutex.lock();
  bool writable = m_header->nword_in - m_header->nword_out < m_nword - (nword + 1);
  m_mutex.unlock();
  return writable;
}

◆ lock()

bool lock	(	)
throw	(
	)

Definition at line 88 of file SharedEventBuffer.cc.

{
  if (m_buf == NULL) return false;
  return m_mutex.lock();
}

◆ notify()

bool notify	(	)
throw	(
	)

Definition at line 112 of file SharedEventBuffer.cc.

{
  if (m_buf == NULL) return false;
  return m_cond.broadcast();
}

◆ open()

bool open	(	const std::string &	nodename,
		size_t	nword,
		bool	recreate = `false`
	)

Definition at line 23 of file SharedEventBuffer.cc.

{
  m_nword = nword;
  std::string username = getenv("USER");
  m_path = "/storage_info_" + username + "_" + nodename;
  LogFile::debug("%s", m_path.c_str());
  //if (recreate) SharedMemory::unlink(m_path);
  if (!m_memory.open(m_path, size())) {
    perror("shm_open");
    LogFile::fatal("Failed to open %s", m_path.c_str());
    return false;
  }
  char* buf = (char*) m_memory.map(0, size());
  if (buf == NULL) {
    return false;
  }
  m_mutex = MMutex(buf);
  buf += m_mutex.size();
  m_cond = MCond(buf);
  buf += m_cond.size();
  m_header = reinterpret_cast<Header*>(buf);
  buf += sizeof(Header);
  m_buf = (int*)buf;
  if (recreate) init();
  return true;
}

◆ read()

unsigned int read	(	int *	buf,
		bool	fouce,
		bool	unlocked,
		SharedEventBuffer::Header *	hdr = `NULL`
	)

Definition at line 183 of file SharedEventBuffer.cc.

{
  if (m_buf == NULL) return 0;
  if (!unlocked) m_mutex.lock();
  m_header->nreader++;
  while (!fouce && m_header->nwriter > 0) {
    m_cond.wait(m_mutex);
  }
 
  unsigned int nword = 0;
  while (true) {
    unsigned int i_w = m_header->nword_in % m_nword;
    unsigned int i_r = m_header->nword_out % m_nword;
    nword = m_buf[i_r];
    if (nword > 0) {
      if (m_header->nword_in - m_header->nword_out >= (nword + 1)) {
        if (i_w > i_r) {
          memcpy(buf, (m_buf + i_r + 1), sizeof(int) * nword);
          break;
        } else if (i_w < i_r) {
          if (m_nword - i_r > nword) {
            memcpy(buf, (m_buf + i_r + 1), sizeof(int) * nword);
            break;
          } else {
            unsigned int count = m_nword - i_r;
            memcpy(buf, (m_buf + i_r + 1), sizeof(int) * count);
            if (nword > count) {
              memcpy(buf + count, m_buf, sizeof(int) * (nword - count));
            }
            break;
          }
        }
      }
    }
    m_header->nreader--;
    m_cond.wait(m_mutex);
    m_header->nreader++;
  }
  m_header->nword_out += nword + 1;
  unsigned int count = ++m_header->count_out;
  m_header->nreader--;
  if (hdr != NULL) {
    memcpy(hdr, m_header, sizeof(SharedEventBuffer::Header));
  }
  m_cond.broadcast();
  if (!unlocked) m_mutex.unlock();
  return count;
}

◆ size()

size_t size	(	)
throw	(
	)

Definition at line 17 of file SharedEventBuffer.cc.

{
  return m_mutex.size() + m_cond.size() +
         sizeof(Header) + sizeof(int) * (m_nword);
}

◆ unlink()

bool unlink ( )

Definition at line 81 of file SharedEventBuffer.cc.

{
  m_memory.close();
  m_memory.unlink();
  return true;
}

◆ unlock()

bool unlock	(	)
throw	(
	)

Definition at line 94 of file SharedEventBuffer.cc.

{
  if (m_buf == NULL) return false;
  return m_mutex.unlock();
}

◆ wait() [1/2]

bool wait	(	)
throw	(
	)

Definition at line 100 of file SharedEventBuffer.cc.

{
  if (m_buf == NULL) return false;
  return m_cond.wait(m_mutex);
}

◆ wait() [2/2]

bool wait	(	int	time	)
throw	(
	)

Definition at line 106 of file SharedEventBuffer.cc.

{
  if (m_buf == NULL) return false;
  return m_cond.wait(m_mutex, time, 0);
}

◆ write() [1/2]

unsigned int write	(	const int *	buf,
		unsigned int	nword,
		bool	fouce,
		unsigned int	serial,
		bool	unlocked
	)

Definition at line 137 of file SharedEventBuffer.cc.

{
  if (m_buf == NULL) return 0;
  if (nword == 0) return 0;
  if (nword > m_nword) return -1;
  if (!unlocked) m_mutex.lock();
  m_header->nwriter++;
  while (!fouce && m_header->nreader > 0) {
    m_cond.wait(m_mutex);
  }
 
  while (true) {
    unsigned int i_w = m_header->nword_in % m_nword;
    unsigned int i_r = m_header->nword_out % m_nword;
    if ((serial == 0 || serial - 1 == m_header->count_in) &&
        m_header->nword_in - m_header->nword_out < m_nword - (nword + 1)) {
      if (i_w >= i_r) {
        unsigned int count = m_nword - i_w;
        if (nword + 1 < count) {
          m_buf[i_w] = nword;
          memcpy((m_buf + i_w + 1), buf, sizeof(int) * nword);
        } else {
          m_buf[i_w] = nword;
          memcpy((m_buf + i_w + 1), buf, sizeof(int) * count);
          if (nword >= count)
            memcpy(m_buf, buf + count, sizeof(int) * (nword - count));
        }
      } else {
        m_buf[i_w] = nword;
        memcpy((m_buf + i_w + 1), buf, sizeof(int) * nword);
      }
      break;
    }
    m_header->nwriter--;
    m_cond.wait(m_mutex);
    m_header->nwriter++;
  }
  m_header->nword_in += nword + 1;
  unsigned int count = ++m_header->count_in;
  m_header->nwriter--;
  m_cond.broadcast();
  if (!unlocked) m_mutex.unlock();
  return count;
}

◆ write() [2/2]

unsigned int write	(	const int *	buf,
		unsigned int	nword,
		bool	fouce,
		unsigned int	serial = `0`
	)

inline

Definition at line 68 of file SharedEventBuffer.h.

    {
      return write(buf, nword, fouce, serial, false);
    }

Member Data Documentation

◆ m_buf

int* m_buf

private

Definition at line 83 of file SharedEventBuffer.h.

◆ m_cond

MCond m_cond

private

Definition at line 81 of file SharedEventBuffer.h.

◆ m_header

Header* m_header

private

Definition at line 82 of file SharedEventBuffer.h.

◆ m_memory

SharedMemory m_memory

private

Definition at line 79 of file SharedEventBuffer.h.

◆ m_mutex

MMutex m_mutex

private

Definition at line 80 of file SharedEventBuffer.h.

◆ m_nword

unsigned int m_nword

private

Definition at line 84 of file SharedEventBuffer.h.

◆ m_path

std::string m_path

private

Definition at line 78 of file SharedEventBuffer.h.

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

daq/storage/include/SharedEventBuffer.h
daq/storage/src/SharedEventBuffer.cc

Classes

Public Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ SharedEventBuffer()

◆ ~SharedEventBuffer()

Member Function Documentation

◆ clear()

◆ close()

◆ getBuffer()

◆ getHeader()

◆ getPath()

◆ init()

◆ isOpened()

◆ isReadable()

◆ isWritable()

◆ lock()

◆ notify()

◆ open()

◆ read()

◆ size()

◆ unlink()

◆ unlock()

◆ wait() [1/2]

◆ wait() [2/2]

◆ write() [1/2]

◆ write() [2/2]

Member Data Documentation

◆ m_buf

◆ m_cond

◆ m_header

◆ m_memory

◆ m_mutex

◆ m_nword

◆ m_path