socket.h

#ifndef INCLUDE_COMMUNICATION_SOCKET_H_
#define INCLUDE_COMMUNICATION_SOCKET_H_
#include <map>
#include <vector>
#include "communication/msg.h"
namespace singa {

const string kInprocRouterEndpoint="inproc://router";
class Socket{
  public:
  Socket(){}
  virtual ~Socket(){}
  virtual int Send(Msg** msg)=0;
  virtual Msg* Receive()=0;
  virtual void* InternalID() const=0;

 protected:
  int local_id_;
};

class BasePoller{
 public:
  virtual void Add(Socket* socket)=0;
  virtual Socket* Wait(int timeout)=0;
};

#define USE_ZMQ
#include <czmq.h>

#ifdef USE_ZMQ
class Poller: public BasePoller{
 public:
  Poller();
  virtual void Add(Socket* socket);
  virtual Socket* Wait(int duration);
 protected:
  zpoller_t *poller_;
  std::map<zsock_t*, Socket*> zsock2Socket_;
};

class Dealer : public Socket{
 public:
  /*
   * @param id local dealer ID within a procs if the dealer is from worker or
   * server thread, starts from 1 (0 is used by the router); or the connected
   * remote procs ID for inter-process dealers from the stub thread.
   */
  Dealer(int id=-1);
  virtual ~Dealer();
  virtual int Connect(string endpoint);
  virtual int Send(Msg** msg);
  virtual Msg* Receive();
  virtual void* InternalID() const{
    return dealer_;
  }
 protected:
  int id_;
  zsock_t* dealer_;
  zpoller_t* poller_;
};

class Router : public Socket{
 public:
  virtual ~Router();
  Router(int bufsize=100);
  virtual int Bind(string endpoint);
  virtual int Send(Msg** msg);
  virtual Msg* Receive();
  virtual void* InternalID() const{
    return router_;
  }
 protected:
  zsock_t* router_;
  zpoller_t* poller_;
  std::map<int, zframe_t*> id2addr_;
  std::map<int, std::vector<zmsg_t*>> bufmsg_;
  int nBufmsg_, bufsize_;
};

#elif USE_MPI
vector<shared_ptr<SafeQueue>> MPIQueues;
#endif
} /* singa */

#endif // INCLUDE_COMMUNICATION_SOCKET_H_