ngcbSHM.hpp

Go to the documentation of this file.

#ifndef ngcbSHM_H
#define ngcbSHM_H
 
 
#ifndef __cplusplus
#error This is a C++ include file and cannot be used from plain C
#endif
 
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <poll.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <sys/time.h>
#include <atomic>
 
#include <ngc/core/ngcbSHMDB.hpp>
 
#ifdef _PTHREAD_H
#define NGCB_SHM_THR 1
#endif

#define ngcbSHM_MAX_CLIENTS 8
 
/*
 * Shared memory id special result codes
 */
#define ngcbSHM_ID_ERR   -1  //< IPC error
#define ngcbSHM_ID_RETRY -2  //< try again
#define ngcbSHM_ID_IVLD  -3  //< invalid

#define ngcbSHM_VERSION 1

struct ngcb_shmevt_t {
  int semId;

  int reserved1;

  int shmId;

  unsigned int event;

  ngcb_shmevt_t() {semId = -1; reserved1 = 0x0; shmId = -1; event = 0;}
};

struct ngcb_shmio_t {
  int semId;
  
  int retry;

  unsigned int event;

  int reserved;

  ngcb_shmio_t() {semId = -1; retry = 0; event = 0; reserved = 0x0;}
};

struct ngcb_shmhdr_t {
  ngcb_shmio_t io;
  
  int frame;

  int bitPix;

  int sx;
  
  int sy;
  
  int nx;
  
  int ny;

  int endian;

  int fullFrame;

  int bzero;

  int fcnt;

  int fcnt0;

  int expCnt;

  int ndit;

  int flags;

  int status;

  char bytes[308];

  ngcb_shmhdr_t() {
    frame = 0;
    bitPix = 8;
    sx = 0; sy = 0; nx = 0; ny = 0;
    endian = 1;
    fullFrame = 0;
    bzero = 0;
    status = 0;
    fcnt = 0;
    fcnt0 = 0;
    ndit = 0;
    expCnt = 0;
    flags = 0;
    memset(bytes, 0, sizeof(bytes));
  }
};

class ngcbSHMTHR {
public:
  ngcbSHMTHR() {}

  virtual ~ngcbSHMTHR() {}
  
  virtual void Thread() {};
 
protected:
private:
};
 
#ifdef NGCB_SHM_THR
static void *ngcbShmThread(void *arg) {
  ngcbSHMTHR *evt = static_cast<ngcbSHMTHR *>(arg);
  sigset_t set;
 
  sigfillset(&set);
  pthread_sigmask(SIG_BLOCK, &set, (sigset_t *)NULL);
 
  if (evt != NULL) evt->Thread();
  
  return (NULL);
}
#endif

class ngcbSHM: public ngcbSHMTHR {
public:
  char name[64];

  ngcb_shmhdr_t *sharedHeader;  

  ngcbSHM() {
    /*
     * Initialize member data
     */
    Init_();
    initialized_ = 1;
  }

  explicit ngcbSHM(const char *s) {
    /*
     * Initialize member data
     */
    Init_();
 
    /*
     * Check name
     */
    if (s == NULL)
      {
      strcpy(ErrMsg(), "invalid name");
      return;
      }
    else if ((strlen(s) < 1) || (strlen(s) >= (sizeof(name) - 1)))
      {
      strcpy(ErrMsg(), "invalid name");
      return;
      }
    else if (s[0] == '.')
      {
      extPipe_ = 1; // use external pipe (ngcbShmEvt dispatcher)
      strcpy(name, s + 1);
      }
    else if (s[0] == ':')
      {
      intPipe_ = 1; // use internal pipe (dispatcher thread)
      strcpy(name, s + 1);
      }
    else
      {
      strcpy(name, s);
      }
 
    if (strlen(name) == 0)
      {
      strcpy(ErrMsg(), "invalid name");
      return;
      }
    else if (extPipe_)
      {
      /*
       * Open external event pipe
       */
      regCnt_ = 0; // not used
      fd_ = OpenPipe_();
      if (fd_ < 0)
    {
    return;
    }
      }
    else
      {
      /*
       * Register to event interface
       */
      regCnt_ = 10;
      if (Register_() != 0)
    {
    return;
    }
      }
 
    initialized_ = 1;
  }

  virtual ~ngcbSHM() {
    /*
     * Shutdown event pipes (if applicable)
     */
    ClosePipe_();
 
    if (shm_ != NULL)
      {
      /*
       * Detach from shared memory
       */
      shmdt(shm_); shm_ = NULL;
      }
 
    if (evt_ != NULL)
      {
      /*
       * Detach from event interface
       */
      shmdt(evt_); evt_ = NULL;
      }
 
    /*
     * Check, if we have to cleanup the event resources
     */
    if (semEvt_ >= 0)
      {
      /*
       * Detach from events
       */
      if (clientId_ >= 0) DetachEvt_();
 
      if (EvtTryLock_() == 0)
    {
    if (NumClient() == 1)
      {
      /*
       * We are the last user - remove the shared memory event
       */
      if (evtId_ != -1)
        {
        shmctl(evtId_, IPC_RMID, 0); evtId_ = -1;
        }
      
      /*
       * Remove the semaphore
       */
      SemEvtRemove_();
 
      /*
       * Remove the IPC-key file
       */
      if (strlen(name) > 0)
        {
        char fileName[256];
        snprintf(fileName, sizeof(fileName), "/tmp/ngcshmio_%u_%s", 
             getuid(), name);
        remove(fileName);
        }
      }
    else
      {
      struct sembuf semOp[4];
      
      /*
       * Unregister
       */
      memset(&semOp[0], 0, sizeof(struct sembuf));
      semOp[0].sem_num = 1;
      semOp[0].sem_op = -1;
      semOp[0].sem_flg = SEM_UNDO;
      if (semop(semEvt_, semOp, 1) != 0)
        {
        if (errno == EIDRM) semEvt_ = -1;  // semaphore has been removed
        }
      EvtUnlock_();
      }
    }
      }
  }

  int Initialized() {return (initialized_);}
  
  char *ErrMsg() {return (erms_);}

  int Fd() {return (fd_);}

  int IntPipe() {return (intPipe_);}

  int NumClient() {
    if (semEvt_ >= 0)
      {
      ngcbSEM_CTL ctl;
      int val = semctl(semEvt_, 1, GETVAL, ctl);
      if (val == -1)
    {
    sprintf(ErrMsg(), "semaphore error - %s", strerror(errno));
    return (0);
    }
      erms_[0] = '\0';
      return (val);
      }
    else
      {
      return (1);
      }
  }

  void Remove() {
    /*
     * Remove event semaphore set
     */
    SemEvtRemove_();
 
    if ((evtId_ != -1) && (evt_ != NULL))
      {
      /*
       * Delete the shared memory event
       */ 
      shmdt(evt_); evt_ = NULL;
      shmctl(evtId_, IPC_RMID, 0); evtId_ = -1;
      }
 
    /*
     * Remove the IPC-key file
     */
    if (strlen(name) > 0)
      {
      char fileName[256];
      snprintf(fileName, sizeof(fileName), "/tmp/ngcshmio_%u_%s", getuid(), 
           name);
      remove(fileName);
      }
    
    /*
     * Interface can no longer be used
     */
    initialized_ = 0;
  }

  int WrInit(int semId) {
    ngcbSEM_CTL ctl;
 
    /*
     * Initialize semaphore
     */
    ctl.val = 1; // write lock
    if (semctl(semId, 0, SETVAL, ctl) == -1)
      {
      sprintf(ErrMsg(), "semaphore init failed - %s", strerror(errno));
      return (-1);
      }
    else
      {
      ctl.val = 0; // readers unlocked
      if (semctl(semId, 1, SETVAL, ctl) == -1)
    {
    sprintf(ErrMsg(), "semaphore init failed - %s", strerror(errno));
    return (-1);
    }
      }
 
    return (0);
  }

  int WrLock(int semId) {
    if (semop(semId, semOpWrLock_, 3) != 0)
      {
      if (errno == EINTR)
    {
    while (semop(semId, semOpWrLock_, 3) != 0)
      {
      if (errno != EINTR)
        {
        sprintf(ErrMsg(), "cannot lock semaphore - %s", strerror(errno));
        return (-1);
        }
      }
    }
      else
    {
    sprintf(ErrMsg(), "cannot lock semaphore - %s", strerror(errno));
    return (-1);
    }
      }
 
    return (0);
  }
 
  int WrTryLock(int semId) {
    /*
     * Try to get write-lock
     */
    if (semop(semId, semOpWrTryLock_, 3) != 0)
      {
      return (-1);
      }
    else
      {
      return (0);
      }
  }

  int WrUnlock(int semId) {
    if (semop(semId, semOpWrUnlock_, 1) != 0)
      {
      sprintf(ErrMsg(), "semaphore error - %s", strerror(errno));
      return (-1);
      }
    else
      {
      return (0);
      }
  }

  void RdWait(int semId) {
    if (semId >= 0)
      {
      if (semop(semId, semOpRdWait_, 1) != 0)
    {
    if (errno == EINTR)
      {
      while (semop(semId, semOpRdWait_, 1) != 0)
        {
        if (errno != EINTR)
          {
          return;
          }
        }
      }
    }
      }
  }

  int RdCnt(int semId) {
    ngcbSEM_CTL ctl;
    return (semctl(semId, 1, GETVAL, ctl));
  }

  void RdUnlock(int semId) {
    if (extPipe_)
      {
      if ((fd_ >= 0) && !extPipeErr_)
    {
    int ack = 1;
    if (write(fd_, &ack, sizeof(int)) != sizeof(int)) extPipeErr_ = 1;
    }
      }
    else
      {
      RdUnlock_(semId);
      }
  }

  void RdUnlock(ngcb_shmhdr_t &hdr) {RdUnlock(hdr.io.semId);}

  int ShmLock(int shmId, int noWait = 0) {
    void *m;
 
    if (shmId < 0)
      {
      strcpy(ErrMsg(), "invalid shared memory id");
      return (-1);
      }
 
    /*
     * Attach to shared memory
     */
    m = shmat(shmId, 0, SHM_RDONLY);
    if (m == (void *)-1)
      {
      sprintf(ErrMsg(), "cannot lock shared memory - %s", strerror(errno));
      return (-1);
      }
    else
      {
      ngcb_shmio_t *io = static_cast<ngcb_shmio_t *>(m);
 
      /*
       * Get read-lock
       */
      if (RdLock_(io->semId, noWait) != 0)
    {
    sprintf(ErrMsg(), "cannot lock shared memory - %s", strerror(errno));
    shmdt(m);
    return (-1);
    }
 
      shmdt(m);
      return (0);
      }
  }

  void ShmNoUndo() {
    semOpRdLock_[1].sem_flg = 0;
    semOpRdUnlock_[0].sem_flg = IPC_NOWAIT;
  }

  int ShmUnlock(int shmId) {
    void *m;
 
    if (shmId < 0)
      {
      strcpy(ErrMsg(), "invalid shared memory id");
      return (-1);
      }
 
    /*
     * Attach to shared memory
     */
    m = shmat(shmId, 0, SHM_RDONLY);
    if (m == (void *)-1)
      {
      sprintf(ErrMsg(), "cannot lock shared memory - %s", strerror(errno));
      return (-1);
      }
    else
      {
      ngcb_shmio_t *io = static_cast<ngcb_shmio_t *>(m);
 
      /*
       * Release read-lock
       */
      RdUnlock_(io->semId);
      shmdt(m);
      return (0);
      }
 
  }

  void *AttachBuffer(int shmId, void *hdr, int noWait = 0) {
    void *buffer = NULL;
    ngcb_shmio_t *hdr_io = static_cast<ngcb_shmio_t *>(hdr);
 
    /*
     * Reset the retry-flag in any case
     */
    hdr_io->retry = 0;
 
    if (shmId < 0)
      {
      strcpy(ErrMsg(), "invalid shared memory id");
      return (NULL);
      }
 
    /*
     * Attach to shared memory
     */
    shm_ = shmat(shmId, 0, SHM_RDONLY);
    if (shm_ != (void *)-1)
      {
      ngcb_shmio_t *io = static_cast<ngcb_shmio_t *>(shm_);
 
      /*
       * Get read-lock
       */
      if (RdLock_(io->semId, noWait) != 0)
    {
    hdr_io->retry = 1;
    sprintf(ErrMsg(), "cannot lock buffer - %s", strerror(errno));
    shmdt(shm_); shm_ = NULL;
    return (NULL);
    }
 
      /*
       * Get header
       */
      memcpy(hdr, shm_, sizeof(ngcb_shmhdr_t));
 
      /*
       * Align the event counter. So the Wait() function can ensure 
       * that we do not receive the same event data again.
       */
      evtCnt_ = io->event;
 
      /*
       * Assign buffer
       */
      buffer = (void *)(((char *)shm_) + sizeof(ngcb_shmhdr_t));
      }
    else
      {
      shm_ = NULL;
      sprintf(ErrMsg(), "cannot attach to shared memory - %s", 
          strerror(errno));
      }
 
    return (buffer);
  }

  void *TryAttachBuffer(int shmId, void *hdr) {
    return (AttachBuffer(shmId, hdr, 1));
  }

  void DetachBuffer() {
    if (shm_ != NULL)
      {
      ngcb_shmio_t *io = static_cast<ngcb_shmio_t *>(shm_);
 
      /*
       * Release read-lock
       */
      RdUnlock_(io->semId);
      
      /*
       * Detach from shared memory
       */
      shmdt(shm_); shm_ = NULL;
      }
  }

  unsigned int EvtCnt() {return (evtCnt_);}

  unsigned int NextEvt() {
    unsigned int n = evtCnt_ + 1;
    if (n == 0) n = 1;
    return (n);
  }

  int AttachEvt() {
    struct sembuf semOp[1];
    int i;
    
    if (extPipe_)
      {
      if ((fd_ < 0) || extPipeErr_)
    {
    strcpy(ErrMsg(), "event interface not initialized");
    return (-1);
    }
 
      return (0);
      }
    else if (semEvt_ < 0)
      {
      /*
       * Not yet initialized
       */
      strcpy(ErrMsg(), "event interface not initialized");
      return (-1);
      }
    else if (clientId_ >= 0)
      {
      /*
       * We are already attached
       */
      return (0);
      }
    else if (EvtLock_() != 0)
      {
      /*
       * We cannot proceed
       */
      return (-1);
      }
    
    cancelled_ = false;
    memset(&semOp[0], 0, sizeof(struct sembuf));
    semOp[0].sem_op = -2;
    semOp[0].sem_flg = IPC_NOWAIT|SEM_UNDO;
    
    for (i=0;i<ngcbSHM_MAX_CLIENTS;i++)
      {
      semOp[0].sem_num = static_cast<unsigned short>(2 + i);
      if (semop(semEvt_, semOp, 1) == 0)
    {
    /*
     * We are attached
     */
    clientId_ = i;
    EvtUnlock_();
 
    if (intPipe_)
      {
      /*
       * Start event dispatcher thread
       */
      thr_ = 1; // thread interface enabled
      if (StartThread_() != 0)
        {
        thr_ = 0; // thread interface disabled
        return (-1);
        }
      }
 
    return (0);
    }
      else if (errno == EIDRM)
    {
    /*
     * Event semaphore has been removed
     */
    semEvt_ = -1;
    }
      }
 
    EvtUnlock_();
    
    strcpy(ErrMsg(), "too many clients");
    return (-1);
  }

  int Wait(int noWait = 0) {
    int shmId = ngcbSHM_ID_IVLD;
    if (extPipe_)
      {
      if ((fd_ >= 0) && !extPipeErr_)
    {
    int ack = 1;
    int res = WaitPipe_(noWait);
    if (res < 0)
      {
      return (ngcbSHM_ID_ERR);
      }
    else if (res == 0)
      {
      return (ngcbSHM_ID_RETRY);
      }
 
    /*
     * Read shared memory id
     */
    if (read(fd_, &shmId, sizeof(int)) != sizeof(int))
      {
      extPipeErr_ = 1;
      strcpy(ErrMsg(), "error reading event header");
      return (ngcbSHM_ID_ERR);
      }
 
    if (shmId >= 0)
      {
      shm_ = shmat(shmId, 0, SHM_RDONLY);
      if (shm_ != (void *)-1)
        {
        ngcb_shmio_t *io = static_cast<ngcb_shmio_t *>(shm_);
 
        /*
         * Get event counter
         */
        evtCnt_ = io->event;
        shmdt(shm_); shm_ = NULL;
        }
      else
        {
        shm_ = NULL;
        sprintf(ErrMsg(), "shared memory error - %s", strerror(errno));
        return (ngcbSHM_ID_ERR);
        }
      }
 
    /*
     * Send acknowledge
     */
    if (write(fd_, &ack, sizeof(int)) != sizeof(int))
      {
      extPipeErr_ = 1;
      strcpy(ErrMsg(), "error sending event acknowledge");
      return (ngcbSHM_ID_ERR);
      }
 
    return (shmId);
    }
      else
    {
    strcpy(ErrMsg(), "event dispatcher is not running");
    return (ngcbSHM_ID_ERR);
    }
      }
 
    /*
     * Wait for next event and lock event
     */
    shmId = Wait_(noWait);
    if ((shmId >= 0) || (shmId == ngcbSHM_ID_IVLD))
      {
      /*
       * Unlock the event
       */
       EvtUnlock_();
      }
 
    return (shmId);
  }
  
  int TryWait() {return (Wait(1));}

  void *GetBuffer(void *hdr, int noWait = 0) {
    void *buffer = NULL;
    int shmId = ngcbSHM_ID_IVLD;
 
    if (extPipe_)
      {
      if ((fd_ >= 0) && !extPipeErr_)
    {
    if (noWait)
      {
      int res = WaitPipe_(1);
      if (res < 0)
        {
        strcpy(ErrMsg(), "error reading event header");
        return (NULL);
        }
      else if (res == 0)
        {
        strcpy(ErrMsg(), "busy");
        return (NULL);
        }
      }
 
    /*
     * Read shared memory id
     */
    if (read(fd_, &shmId, sizeof(int)) != sizeof(int))
      {
      extPipeErr_ = 1;
      strcpy(ErrMsg(), "error reading event header");
      return (NULL);
      }
 
    if (shmId >= 0)
      {
      shm_ = shmat(shmId, 0, SHM_RDONLY);
      if (shm_ != (void *)-1)
        {
        ngcb_shmio_t *io = static_cast<ngcb_shmio_t *>(shm_);
 
        /*
         * Get header
         */
        memcpy(hdr, shm_, sizeof(ngcb_shmhdr_t));
        evtCnt_ = io->event;
 
        /*
         * Assign buffer
         */
        buffer = (void *)(((char *)shm_) + sizeof(ngcb_shmhdr_t));
        }
      else
        {
        shm_ = NULL;
        sprintf(ErrMsg(), "cannot attach to shared memory - %s", 
            strerror(errno));
        return (NULL);
        }
      }
    else
      {
      int ack = 1;
 
      /*
       * Send acknowledge
       */
      if (write(fd_, &ack, sizeof(int)) != sizeof(int))
        {
        extPipeErr_ = 1;
        strcpy(ErrMsg(), "error sending event acknowledge");
        return (NULL);
        }
      }
 
    return (buffer);
    }
      else
    {
    strcpy(ErrMsg(), "event dispatcher is not running");
    return (NULL);
    }
      }
 
    /*
     * Wait for next event and lock event
     */
    shmId = Wait_(noWait);
    if (shmId >= 0)
      {    
      /*
       * Attach buffer and unlock the event
       */
      buffer = AttachBuffer(shmId, hdr, 1);
      EvtUnlock_();
      }
    else if (shmId == ngcbSHM_ID_RETRY)
      {
      strcpy(ErrMsg(), "busy");
      }
    else if (shmId == ngcbSHM_ID_IVLD)
      {
      /*
       * Unlock the event
       */
      EvtUnlock_();
      }
 
    return (buffer);
  }

  void *TryGetBuffer(void *hdr) {return (GetBuffer(hdr, 1));}

  int GetHdr(void *hdr, int noWait = 0) {
    int shmId = ngcbSHM_ID_IVLD;
 
    if (extPipe_)
      {
      if ((fd_ >= 0) && !extPipeErr_)
    {
    if (noWait)
      {
      int res = WaitPipe_(1);
      if (res < 0)
        {
        strcpy(ErrMsg(), "error reading event header");
        return (ngcbSHM_ID_ERR);
        }
      else if (res == 0)
        {
        return (ngcbSHM_ID_RETRY);
        }
      }
 
    /*
     * Read shared memory id
     */
    if (read(fd_, &shmId, sizeof(int)) != sizeof(int))
      {
      extPipeErr_ = 1;
      strcpy(ErrMsg(), "error reading event header");
      return (ngcbSHM_ID_ERR);
      }
 
    if (shmId >= 0)
      {
      shm_ = shmat(shmId, 0, SHM_RDONLY);
      if (shm_ != (void *)-1)
        {
        ngcb_shmio_t *io = static_cast<ngcb_shmio_t *>(shm_);
 
        /*
         * Get header
         */
        memcpy(hdr, shm_, sizeof(ngcb_shmhdr_t));
        evtCnt_ = io->event;
        shmdt(shm_); shm_ = NULL;
        }
      else
        {
        shm_ = NULL;
        sprintf(ErrMsg(), "cannot attach to shared memory - %s", 
            strerror(errno));
        return (ngcbSHM_ID_ERR);
        }
      }
    else
      {
      int ack = 1;
 
      /*
       * Send acknowledge
       */
      if (write(fd_, &ack, sizeof(int)) != sizeof(int))
        {
        extPipeErr_ = 1;
        strcpy(ErrMsg(), "error sending event acknowledge");
        return (ngcbSHM_ID_ERR);
        }
      }
 
    return (shmId);
    }
      else
    {
    strcpy(ErrMsg(), "event dispatcher is not running");
    return (ngcbSHM_ID_ERR);
    }
      }
    
    /*
     * Wait for next event and lock event
     */
    shmId = Wait_(noWait);
    if (shmId >= 0)
      {
      /*
       * Attach buffer
       */
      if (AttachBuffer(shmId, hdr, 1) != NULL)
    {
    /*
     * The header is filled - we detach again
     */
    if (shm_ != NULL)
      {
      shmdt(shm_); shm_ = NULL;
      }
    }
      else
    {
    ngcb_shmio_t *io = static_cast<ngcb_shmio_t *>(hdr);
    if (io->retry)
      {
      /*
       * We just declare it as invalid
       */ 
      shmId = ngcbSHM_ID_IVLD;
      }
    else
      {
      /*
       * This was a serious error
       */ 
      shmId = ngcbSHM_ID_ERR;
      }
    }
      
      /*
       * Unlock the event
       */
      EvtUnlock_();
      }
    else if (shmId == ngcbSHM_ID_IVLD)
      {
      /*
       * Unlock the event
       */
      EvtUnlock_();
      }
 
    return (shmId);
  }

  int TryGetHdr(void *hdr) {return (GetHdr(hdr, 1));}

  int Send(int shmId) {
    struct sembuf semOp[2];
    int i;
 
    /*
     * Next event id
     */
    if (++evtCnt_ == 0) evtCnt_ = 1;
 
    /*
     * Write share memory id
     */
    if (EvtLock_() != 0)
      {
      return (-1);
      }
    evt_->shmId = shmId;
    evt_->event = evtCnt_;
    EvtUnlock_();
 
    /*
     * Initialize semaphore operation
     */
    memset(&semOp[0], 0, sizeof(struct sembuf));
    memset(&semOp[1], 0, sizeof(struct sembuf));
    semOp[0].sem_op = 0;
    semOp[0].sem_flg = IPC_NOWAIT;
    semOp[1].sem_op = 1;
    semOp[1].sem_flg = 0;
    
    /*
     * Send event to all clients
     */
    for (i=0;i<ngcbSHM_MAX_CLIENTS;i++)
      {
      semOp[0].sem_num = static_cast<unsigned short>(2 + i);
      semOp[1].sem_num = static_cast<unsigned short>(2 + i);
      if (semop(semEvt_, &semOp[0], 2) != 0)
    {
    if (errno == EAGAIN)
      {
      /*
       * Client was not ready
       */
      continue;
      }
    else if (errno == EIDRM)
      {
      /*
       * Event semaphore has been removed
       */
      semEvt_ = -1;
      }
    sprintf(ErrMsg(), "semaphore error - %s\n", strerror(errno));
    return (-1);
    }
      }
 
    return (0);
  }

  int Release() {
    struct sembuf semOp[2];
 
    if (semEvt_ < 0)
      {
      strcpy(ErrMsg(), "event interface not initialized");
      return (-1);
      }
    else if (clientId_ < 0)
      {
      strcpy(ErrMsg(), "not attached to events");
      return (-1);
      }
 
    if (EvtLock_() == 0)
      {
      cancelled_ = true;
      EvtUnlock_();
      }
    else
      {
      cancelled_ = true;
      }
 
    /*
     * Initialize semaphore operation
     */
    memset(&semOp[0], 0, sizeof(struct sembuf));
    memset(&semOp[1], 0, sizeof(struct sembuf));
    semOp[0].sem_op = 0;
    semOp[0].sem_flg = IPC_NOWAIT;
    semOp[0].sem_num = static_cast<unsigned short>(2 + clientId_);
    semOp[1].sem_op = 1;
    semOp[1].sem_flg = 0;
    semOp[1].sem_num = static_cast<unsigned short>(2 + clientId_);
 
    /*
     * Send event
     */
    if (semop(semEvt_, &semOp[0], 2) != 0)
      {
      if (errno == EAGAIN)
    {
    /*
     * Event has already been sent but client was not ready
     */
    return (0);
    }
      else if (errno == EIDRM)
    {
    /*
     * Event semaphore has been removed
     */
    semEvt_ = -1;
    }
      sprintf(ErrMsg(), "semaphore error - %s\n", strerror(errno));
      return (-1);
      }
 
    return (0);
  }

  void Invalidate() {
    if (EvtLock_() == 0)
      {
      evt_->shmId = ngcbSHM_ID_ERR;
      evt_->event = 0;
      EvtUnlock_();
      }
  }

  int RecvEvt() {
    int shmId = ngcbSHM_ID_IVLD;
    if (intPipe_ && (fd_ >= 0)) 
      {
      if (read(fd_, &shmId, sizeof(int)) != sizeof(int))
    {
    strcpy(ErrMsg(), "error receiving immage event from pipe");
    return (ngcbSHM_ID_ERR);
    }      
      }
    return (shmId);
  }

  int SendAck() {
    if (intPipe_ && (fd_ >= 0)) 
      {
      int ack = 1;
      if (write(fd_, &ack, sizeof(int)) != sizeof(int))
    {
    strcpy(ErrMsg(), "error sending event acknowledge to pipe");
    return (-1);
    }
      }
    return (0);
  }

  void Thread() {
    int fd = pipe_;
 
    if ((fd >= 0) && (sharedHeader != NULL))
      {
      int ack = 1;
      int shmId = ngcbSHM_ID_IVLD;
 
      /*
       * Thread main-loop
       */
      while (ack)
    {
    /*
     * Get next event header and acquire read-lock
     */
    shmId = GetHdr(sharedHeader);
    if ((shmId == ngcbSHM_ID_ERR) || (shmId == ngcbSHM_ID_RETRY))
      {
      /*
       * An error occured or the event has been cancelled due to
       * an interface shutdown
       */
      break;
      }
    
    if ((shmId >= 0) || (shmId == ngcbSHM_ID_IVLD))
      {
      if (!sharedHeader->io.retry && (EvtCnt() != 0))
        {
        /*
         * Send shared memory id
         */
        if (write(fd, &shmId, sizeof(int)) != sizeof(int))
          {
          break;
          }
        
        /*
         * Wait for acknowledge
         */
        if (read(fd, &ack, sizeof(int)) != sizeof(int))
          {
          break;
          }
        }
      
      if (shmId >= 0)
        {
        /*
         * Release read-lock
         */
        RdUnlock(sharedHeader->io.semId);  
        shmId = ngcbSHM_ID_IVLD;
        }
      else
        {
        /*
         * Check, if our event pipe is still open
         */
        if (!pipeOpen_)
          {
          break;
          }
        }
      }
    }
      
      if (shmId >= 0)
    {
    /*
     * Release read-lock
     */
    RdUnlock(sharedHeader->io.semId);  
    }
      }
  }
 
protected: 
 
private:
  char erms_[256];

  int initialized_;         

  void *shm_;

  struct sembuf semOpWrLock_[3];

  struct sembuf semOpWrTryLock_[3];

  struct sembuf semOpWrUnlock_[1];

  struct sembuf semOpRdWait_[1];

  struct sembuf semOpRdLock_[2];

  struct sembuf semOpRdUnlock_[1];

  struct sembuf semOpEvtLock_[2];

  struct sembuf semOpEvtTryLock_[2];

  struct sembuf semOpEvtUnlock_[1];

  int regCnt_;

  int evtId_;

  ngcb_shmevt_t *evt_;

  int semEvt_;

  unsigned int evtCnt_;

  int clientId_;

  int fd_;

  int intPipe_;

  int pipe_;

  int pipeOpen_;

  int extPipe_;

  int extPipeErr_;

  int pid_;
 
#ifdef NGCB_SHM_THR
  pthread_t thrId_;
#endif

  int thr_;
  
  ngcb_shmhdr_t sharedHeader_;

  std::atomic_bool cancelled_;

  void Init_() {
    initialized_ = 0;
    name[0] = '\0';
    sharedHeader = &sharedHeader_; // link to our private resource
    erms_[0] = '\0';
    semEvt_ = -1;
    evtId_ = -1;
    evt_ = NULL;
    evtCnt_ = 0;
    shm_ = NULL;
    clientId_ = -1;
    fd_ = -1; 
    intPipe_ = 0; pipe_ = -1; pipeOpen_ = 0; thr_ = 0;
    extPipe_ = 0; extPipeErr_ = 0; pid_ = 0;
    cancelled_ = false;
    regCnt_ = 0; 
 
#ifdef NGCB_SHM_THR
    /*
     * Initialize thread id
     */
    thrId_ = pthread_self();
#endif
 
    /*
     * Semaphore operation: 
     * - wait until there are no more readers
     * - wait until there is no writer
     * - increment writer by one
     */
    memset(&semOpWrLock_[0], 0, sizeof(struct sembuf));
    memset(&semOpWrLock_[1], 0, sizeof(struct sembuf));
    memset(&semOpWrLock_[2], 0, sizeof(struct sembuf));
    semOpWrLock_[0].sem_num = 1;
    semOpWrLock_[0].sem_op = 0;  
    semOpWrLock_[0].sem_flg = 0;
    semOpWrLock_[1].sem_num = 0;
    semOpWrLock_[1].sem_op = 0; 
    semOpWrLock_[1].sem_flg = 0;
    semOpWrLock_[2].sem_num = 0;
    semOpWrLock_[2].sem_op = 1; 
    semOpWrLock_[2].sem_flg = SEM_UNDO;
    
    /*
     * Semaphore operation: same as above with IPC_NOWAIT
     */
    memset(&semOpWrTryLock_[0], 0, sizeof(struct sembuf));
    memset(&semOpWrTryLock_[1], 0, sizeof(struct sembuf));
    memset(&semOpWrTryLock_[2], 0, sizeof(struct sembuf));
    semOpWrTryLock_[0].sem_num = 1;
    semOpWrTryLock_[0].sem_op = 0;  
    semOpWrTryLock_[0].sem_flg = IPC_NOWAIT;
    semOpWrTryLock_[1].sem_num = 0;
    semOpWrTryLock_[1].sem_op = 0; 
    semOpWrTryLock_[1].sem_flg = IPC_NOWAIT;
    semOpWrTryLock_[2].sem_num = 0;
    semOpWrTryLock_[2].sem_op = 1; 
    semOpWrTryLock_[2].sem_flg = SEM_UNDO;
    
    /*
     * Semaphore operation: 
     * - check, if we own the write-lock - if not return error
     * - release write lock (decrement by one)
     */
    memset(&semOpWrUnlock_[0], 0, sizeof(struct sembuf));
    semOpWrUnlock_[0].sem_num = 0;
    semOpWrUnlock_[0].sem_op = -1;
    semOpWrUnlock_[0].sem_flg = IPC_NOWAIT|SEM_UNDO;
    
    /*
     * Semaphore operation: 
     * - wait until there are no more readers
     */
    memset(&semOpRdWait_[0], 0, sizeof(struct sembuf));
    semOpRdWait_[0].sem_num = 1;
 
    /*
     * Semaphore operation: 
     * - wait until there is no writer
     * - increment readers by one
     */
    memset(&semOpRdLock_[0], 0, sizeof(struct sembuf));
    memset(&semOpRdLock_[1], 0, sizeof(struct sembuf));
    semOpRdLock_[0].sem_num = 0;
    semOpRdLock_[0].sem_op = 0;
    semOpRdLock_[0].sem_flg = 0;
    semOpRdLock_[1].sem_num = 1;
    semOpRdLock_[1].sem_op = 1;
    semOpRdLock_[1].sem_flg = SEM_UNDO;
 
    /*
     * Semaphore operation: 
     * - check, if there is at least one read-lock - if not, return error
     * - decrement readers by one
     */
    memset(&semOpRdUnlock_[0], 0, sizeof(struct sembuf));
    semOpRdUnlock_[0].sem_num = 1;
    semOpRdUnlock_[0].sem_op = -1;
    semOpRdUnlock_[0].sem_flg = IPC_NOWAIT|SEM_UNDO;
 
    /*
     * Semaphore operation: 
     * - lock event access semaphore
     */
    memset(&semOpEvtLock_[0], 0, sizeof(struct sembuf));
    memset(&semOpEvtLock_[1], 0, sizeof(struct sembuf));
    semOpEvtLock_[1].sem_num = 0;
    semOpEvtLock_[1].sem_op = 1;
    semOpEvtLock_[1].sem_flg = SEM_UNDO;
 
    /*
     * Semaphore operation: 
     * - try to lock event access semaphore
     */
    memset(&semOpEvtTryLock_[0], 0, sizeof(struct sembuf));
    memset(&semOpEvtTryLock_[1], 0, sizeof(struct sembuf));
    semOpEvtTryLock_[0].sem_num = 0;
    semOpEvtTryLock_[0].sem_op = 0;
    semOpEvtTryLock_[0].sem_flg = IPC_NOWAIT;
    semOpEvtTryLock_[1].sem_num = 0;
    semOpEvtTryLock_[1].sem_op = 1;
    semOpEvtTryLock_[1].sem_flg = SEM_UNDO;
 
    /*
     * Semaphore operation: 
     * - unlock event access semaphore
     */
    memset(&semOpEvtUnlock_[0], 0, sizeof(struct sembuf));
    semOpEvtUnlock_[0].sem_num = 0;
    semOpEvtUnlock_[0].sem_op = -1;
    semOpEvtUnlock_[0].sem_flg = IPC_NOWAIT|SEM_UNDO;
  }

  int Register_() {
    int newShm = 0;
    void *m;
    key_t key = (key_t)-1;
 
    /*
     * Ensure that we are not yet bound to any resources
     */
    if (evt_ != NULL)
      {
      shmdt(evt_); evt_ = NULL;
      }
 
    /*
     * Get IPC-key
     */
    if (strlen(name) > 0)
      {
      char fileName[256];
      int fd;
      snprintf(fileName, sizeof(fileName), "/tmp/ngcshmio_%u_%s", getuid(), 
           name);
      fd = open(fileName, O_RDWR|O_CREAT|O_TRUNC, 0644);
      if (fd >= 0)
    {
    close(fd);
    key = ftok(fileName, 0x1f);
    }
      }
    else
      {
      char *evar = getenv("INS_ROOT");
      if (evar != NULL) key = ftok(evar, 0x1f);
      if (key == -1)
    {
    evar = getenv("INTROOT");
    if (evar != NULL) key = ftok(evar, 0x1f);
    }
      if (key == -1)
    {
    evar = getenv("HOME");
    if (evar != NULL) key = ftok(evar, 0x1f);
    }
      }
 
    if (key == (key_t)-1)
      {
      strcpy(ErrMsg(), "unable to retrieve IPC-key");
      return (-1);
      }
 
    /*
     * Try to access the shared memory event
     */
    evtId_ = shmget(key, sizeof(ngcb_shmevt_t), 0666);
    if (evtId_ == -1)
      {
      int tmp;
 
      if (errno == EINVAL)
    {
    /*
     * The shared memory size may have changed. We remove the
     * exising region and re-create.
     */ 
    evtId_ = shmget(key, 0, 0666);
    if (evtId_ == -1)
      {
      sprintf(ErrMsg(), "error creating shared memory - %s", 
          strerror(errno));
      return (-1);
      }
      
    /*
     * Remove event id and re-create
     */
    shmctl(evtId_, IPC_RMID, 0); evtId_ = -1;
    }
      
      /*
       * Ensure that the event semaphore does not yet exist
       */
      tmp = semget(key, 0, 0666);
      if (tmp >= 0)
    {
    ngcbSEM_CTL ctl;
    
    /*
     * Remove the semaphore
     */
    ctl.val = 0;
    semctl(tmp, 0, IPC_RMID, ctl);    
    }
 
      /*
       * Create a new shared memory region
       */
      evtId_ = shmget(key, sizeof(ngcb_shmevt_t), 0666|IPC_CREAT|IPC_EXCL);
      if (evtId_ == -1)
    {
    if (errno == EEXIST)
      {
      /*
       * We may have had a race condition here. So we try once
       * again to access it.
       */
      evtId_ = shmget(key, sizeof(ngcb_shmevt_t), 0666);
      if (evtId_ == -1)
        {
        if ((errno == ENOENT) && (regCnt_ > 0))
          {
          /*
           * Retry
           */
          regCnt_--;
          return(Register_());
          }
        sprintf(ErrMsg(), "error creating shared memory - %s", 
            strerror(errno));
        return (-1);
        }
      }
    else
      {
      sprintf(ErrMsg(), "error creating shared memory - %s", 
          strerror(errno));
      return (-1);
      }
    }
      else
    {
    /*
     * We actually are the creator
     */
    newShm = 1;
    }
      }
 
    /*
     * Attach to the shared memory event
     */
    m = shmat(evtId_, 0, 0);
    if (m == (void *)-1)
      {
      int e = errno;
 
      if (newShm)
    {
    /*
     * Delete the shared memory event again
     */ 
    shmctl(evtId_, IPC_RMID, 0); evtId_ = -1;
    }
 
      if ((e == EINVAL) && (regCnt_ > 0))
    {
    /*
     * Retry
     */
    regCnt_--;
    return (Register_());
    }
      else
    {
    sprintf(ErrMsg(), "cannot attach to shared memory - %s", strerror(e));
    return (-1);
    }
      }
    else
      {
      evt_ = static_cast<ngcb_shmevt_t *>(m);
      }
 
    if (newShm)
      {
      ngcbSEM_CTL ctl;
      ngcb_shmevt_t initStat;
      unsigned short semArray[2 + ngcbSHM_MAX_CLIENTS];
      int i;
 
      /*
       * Initialize the event. This will explicitly mark
       * the event semaphore as invalid.
       */
      memcpy(evt_, &initStat, sizeof(ngcb_shmevt_t));
      semArray[0] = 0;
      semArray[1] = 0;
      for (i=0;i<ngcbSHM_MAX_CLIENTS;i++) semArray[2 + i] = 2;
      
      /*
       * Create the event semaphore
       */
      semEvt_ = semget(key, 2 + ngcbSHM_MAX_CLIENTS, 0666|IPC_CREAT|IPC_EXCL);
      if (semEvt_ == -1)
    {
    sprintf(ErrMsg(), "error creating semaphore - %s", strerror(errno));
    
    /*
     * Delete the shared memory event
     */ 
    shmdt(evt_); evt_ = NULL;
    shmctl(evtId_, IPC_RMID, 0); evtId_ = -1;
    return (-1);
    }
      
      /*
       * Initialize the event semaphore
       */
      ctl.array = semArray;
      if (semctl(semEvt_, 0, SETALL, ctl) == -1)
    {
    int e = errno;
 
    sprintf(ErrMsg(), "semaphore error - %s", strerror(e));
 
    if (e == EIDRM)
      {
      /*
       * Semaphore has been removed
       */
      semEvt_ = -1;
      }
    else
      {
      /*
       * Delete the event semaphore again
       */ 
      SemEvtRemove_();
      }
 
    /*
     * Delete the shared memory event
     */ 
    shmdt(evt_); evt_ = NULL;
    shmctl(evtId_, IPC_RMID, 0); evtId_ = -1;
 
    if ((e == EIDRM) && (regCnt_ > 0))
      {
      /*
       * Retry
       */
      regCnt_--;
      return (Register_());
      }
    else
      {
      return (-1);
      }
    }
 
      /*
       * Assign the event semaphore
       */
      if (EvtLock_() != 0)
    {
    /*
     * Delete the event semaphore again
     */ 
    SemEvtRemove_();
 
    /*
     * Delete the shared memory event
     */ 
    shmdt(evt_); evt_ = NULL;
    shmctl(evtId_, IPC_RMID, 0); evtId_ = -1;
    if (regCnt_ > 0)
      {
      /*
       * Retry
       */
      regCnt_--;
      return (Register_());
      }
    else
      {
      return (-1);
      }
    }
      evt_->semId = semEvt_;
      EvtUnlock_();
      }
    else
      {
      /*
       * Get event semaphore - this should exist as the associated
       * shared memory has already been created
       */
      semEvt_ = semget(key, 0, 0666);
      if (semEvt_ == -1)
    {
    if ((errno == ENOENT) && (regCnt_ > 0))
      {
      /*
       * We may have had a race condition with the semaphore 
       * creator. Just retry.
       */
      shmdt(evt_); evt_ = NULL;
      regCnt_--;
      return (Register_());
      }
    else
      {
      /*
       * Any other error causes the registration to fail
       */
      sprintf(ErrMsg(), "semaphore error - %s", strerror(errno));
      return (-1);
      }
    }
      
      /*
       * Verify semaphore - we cannot lock here due to a race condition
       * with the creators semctl. This should not matter because only 
       * a non-faulty matching read indicates that the semaphore is 
       * both created and initialized.
       */
      if (semEvt_ != evt_->semId)
    {
    if (regCnt_ > 0)
      {
      /*
       * Retry
       */
      semEvt_ = -1;
      shmdt(evt_); evt_ = NULL;
      regCnt_--;
      return (Register_());
      }
    else
      {
      strcpy(ErrMsg(), "semaphore mismatch");
      return (-1);
      }
    }
      
      /*
       * Align event counter
       */
      evtCnt_ = evt_->event;
      }
 
    /*
     * Register to events
     */
    if (semEvt_ >= 0)
      {
      struct sembuf semOp[4];
 
      memset(&semOp[0], 0, sizeof(struct sembuf));
      memset(&semOp[1], 0, sizeof(struct sembuf));
      memset(&semOp[2], 0, sizeof(struct sembuf));
      memset(&semOp[3], 0, sizeof(struct sembuf));
 
      semOp[1].sem_num = 0;
      semOp[1].sem_op = 1;
      semOp[1].sem_flg = SEM_UNDO;
 
      semOp[2].sem_num = 1;
      semOp[2].sem_op = 1;
      semOp[2].sem_flg = SEM_UNDO;
 
      semOp[3].sem_num = 0;
      semOp[3].sem_op = -1;
      semOp[3].sem_flg = SEM_UNDO;
      if (semop(semEvt_, semOp, 4) != 0)
    {
    int e = errno;
    
    shmdt(evt_); evt_ = NULL;
    if (e == EIDRM) semEvt_ = -1;  // semaphore has been removed
 
    if (newShm) 
      {
      /*
       * Delete the semaphore again
       */ 
      SemEvtRemove_();
      
      /*
       * Delete the shared memory
       */ 
      shmctl(evtId_, IPC_RMID, 0); evtId_ = -1;
      }
 
    if (regCnt_ > 0)
      {
      /*
       * Retry
       */
      regCnt_--;
      return (Register_());
      }
    else
      {
      sprintf(ErrMsg(), "semaphore error - %s", strerror(e));
      return (-1);
      }
    }
      }
 
    return (0);
  }

  void DetachEvt_() {
    if (!extPipe_ && (semEvt_ >= 0) && (clientId_ >= 0))
      {
      ngcbSEM_CTL ctl;
      ctl.val = 2;
      semctl(semEvt_, 2 + clientId_, SETVAL, ctl);
      }
  }

  int EvtLock_() {
    if (semEvt_ >= 0)
      {
      if (semop(semEvt_, semOpEvtLock_, 2) != 0)
    {
    if (errno == EINTR)
      {
      while (semop(semEvt_, semOpEvtLock_, 2) != 0)
        {
        if (errno != EINTR)
          {
          if (errno == EIDRM) semEvt_ = -1;  // removed (and unlocked)
          sprintf(ErrMsg(), "cannot lock semaphore - %s", strerror(errno));
          return (-1);
          }
        }
      }
    else
      {
      if (errno == EIDRM) semEvt_ = -1;  // removed (and unlocked)
      sprintf(ErrMsg(), "cannot lock semaphore - %s", strerror(errno));
      return (-1);
      }
    }
      }
    else
      {
      strcpy(ErrMsg(), "invalid semaphore");
      return (-1);
      }
 
    return (0);
  }

  int EvtTryLock_() {
    if (semEvt_ >= 0)
      {
      if (semop(semEvt_, semOpEvtTryLock_, 2) == 0)
    {
    return (0);
    }
      else
    {
    if (errno == EIDRM) semEvt_ = -1;  // removed (and unlocked)
    sprintf(ErrMsg(), "cannot lock semaphore - %s", strerror(errno));
    return (-1);
    }
      }
 
    strcpy(ErrMsg(), "invalid semaphore");
    return (-1);
  }

  void EvtUnlock_() {
    if (semEvt_ >= 0)
      {
      if (semop(semEvt_, semOpEvtUnlock_, 1) != 0)
    {
    if (errno == EIDRM) semEvt_ = -1;  // removed (and unlocked)
    }
      }
  }

  void SemEvtRemove_() {
    if (semEvt_ >= 0)
      {
      ngcbSEM_CTL ctl;
      ctl.val = 0;
      semctl(semEvt_, 0, IPC_RMID, ctl); semEvt_ = -1;
      }
  }

  int RdLock_(int semId, int noWait = 0) {
    if (semId >= 0)
      {
      if (noWait) semOpRdLock_[0].sem_flg = IPC_NOWAIT;
      else semOpRdLock_[0].sem_flg = 0;
 
      if (semop(semId, semOpRdLock_, 2) != 0)
    {
    if (errno == EINTR)
      {
      while (semop(semId, semOpRdLock_, 2) != 0)
        {
        if (errno != EINTR) return (-1);
        }
      }
    else
      {
      return (-1);
      }
    }
      }
 
    return (0);
  }

  int RdUnlock_(int semId) {
    if (semId >= 0)
      {
      if (semop(semId, semOpRdUnlock_, 1) != 0)
    {
    return (-1);
    }
      }
    return (0);
  }

  int Wait_(int noWait) {
    struct sembuf semOp[3];
    int shmId = ngcbSHM_ID_ERR;
    int valid = 0;
 
    if (semEvt_ < 0)
      {
      strcpy(ErrMsg(), "event interface not initialized");
      return (ngcbSHM_ID_ERR);
      }
    else if (clientId_ < 0)
      {
      strcpy(ErrMsg(), "not attached to events");
      return (ngcbSHM_ID_ERR);
      }
 
    while (!valid)
      {
      /*
       * Wait for event and then lock event structure
       */
      memset(&semOp[0], 0, sizeof(struct sembuf));
      memset(&semOp[1], 0, sizeof(struct sembuf));
      memset(&semOp[2], 0, sizeof(struct sembuf));
      semOp[0].sem_num = static_cast<unsigned short>(2 + clientId_);
      semOp[0].sem_op = -1;
      semOp[2].sem_num = 0;
      semOp[2].sem_op = 1;
      semOp[2].sem_flg = SEM_UNDO;
      if (noWait)
    {
    semOp[0].sem_flg = IPC_NOWAIT;
    semOp[1].sem_flg = IPC_NOWAIT;
    }
      if (semop(semEvt_, semOp, 3) != 0)
    {
    if (errno == EAGAIN)
      {
      if (noWait) return (ngcbSHM_ID_RETRY);
      else continue;
      }
    else if (errno == EINTR)
      {
      continue;
      }
    else if (errno == EIDRM)
      {
      /*
       * Event semaphore has been removed
       */
      semEvt_ = -1;
      }
 
    sprintf(ErrMsg(), "semaphore error - %s", strerror(errno));
    return (ngcbSHM_ID_ERR);
    }
      
      /*
       * Check event
       */
      if (((evt_->event == evtCnt_) || (evt_->event == 0)) || cancelled_)
    {
    /*
     * We still store the event counter
     */
    evtCnt_ = evt_->event;
    if (noWait || cancelled_) shmId = ngcbSHM_ID_RETRY;
    cancelled_ = false;
    EvtUnlock_();
 
    if (shmId == ngcbSHM_ID_RETRY)
      {
      /*
       * No valid event
       */
      return (shmId);
      }
    else
      {
      /*
       * Wait for next event
       */
      continue;
      }
    }
      else
    {
    /*
     * Store event. We do not unlock here - the public
     * function will handle this
     */
    shmId = evt_->shmId;
    evtCnt_ = evt_->event;
    if (shmId < 0) shmId = ngcbSHM_ID_IVLD;
    valid = 1;
    }
      }
 
    return (shmId);
  }

  int OpenPipe_() {
    int fd[2];
    int ack;
    int fdFlags;
    char arg[32];
 
    /*
     * Socket pair for internal thread event-communication
     */ 
    if (socketpair(AF_UNIX, SOCK_STREAM, 0, fd) == -1)
      {
      sprintf(ErrMsg(), "error opening event pipe - %s", strerror(errno));
      return(-1);
      }
    
    /*
     * We fork here
     */
    sprintf(arg, "%d", fd[1]);
    pid_ = fork();
    if (pid_ < 0)
      {
      /*
       * Cannot fork
       */
      sprintf(ErrMsg(), "error opening event pipe - %s", strerror(errno));
      close(fd[0]); close(fd[1]);
      return (-1);
      }
    else if (pid_ == 0)
      {
      /*
       * Close the end we don't need
       */
      close(fd[0]);
 
      /*
       * Execute event dispatcher
       */
      execlp("ngcbShmEvt", "ngcbShmEvt", name, arg, (char *)NULL);
     
      /*
       * If we are still here then the execution has failed
       */
      close(fd[1]);
      _exit(127);
      } // child process
    
    /*
     * Close the end we don't need
     */
    close(fd[1]);
 
    /*
     * Set close-on-exec flag
     */
    fdFlags = fcntl(fd[0], F_GETFD);
    if (fdFlags >= 0) fcntl(fd[0], F_SETFD, fdFlags | FD_CLOEXEC);
 
    /*
     * Read acknowledge from the other side
     */
    if (read(fd[0], &ack, sizeof(int)) != sizeof(int))
      {
      strcpy(ErrMsg(), "unable to launch event dispatcher");
      close(fd[0]); waitpid(pid_, NULL, 0); pid_ = 0;
      return (-1);
      }
 
    if (!ack)
      {
      int size;
 
      /*
       * Read back error message
       */
      if (read(fd[0], &size, sizeof(int)) != sizeof(int))
    {
    strcpy(ErrMsg(), "event dispatcher connection broken");
    close(fd[0]); waitpid(pid_, NULL, 0); pid_ = 0;
    return (-1);
    }
      if (size > 0)
    {
    size++;  // including the terminating NULL-character
    if (read(fd[0], ErrMsg(), size) != size)
      {
      strcpy(ErrMsg(), "event dispatcher connection broken");
      }
    }
      else
    {
    strcpy(ErrMsg(), "unable to launch event dispatcher");
    }
      close(fd[0]); waitpid(pid_, NULL, 0); pid_ = 0;
      return (-1);
      }
 
    return (fd[0]);
  }

  void ClosePipe_() {
    if (fd_ >= 0)
      {
      pipeOpen_ = 0; close(fd_); fd_ = -1; 
      }
    if (pid_ > 0)
      {
      kill(pid_, SIGTERM); waitpid(pid_, NULL, 0); pid_ = 0;
      }
#ifdef NGCB_SHM_THR
    if (thr_)
      {
      Release();
      pthread_join(thrId_, NULL);
      thr_ = 0;
      }
#endif
    if (pipe_ >= 0) 
      {
      close(pipe_); pipe_ = -1;
      }
  }

  int WaitPipe_(int noWait=0) {
    struct pollfd fds;
    int res;
 
    if ((fd_ < 0) || extPipeErr_)
      {
      strcpy(ErrMsg(), "event dispatcher not running");
      return (-1);
      }
 
    fds.fd = fd_;
    fds.events = POLLIN;
    fds.revents = 0;
    if (noWait)
      {
      res = poll(&fds, 1, 0);
      }
    else
      {
      res = poll(&fds, 1, -1);
      }
 
    if (res < 0)
      {
      /*
       * An error occured - file descriptore no longer valid
       */
      extPipeErr_ = 1;
      return (-1);
      }
    else if (res == 0)
      {
      /*
       * File descriptor valid but not ready
       */
      return (0);
      }
    else 
      {
      return (1);
      }
  }

  int StartThread_() {
#ifdef NGCB_SHM_THR
    int fd[2];
    int fdFlags;
    pthread_attr_t attr;
    
    /*
     * Socket pair for internal thread event-communication
     */ 
    if (socketpair(AF_UNIX, SOCK_STREAM, 0, fd) == -1)
      {
      sprintf(ErrMsg(), "error opening event pipe - %s", strerror(errno));
      return(-1);
      }
 
    /*
     * Set close-on-exec flag
     */
    fdFlags = fcntl(fd[0], F_GETFD);
    if (fdFlags >= 0) fcntl(fd[0], F_SETFD, fdFlags | FD_CLOEXEC);
    fdFlags = fcntl(fd[1], F_GETFD);
    if (fdFlags >= 0) fcntl(fd[1], F_SETFD, fdFlags | FD_CLOEXEC);
 
    /*
     * Set the posix thread attributes 
     */
    pthread_attr_init(&attr);
    pthread_attr_setinheritsched(&attr, PTHREAD_INHERIT_SCHED);
    pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
    pthread_attr_setstacksize(&attr, 1024 * 1024);
 
    /*
     * Create thread
     */
    pipe_ = fd[1]; pipeOpen_ = 1;
    if (pthread_create(&thrId_, &attr, ngcbShmThread, this) != 0)
      {
      sprintf(ErrMsg(), "thread creation failed - %s", strerror(errno));
      close(fd[0]); close(fd[1]); pipe_ = -1; pipeOpen_ = -1;
      return(-1);
      }
    
    /*
     * We have a valid file descriptor to read events
     */
    fd_ = fd[0];
 
    return (0);
#else
    strcpy(ErrMsg(), "event dispatcher thread not installed");
    return (-1);
#endif
  }
};
 
#endif