sensornodeipc.hpp

#ifndef DDF_SENSORENODEIPC_HPP
#define DDF_SENSORENODEIPC_HPP

#include "sensornodegeneric.hpp"
#include "mbx.hpp"

namespace jafar
{
  namespace ddf
  {
    struct SNConnectCore
    {
      unsigned int from_node;       // the id of the node who sent the message
      unsigned int code;            // contains ok or the reason of failure (SN_CODE_TYPE)
      key_t        chan_fltr_key; // the key for the connection between channel filters
      // the node who receive the connection request choose the key (ftok)
    };

    struct SNConnectStruct
    {
      long            mtype;  // one of the SN_CONNECT_TYPE
      SNConnectCore   mtext;
    };

    void PrintSNConnect(SNConnectStruct const& sn)
    {
      JFR_DEBUG("Type: " << SN_CONNECT_STRING[sn.mtype] << "\n");
      JFR_DEBUG("Node id: " << sn.mtext.from_node << "\n");
      JFR_DEBUG("Chan fltr key: " << sn.mtext.chan_fltr_key << "\n");
      JFR_DEBUG("Code: " << SN_CODE_STRING[sn.mtext.code] << "\n");
    }

    template<typename PARENT>
    class ManagerThreadFunc
    {
      PARENT *m_par;

    public:
      ManagerThreadFunc(PARENT *parent): m_par(parent) {}
      ~ManagerThreadFunc() {}
      void operator()()
      {
        m_par->SetupChanMgrComm();
      
        while(m_par->ChanMgrAlive())
        {
          m_par->Listen(); // function managing connection/deconnection of sensor nodes
        }

        m_par->CleanChanMgrComm();
      }
    };

    template<typename PARAMS>
    class SensorNodeIPC : public SensorNodeGeneric<PARAMS>
    {
      static const int  MBOX_OFFSET = 1987;

      ManagerThreadFunc<SensorNodeIPC<PARAMS> > m_ChanMgr_Fun; // Function associated to the channel filters manager
      int               m_msqid;          // the message queue id for connection/deconnection from other nodes
      bool              m_chanMgr_alive;  // the channel manager quits if set to false
      boost::thread *m_pChanMgr;      // this is the channel manager thread
      mutable boost::mutex m_msqmutex;

    public:
      SensorNodeIPC(unsigned short sv_size, time const& sync_period, PredictModelFactoryBase *pred_factory, \
                CommFactoryBase *comm_factory, bool use_delayed)
      : SensorNodeGeneric<PARAMS>(sv_size, this->m_static_id++, sync_period, pred_factory, comm_factory,\
                            use_delayed),m_ChanMgr_Fun(this),m_chanMgr_alive(true)
      {
        m_pChanMgr = new boost::thread(m_ChanMgr_Fun);
      }

      virtual ~SensorNodeIPC() {
        if (m_pChanMgr != NULL){
          m_chanMgr_alive = false;
          m_pChanMgr->join();
          delete m_pChanMgr;
        }
      }

      virtual void InitNode() {}
      virtual void Start(time const& first_deadling) {}
      virtual void KillNode() {}

      key_t GetChanMgrKey() { return this->GetID() + MBOX_OFFSET; }
      bool ChanMgrAlive() { return m_chanMgr_alive; }

      // Wait until a message has been received from node_id
      void DbgWaitInfoFromNode(unsigned short node_id) {
        time delay = time::from_us(2);

        if(!this->IsNodeConnected(node_id)) return;
      
        // Get the right comm pointer
        ChannelFilterBase * cptr = this->GetChanFltbyNodeID(node_id);
        JFR_POSTCOND(cptr, "NULL");
        while(!cptr->HasJustReceivedInfo()) delay.sleep_period();
        cptr->ResetJustReceived();
      }

      // Create the message queue for receiving connection requests and sending confirmation
      void SetupChanMgrComm() { m_msqid = mbx_create(GetChanMgrKey());  }

      // Called by the channel manager
      void Listen()
      {
        SNConnectStruct sn_connect;
        int res, from_node;
        int buf_size = sizeof(SNConnectCore);

        // Listen to connection requests
        {
          boost::mutex::scoped_lock lock(m_msqmutex);
          res = mbx_rec(m_msqid, (msgbuf*) &sn_connect, buf_size, SN_CONNECT);
        }
      
        if (res == MSG_REC)
        {
          JFR_VDEBUG( "Node " << this->m_id << " : received connection request from node " << sn_connect.mtext.from_node << "\n");

          from_node = sn_connect.mtext.from_node;
          sn_connect.mtype = SN_CONFIRM_CONNECTION;
          sn_connect.mtext.from_node = this->GetID();

          // Check if the node is already connected
          if (this->IsNodeConnected(from_node))
          {
            sn_connect.mtext.code = SN_CODE_ALREADY_CONNECTED;
          }
          else
          {
          // Create a new channel filter
          CommBase & Comm = this->CreateChannelFilter(from_node);
          sn_connect.mtext.chan_fltr_key = Comm.GetCommId();
          sn_connect.mtext.code = SN_CODE_OK;
          }

          // Send confirmation
          {
            boost::mutex::scoped_lock lock(m_msqmutex);
            mbx_send(m_msqid, (msgbuf *) &sn_connect, buf_size);
            JFR_VDEBUG( "Node " << this->m_id << " : send confirmation to node " << from_node
                  << " with key " << sn_connect.mtext.chan_fltr_key << "\n");
          }
        }

        // Listen to disconnection requests
        {
          boost::mutex::scoped_lock lock(m_msqmutex);
          res = mbx_rec(m_msqid, (msgbuf*) &sn_connect, buf_size, SN_DISCONNECT);
        }
      
        if (res == MSG_REC) {
          JFR_VDEBUG( "Received de-connection request -> not yet implemented\n");
        }
      }

      void CleanChanMgrComm() { boost::mutex::scoped_lock lock(m_msqmutex); mbx_destroy(m_msqid); }

      SN_CODE_TYPE RequestConnection(unsigned short node_id)
      {
        int msqid, res, buffer_length;
        SNConnectStruct sbuf;

        buffer_length = sizeof(SNConnectCore);

        {
          boost::mutex::scoped_lock lock(m_msqmutex);
          msqid = mbx_getq((key_t) (node_id + MBOX_OFFSET)); // Get the message queue id of node_id (for sending connection request)
        }
      
        sbuf.mtype = (long) SN_CONNECT;
        sbuf.mtext.from_node = this->m_id;    // identify the connection request source
        sbuf.mtext.chan_fltr_key = (key_t) 0; // the receiving node will fill this field
        sbuf.mtext.code = SN_CODE_VOID;

        JFR_VDEBUG( "Node " << this->m_id << " : send connection request to node " << node_id << "\n");
  
        {
          boost::mutex::scoped_lock lock(m_msqmutex);
          mbx_blksend(msqid, (msgbuf *) &sbuf, buffer_length);  // Blocking behaviour for sending
          res = mbx_blkrec(msqid, (msgbuf *) &sbuf, buffer_length, SN_CONFIRM_CONNECTION);
        }
      
        if (res == MSG_REC)
        {
          JFR_VDEBUG( "Node " << this->m_id << " : receive confirmation from node " << sbuf.mtext.from_node << "\n");

          if (sbuf.mtext.code == SN_CODE_ALREADY_CONNECTED) return SN_CODE_ALREADY_CONNECTED;
          else
          {
              this->CreateChannelFilter(sbuf.mtext.chan_fltr_key, sbuf.mtext.from_node);
          }
        }
        else {
          perror("could not receive confirmation");
          exit(EXIT_FAILURE);
        }

        return SN_CODE_OK;
      }
    };

    template<typename PARAMS>
    class SensorNodePeriodic : public SensorNodeIPC<PARAMS>
    {
      struct PerProcStruct // container structure for a periodic process
      {
        bool            initialized; // tells if the structure is initialized (valid pointers etc.)
        boost::thread *pthread;    // the main thread
        Timer_Command *cmd;        // for controlling the execution of the thread
        Thread_Periodic_Func<SensorNodeBase> *pthread_func; // the function executed by the thread (associated to the thread)
        time            period; // period of time between the calls of pthread_func
      };

      PerProcStruct m_sync_loop; // the synchronization process (periodic task)

    protected:
      using SensorNodeIPC<PARAMS>::m_next_sync_horizon;
      using SensorNodeIPC<PARAMS>::m_last_sync;
      
    public:
      SensorNodePeriodic(unsigned short sv_size, time const& sync_period, PredictModelFactoryBase *pred_factory, \
                  CommFactoryBase *comm_factory, bool use_delayed)
        : SensorNodeIPC<PARAMS>(sv_size, sync_period, pred_factory, comm_factory, use_delayed)
      {
        m_sync_loop.period = sync_period;
        m_sync_loop.initialized = false;
      }

      virtual ~SensorNodePeriodic() { KillThread(); }

      void Start(time const& first_deadline){
        time tmp(1e-3);
        m_next_sync_horizon = first_deadline;
        m_last_sync = first_deadline;
        m_sync_loop.cmd->Run(true);
        while (!m_sync_loop.cmd->IsInitialized()) tmp.sleep_period(); // wait until the periodic thread is running
      }

      void InitNode() {
        JFR_PRECOND(!m_sync_loop.initialized, "not initialized");
        CreatePeriodicThread();
      }

      void KillNode(){
        KillThread();
      }

      void Suspend() { m_sync_loop.cmd->Run(false); }
      
    private:

      void CreatePeriodicThread()
      {
        JFR_PRECOND(!m_sync_loop.initialized, "not initialized");
        m_sync_loop.cmd = new Timer_Command();
        m_sync_loop.pthread_func = new Thread_Periodic_Func<SensorNodeBase>(m_sync_loop.period, *m_sync_loop.cmd, this);
        m_sync_loop.pthread = new boost::thread(*m_sync_loop.pthread_func);
        m_sync_loop.initialized = true;
      }
      
      void KillThread() {
        if(m_sync_loop.initialized)
        {
          m_sync_loop.cmd->Kill(true); // Ask the thread to return
          m_sync_loop.pthread->join(); // Wait until it joins
          Destroy();
        }
      }

      void Destroy() {
        if (m_sync_loop.initialized) {
        delete m_sync_loop.cmd;
          delete m_sync_loop.pthread;
          delete m_sync_loop.pthread_func;
          m_sync_loop.initialized = false;
        }
      }
    };

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