commbase.hpp

#ifndef DDF_COMMBASE_HPP
#define DDF_COMMBASE_HPP

#include <deque>

extern "C"
{
#include <sys/types.h>
#include <sys/ipc.h>
}

#include "infocontainer.hpp"
#include "definitions.hpp"

namespace jafar
{
  namespace ddf
  {

    typedef enum {SN_INFO_IJ=1,SN_INFO_JI} SN_INFO_TYPE;
    #define TOGGLE_INFO_TYPE(info_type) ((info_type) == SN_INFO_IJ ? SN_INFO_JI : SN_INFO_IJ)

    const char NODE_IJ_TEXT[][20] = {"VOID", "NODE j receive", "NODE i receive"};

    struct ChanFltrMsg
    {
      long              mtype;
      unsigned char     mtext[4];   // contains the information vector, half of the information matrix (symmetric) and a timeval
    };

    class CommBase
    {
      typedef std::deque<InfoContainer>   Fifo_chan_type;
      typedef enum {CHAN_FIFO_EMPTY, CHAN_FIFO_OK, CHAN_FIFO_FAILURE} CHAN_SYNC_STATUS;

      static unsigned short   m_static_id;         // used for generating a unique key for communication (see m_comm_id)
      unsigned short          m_sv_size;
      key_t                   m_comm_id;           // a unique identifier for the communication channel
                                                   // communication endpoints with the same id communicate together
      unsigned short          m_id;                // a unique id for the comm object (the endpoint)
      Fifo_chan_type          m_fifo;              // Buffer storing incoming data
      mutable  boost::mutex   m_fifomutex;         // m_fifo is accessed asynchronously
      CHAN_SYNC_STATUS        m_sync_status;       // return status set by the last call of GetClosestInfo
      SN_INFO_TYPE            m_direction;         // from i to j or j to i (mtype of a ChanFltrMsg structure)
      bool                    m_initialized;       // set to true when it is ready to handle communication

    protected:
      bool                    m_just_received;     // debug purpose
   
    public:
      typedef enum{EC_OK, EC_TIMEOUT, EC_SEND_FAILURE, EC_READ_FAILURE, EC_READ_EMPTY} EC_STATUS;
   
      CommBase(unsigned short sv_size);
      CommBase(unsigned short sv_size, key_t comm_key);
      virtual ~CommBase();

      void SetInitialized() { m_initialized = true; }
      void WaitInitialized() {time per = time::from_us(1); while(!m_initialized) per.sleep_period(); }
   
      unsigned short GetSvSize() const { return m_sv_size; }
      SN_INFO_TYPE   GetDirection() const { return m_direction; }
      key_t GetCommId() const { return m_comm_id; }
      unsigned short GetID() { return m_id; }
      void ResetJustReceived() { m_just_received = false; }
      bool HasJustReceivedInfo() const { return m_just_received; }
   
      void StoreIncoming(InfoContainer const& info) { boost::mutex::scoped_lock lock(m_fifomutex);  m_fifo.push_back(info);}     // called by Read
      void GetClosestInfo(time const& horizon, InfoContainer &info);                // called when the channel filter synchronize
      bool Sync_Ok() const { return m_sync_status == CHAN_FIFO_OK; }
   
      virtual EC_STATUS Send(InfoContainer const& inno)=0;
      virtual EC_STATUS Read()=0;                                                   // fill the fifo with incoming information
    };

  } // namespace ddf
} // namespace jafar
#endif