ngcdcsMRGSHM.hpp

Go to the documentation of this file.

#ifndef ngcdcsMRGSHM_H
#define ngcdcsMRGSHM_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 <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include <sys/socket.h>
 
#include <ngc/core/ngcb.h>
#include <ngc/core/ngcbSocket.h>
 
/*
 * Merger states
 */
#define ngcdcsMSTAT_IDLE              0  //< idle
#define ngcdcsMSTAT_PENDING           1  //< pending
#define ngcdcsMSTAT_ACTIVE            2  //< active
#define ngcdcsMSTAT_VERIFYING         4  //< verifying
#define ngcdcsMSTAT_MERGING           8  //< merging
#define ngcdcsMSTAT_INVALIDATING     16  //< invalidating
#define ngcdcsMSTAT_COMPL_SUCCESS   128  //< successful completion 
#define ngcdcsMSTAT_COMPL_FAILURE   256  //< failure
#define ngcdcsMSTAT_COMPL_ABORTED   512  //< merging has been aborted

#define ngcdcsMERGE_ACTIVE(status) ((status > ngcdcsMSTAT_IDLE) && \
                    (status < ngcdcsMSTAT_COMPL_SUCCESS))
 

#define ngcdcsMRGSHM_MAXCON 4

struct ngcdcs_mrgevt_t {
  int port;

  ngcdcs_mrgevt_t() {port = 0;}
};

struct ngcdcs_mrgshm_t {
  int semId;

  char expId[64];

  int pid;

  int status;

  char statusName[32];

  int reserved1;

  int abort;

  double vsize;
  
  double tsize;
  
  char newDataFileName[256];
  
  char msg[256];

  ngcdcs_mrgevt_t evt[ngcdcsMRGSHM_MAXCON];

  ngcdcs_mrgshm_t() {
    semId = -1;
    expId[0] = '\0';
    pid = 0;
    status = ngcdcsMSTAT_IDLE;
    strcpy(statusName, "idle");
    reserved1 = 0x0;
    abort = 0;
    vsize = 0.0;
    tsize = 0.0;
    newDataFileName[0] = '\0';
    msg[0] = '\0';
  }
};

class ngcdcsMRGSHM
{
public:
  ngcdcsMRGSHM(int id, int server, int creat = 1) {
    char *evar;
    int newShm = 0;
    void *m;
    int i;
    
    /*
     * Initialize member data
     */
    initialized_ = 0;
    server_ = server;
    strcpy(statStr_, "idle");
    if (id < 0) id_ = 0;
    else id_ = id;
    erms_[0] = '\0';
    msg_[0] = '\0';
    shmId_ = -1;
    shm_ = NULL;
    semId_ = -1;
    key_ = -1;
    port_ = 0;
    locked_ = 0;
 
    /*
     * Semaphore lock operation: 
     * - wait until value is zero
     * - increment value by one (undo)
     */
    memset(&semOpLock_[0], 0, sizeof(struct sembuf));
    memset(&semOpLock_[1], 0, sizeof(struct sembuf));
    semOpLock_[1].sem_op = 1;
    semOpLock_[1].sem_flg = SEM_UNDO;
 
    /*
     * Semaphore unlock operation: 
     * - check if value is one - if not return error
     * - decrement value by one (undo)
     */
    memset(&semOpUnlock_[0], 0, sizeof(struct sembuf));
    semOpUnlock_[0].sem_op = -1;
    semOpUnlock_[0].sem_flg = IPC_NOWAIT|SEM_UNDO;
 
    /*
     * Reset event connection file descriptors
     */
    for (i=0;i<ngcdcsMRGSHM_MAXCON;i++) evtSocket_[i] = -1;
 
    /*
     * Get this host name
     */
    evar = getenv("HOSTNAME");
    if (evar != NULL) str2str(thisHost_, evar, sizeof(thisHost_) - 1);
    else strcpy(thisHost_, "localhost");
 
    /*
     * Get IPC-key
     */
    evar = getenv("HOME");
    if (evar != NULL) key_ = ftok(evar, ('n' + id_) & 0xff);
    if (key_ == -1)
      {
      evar = getenv("INS_ROOT");
      if (evar != NULL) key_ = ftok(evar, ('n' + id_) & 0xff);
      }
    if (key_ == -1)
      {
      evar = getenv("INTROOT");
      if (evar != NULL) key_ = ftok(evar, ('n' + id_) & 0xff);
      }
    if (key_ == -1)
      {
      strcpy(ErrMsg(), "unable to retrieve IPC-key");
      return;
      }
 
    /*
     * Try to access the shared memory
     */
    shmId_ = shmget(key_, sizeof(ngcdcs_mrgshm_t), 0666);
    if (shmId_ == -1)
      {
      if (creat)
    {
    int tmp; 
 
    if (errno == EINVAL)
      {
      /*
       * The shared memory size may have changed. We remove the
       * exising region and re-create.
       */ 
      shmId_ = shmget(key_, 0, 0666);
      if (shmId_ == -1)
        {
        sprintf(ErrMsg(), "error creating shared memory - %s", 
            strerror(errno));
        return;
        }
      else
        {
        /*
         * Remove the shared memory
         */
        shmctl(shmId_, IPC_RMID, 0);
        shmId_ = -1;
        }
      }
 
    /*
     * Ensure that the protection semaphore does not yet exist
     */
    tmp = semget(key_, 0, 0666);
    if (tmp >= 0)
      {
      ngcbSEMIPC_CTL ctl;
 
      /*
       * Remove the semaphore
       */
      ctl.val = 0;
      semctl(tmp, 0, IPC_RMID, ctl);      
      }
 
    /*
     * Create a new shared memory region
     */
    shmId_ = shmget(key_, sizeof(ngcdcs_mrgshm_t), 
            0666|IPC_CREAT|IPC_EXCL);
    if (shmId_ == -1)
      {
      if (errno == EEXIST)
        {
        /*
         * We may have had a race condition here. So we try once
         * again to access it.
         */
        shmId_ = shmget(key_, sizeof(ngcdcs_mrgshm_t), 0666);
        if (shmId_ == -1)
          {
          sprintf(ErrMsg(), "error creating shared memory - %s", 
              strerror(errno));
          return;
          }
        }
      else
        {
        sprintf(ErrMsg(), "error creating shared memory - %s", 
            strerror(errno));
        return;
        }      
      }
    else
      {
      /*
       * We actually are the creator
       */
      newShm = 1;
      }
    }
      else
    {
    /*
     * Resources are not available and we have no permission
     * to create them
     */
    strcpy(ErrMsg(), "resources not available");
    return;
    }
      }
    
    /*
     * Attach to the shared memory
     */
    m = shmat(shmId_, 0, 0);
    if (m == (void *)-1)
      {
      sprintf(ErrMsg(), "error attaching to shared memory - %s", 
          strerror(errno));
      if (newShm)
    {
    /*
     * Delete the shared memory again
     */ 
    shmctl(shmId_, IPC_RMID, 0); shmId_ = -1;
    }
      return;
      }
    else
      {
      shm_ = static_cast<ngcdcs_mrgshm_t *>(m);
      }
 
    if (newShm)
      {
      ngcbSEMIPC_CTL ctl;
      ngcdcs_mrgshm_t initStat;
      
      /*
       * Initialize the shared memory. This will explicitly mark
       * the protection semaphore as invalid.
       */
      memcpy(shm_, &initStat, sizeof(ngcdcs_mrgshm_t));
      if (server_)
    {
    shm_->pid = getpid();
    shm_->abort = 0;
    }
 
      /*
       * Create protection semaphore
       */
      semId_ = semget(key_, 1, 0666|IPC_CREAT|IPC_EXCL);
      if (semId_ == -1)
    {
    sprintf(ErrMsg(), "error creating semaphore - %s", 
        strerror(errno));
    
    /*
     * Delete the shared memory
     */ 
    shmdt(shm_); shm_ = NULL;
    shmctl(shmId_, IPC_RMID, 0); shmId_ = -1;
    return;
    }
      
      /*
       * Initialize the semaphore
       */
      ctl.val = 0;
      if (semctl(semId_, 0, SETVAL, ctl) == -1)
    {
    sprintf(ErrMsg(), "semaphore error - %s", strerror(errno));
    
    /*
     * Delete the semaphore again
     */ 
    SemRemove_();
 
    /*
     * Delete the shared memory
     */ 
    shmdt(shm_); shm_ = NULL;
    shmctl(shmId_, IPC_RMID, 0); shmId_ = -1;
    return;
    }
 
      /*
       * Assign the semaphore
       */
      if (Lock_() != ngcbSUCCESS)
    {
    /*
     * Delete the semaphore again
     */ 
    SemRemove_();
 
    /*
     * Delete the shared memory
     */ 
    shmdt(shm_); shm_ = NULL;
    shmctl(shmId_, IPC_RMID, 0); shmId_ = -1;
    return;
    }
      shm_->semId = semId_;
      Unlock_();
      }
    else
      {
      for (i=0;i<3;i++)
    {
    /*
     * Get semaphore
     */
    semId_ = semget(key_, 0, 0666);
    if (semId_ == -1)
      {
      sprintf(ErrMsg(), "semaphore error - %s", strerror(errno));
      sleep(1); continue;
      }
    
    /*
     * Verify semaphore - we cannot lock here due to a race condition
     * with the creators semctl. This should not matter because we 
     * only continue when the semaphore ids match (i.e. any faulty 
     * read would be equal to "not yet created") and only a non-faulty 
     * matching read indicates that the semaphore is both created and
     * initialized.
     */
    if (semId_ != shm_->semId)
      {
      strcpy(ErrMsg(), "semaphore mismatch");
      semId_ = -1;
      sleep(1); continue;
      }
    else
      {
      /*
       * We have a valid semaphore
       */
      break;
      }
    }
 
      if (semId_ < 0)
    {
    /*
     * We could not access the shared resources
     */
    shmdt(shm_); shm_ = NULL;
    return;
    }
 
      if (server_)
    {
    /*
     * Now we can lock
     */
    if (Lock_() != ngcbSUCCESS)
      {
      shmdt(shm_); shm_ = NULL;
      return;
      }
 
    /*
     * Set process id and reset abort-flag
     */
    shm_->pid = getpid();
    shm_->abort = 0;
 
    Unlock_();
    }
      }
 
    initialized_ = 1;    
  }
  
  virtual ~ngcdcsMRGSHM(){ 
    if ((shmId_ != -1) && (shm_ != NULL))
      {
      int i;
 
      if (server_)
    {
    /*
     * Reset abort-flag and process id
     */
    if (Lock_() == ngcbSUCCESS)
      {
      shm_->pid = 0;
      shm_->abort = 0;
      Unlock_();
      }
 
    /*
     * Broadcast exit event
     */
    BroadcastEvt(0);
    
    /*
     * Shutdown event connections
     */
    for (i=0;i<ngcdcsMRGSHM_MAXCON;i++)
      {
      if (evtSocket_[i] >= 0)
        {
        shutdown(evtSocket_[i], SHUT_WR);
        ngcbTcpClose(evtSocket_[i]);
        evtSocket_[i] = -1;
        }
      }
    }
 
      if (port_ > 0)
    {
    /*
     * Invalidate port
     */
    if (Lock_() == ngcbSUCCESS)
      {
      for (i=0;i<ngcdcsMRGSHM_MAXCON;i++)
        {
        if (shm_->evt[i].port == port_) shm_->evt[i].port = 0;
        }
      Unlock_();
      }
    }
 
      /*
       * Detach from shared memory
       */
      shmdt(shm_); shm_ = NULL;
      }
  } 

  int Initialized() {return (initialized_);}

  char *ErrMsg() {return (erms_);}

  int Id() {return (id_);}

  key_t Key() {return (key_);}

  void Pid(int pid) {
    if (!Initialized()) return;
    else if (Lock_() == ngcbSUCCESS)
      {
      shm_->pid = pid;
      Unlock_();
      }
  }

  int Pid() {
    int pid = 0;
 
    if (Initialized())
      {
      if (Lock_() == ngcbSUCCESS)
    {
    pid = shm_->pid;
    Unlock_();
    }
      }
 
    return (pid);
  }

  void Status(int status) {
    if (!Initialized()) return;
    else if (Lock_() == ngcbSUCCESS)
      {
      shm_->status = status;
      Stat2Str(status, shm_->statusName);
      Unlock_();
      BroadcastEvt(1);
      }
  }

  int Status() {
    int s = ngcdcsMSTAT_IDLE;
 
    if (Initialized())
      {
      if (Lock_() == ngcbSUCCESS)
    {
    s = shm_->status;
    Unlock_();
    }
      }
 
    return (s);
  }

  const char *StatusName() {
    if (Initialized())
      {
      if (Lock_() == ngcbSUCCESS)
    {
    strcpy(statStr_, shm_->statusName);
    Unlock_();
    }
      }
    return (statStr_);
  }

  void Msg(const char *msg) {
    if (!Initialized() || (msg == NULL)) return;
    else if (Lock_() == ngcbSUCCESS)
      {
      strcpy(shm_->msg, msg);
      Unlock_();
      }
  }

  const char *Msg() {
    msg_[0] = '\0';
    if (Initialized())
      {
      if (Lock_() == ngcbSUCCESS)
    {
    strcpy(msg_, shm_->msg);
    Unlock_();
    }
      }
    return (msg_);
  }

  void FileEvent(const char *id, const char *name) {
    if (!Initialized() || (name == NULL)) return;
    else if (Lock_() == ngcbSUCCESS)
      {
      str2str(shm_->expId, id, sizeof(shm_->expId) - 1);
      strcpy(shm_->newDataFileName, name);
      Unlock_();
      BroadcastEvt(2);
      }
  }

  const char *FileEvent() {
    msg_[0] = '\0';
    if (Initialized())
      {
      if (Lock_() == ngcbSUCCESS)
    {
    strcpy(msg_, shm_->newDataFileName);
    Unlock_();
    }
      }
    return (msg_);
  }

  void Abort() {
    if (!Initialized()) return;
    else if (Lock_() == ngcbSUCCESS)
      {
      shm_->abort = 1;
      Unlock_();
      }
  }
  
  int Aborted() {
    int flag = 0;
    if (Initialized())
      {
      if (Lock_() == ngcbSUCCESS)
    {
    flag = shm_->abort;
    Unlock_();
    }
      }
    return (flag);
  }

  void VSize(double s) {
    if (!Initialized()) return;
    else if (Lock_() == ngcbSUCCESS)
      {
      shm_->vsize = s;
      Unlock_();
      BroadcastEvt(3);
      }
  }

  double VSize() {
    double s = 0.0;
 
    if (Initialized())
      {
      if (Lock_() == ngcbSUCCESS)
    {
    s = shm_->vsize;
    Unlock_();
    }
      }
 
    return (s);
  }

  void TSize(double s) {
    if (!Initialized()) return;
    else if (Lock_() == ngcbSUCCESS)
      {
      shm_->tsize = s;
      Unlock_();
      BroadcastEvt(4);
      }
  }

  double TSize() {
    double s = 0.0;
 
    if (Initialized())
      {
      if (Lock_() == ngcbSUCCESS)
    {
    s = shm_->tsize;
    Unlock_();
    }
      }
 
    return (s);
  }

  int AttachEvt(int port) {
    int i;
 
    if (!Initialized())
      {
      strcpy(ErrMsg(), "shared memory interface not initialized");
      return (-1);
      }
    
    /*
     * Search free event connection
     */
    if (Lock_() != ngcbSUCCESS)
      {
      return (-1);
      }
    for (i=0;i<ngcdcsMRGSHM_MAXCON;i++)
      {
      if (shm_->evt[i].port <= 0)
    {
    if (evtSocket_[i] >= 0)
      {
      ngcbTcpClose(evtSocket_[i]); evtSocket_[i] = -1;
      }
 
    /*
     * Set port number
     */
    shm_->evt[i].port = port;
    break;
    }
      }
    Unlock_();
    
    if (i == ngcdcsMRGSHM_MAXCON)
      {
      /*
       * No free connection
       */
      strcpy(ErrMsg(), "no free event connection");
      return (-1);
      }
    
    port_ = port;
    return (0);
  }
 
  // Open event
  int OpenEvt() {
    int srvPort = 0;
    int srvSocket;
    
    /*
     * Open a server socket - let the system choose a free port
     */
    srvSocket = ngcbTcpService(NULL, ErrMsg());
    if (srvSocket < 0)
      {
      return (-1);;
      }
    
    /*
     * Get actual port number
     */
    srvPort = ngcbSocketPort(srvSocket);
    if (srvPort <= 0)
      {
      sprintf(ErrMsg(), "unable to resolve server port number");
      ngcbTcpClose(srvSocket);
      return (-1);
      }
 
    /*
     * Register port number
     */
    if (AttachEvt(srvPort) < 0)
      {
      ngcbTcpClose(srvSocket);
      return (-1);
      }
  
    return (srvSocket);
  }

  void BroadcastEvt(int evt = 0) {
    if (Initialized() && server_)
      {
      int i;
      char msg[512];
      int msgLen;
 
      msg[0] = '\0';
 
      if (Lock_() != ngcbSUCCESS)
    {
    return;
    }
 
      /*
       * Check if we have to establish a new event connection
       */
      for (i=0;i<ngcdcsMRGSHM_MAXCON;i++)
    {
    if ((shm_->evt[i].port <= 0) && (evtSocket_[i] >= 0))
      {
      ngcbTcpClose(evtSocket_[i]); evtSocket_[i] = -1;
      }
    if ((shm_->evt[i].port > 0) && (evtSocket_[i] < 0))
      {
      int sock;
      char srv[256];
      char erms[256];
      
      /*
       * Open connection
       */
      sprintf(srv, "%s,%d", thisHost_, shm_->evt[i].port);
      sock = ngcbTcpConnect(srv, erms);
      if (sock < 0) shm_->evt[i].port = 0;
      else evtSocket_[i] = sock;
      }
    }
 
      /*
       * Compose event message
       */
      if (evt == 0)
    {
    /*
     * Exit event
     */
    sprintf(msg, "!mrgexit: %d\n", id_ + 1);
    }
      else if (evt == 1)
    {
    /*
     * Status event
     */
    sprintf(msg, "!mrgstat: %d %d %s\n", id_ + 1, shm_->status, 
        shm_->statusName);
    }
      else if (evt == 2)
    {
    /*
     * File event
     */
    sprintf(msg, "!mrgfile: %d %s %s\n", id_ + 1, shm_->expId,
        shm_->newDataFileName);
    }
      else if (evt == 3)
    {
    /*
     * Size event (expected after verification)
     */
    sprintf(msg, "!mrgvx: %d %.0f\n", id_ + 1, shm_->vsize);
    }
      else if (evt == 4)
    {
    /*
     * Size event (current)
     */
    sprintf(msg, "!mrgtx: %d %.0f\n", id_ + 1, shm_->tsize);
    }
      else
    {
    /*
     * Unknown event - we do not broadcast this
     */
    Unlock_();
    return;
    }
      
      Unlock_();
 
      msgLen = (int)strlen(msg);
      if (msgLen > 0)
    {
    /*
     * Now send the event message to all established connections
     */
    for (i=0;i<ngcdcsMRGSHM_MAXCON;i++)
      {
      if (evtSocket_[i] >= 0)
        {
        char erms[256];
        if (ngcbTcpWrite(evtSocket_[i], &msgLen, sizeof(int), erms) != 0)
          {
          /*
           * Shutdown event connection
           */
          ngcbTcpClose(evtSocket_[i]); evtSocket_[i] = -1;
          }
        else if (ngcbTcpWrite(evtSocket_[i], msg, msgLen, erms) != 0)
          {
          /*
           * Shutdown event connection
           */
          ngcbTcpClose(evtSocket_[i]); evtSocket_[i] = -1;
          }
        }
      }
    }
      }
  }

  void Stat2Str(int stat, char *s) {
    /*
     * Convert status to string
     */
    switch (stat)
      {
      case ngcdcsMSTAT_IDLE:
    strcpy(s, "idle");
    break;
      case ngcdcsMSTAT_ACTIVE:
    strcpy(s, "active");
    break;
      case ngcdcsMSTAT_VERIFYING:
    strcpy(s, "verifying");
    break;
      case ngcdcsMSTAT_MERGING:
    strcpy(s, "merging");
    break;
      case ngcdcsMSTAT_INVALIDATING:
    strcpy(s, "invalidating");
    break;
      case ngcdcsMSTAT_COMPL_SUCCESS:
    strcpy(s, "success");
    break;
      case ngcdcsMSTAT_COMPL_ABORTED:
    strcpy(s, "aborted");
    break;
      case ngcdcsMSTAT_COMPL_FAILURE:
    strcpy(s, "failure");
    break;
      default:
    strcpy(s, "unknown");
    break;
      }
  }

  void Remove() {
    /*
     * Remove semaphore
     */
    SemRemove_();
 
    if ((shmId_ != -1) && (shm_ != NULL))
      {
      /*
       * Delete the shared memory
       */ 
      shmdt(shm_); shm_ = NULL;
      shmctl(shmId_, IPC_RMID, 0); shmId_ = -1;
      }
 
    /*
     * Interface can no longer be used
     */
    initialized_ = 0;
  }
 
protected:
 
private:
  int initialized_;

  key_t key_;

  char thisHost_[64];

  char erms_[256];

  int shmId_;
  
  ngcdcs_mrgshm_t *shm_;

  int semId_;

  struct sembuf semOpLock_[2];

  struct sembuf semOpUnlock_[1];

  int evtSocket_[ngcdcsMRGSHM_MAXCON];

  int server_;

  int id_;

  char msg_[512];

  char statStr_[32];
  
  int port_;

  int locked_;

  int Lock_() {
    if (locked_)
      {
      return (ngcbSUCCESS);
      }
    else if (semId_ >= 0)
      {
      if (semop(semId_, semOpLock_, 2) != 0)
    {
    if (errno == EINTR)
      {
      while (semop(semId_, semOpLock_, 2) != 0)
        {
        if (errno != EINTR)
          {
          sprintf(ErrMsg(), "cannot lock semaphore - %s", strerror(errno));
          if (errno == EIDRM)
        {
        /*
         * Semaphore has been removed...
         */
        semId_ = -1;
        initialized_ = 0;
        }
          return (ngcbFAILURE);
          }
        }
      }
    else
      {
      sprintf(ErrMsg(), "cannot lock semaphore - %s", strerror(errno));
      if (errno == EIDRM)
        {
        /*
         * Semaphore has been removed...
         */
        semId_ = -1;
        initialized_ = 0;
        }
      return (ngcbFAILURE);
      }
    }
 
      locked_ = 1;
      return (ngcbSUCCESS);
      }
    else
      {
      strcpy(ErrMsg(), "invalid semaphore");
      return (ngcbFAILURE);
      }
  }

  void Unlock_() {
    if (locked_ && (semId_ >= 0))
      {
      if (semop(semId_, semOpUnlock_, 1) == 0)
    {
    locked_ = 0;
    }
      else
    {
    if (errno == EIDRM)
      {
      /*
       * Semaphore has been removed...
       */
      locked_ = 0;
      semId_ = -1;
      initialized_ = 0;
      }
    }
      }
  }

  void SemRemove_() {
    if (semId_ >= 0)
      {
      ngcbSEMIPC_CTL ctl;
      ctl.val = 0;
      semctl(semId_, 0, IPC_RMID, ctl);
      semId_ = -1;
      }
  }
 
};
 
#endif