rpc/connection.h

   1 #ifndef connection_h
   2 #define connection_h
   3
   4 #include <sys/types.h>
   5 #include <sys/socket.h>
   6 #include <arpa/inet.h>
   7 #include <netinet/in.h>
   8 #include <cstddef>
   9 #include <thread>
  10
  11 #include <map>
  12
  13 #include "pollmgr.h"
  14
  15 class thread_exit_exception : std::exception {
  16 };
  17
  18 class connection;
  19
  20 class chanmgr {
  21     public:
  22         virtual bool got_pdu(connection *c, char *b, int sz) = 0;
  23         virtual ~chanmgr() {}
  24 };
  25
  26 class connection : public aio_callback {
  27     public:
  28         struct charbuf {
  29             charbuf(): buf(NULL), sz(0), solong(0) {}
  30             charbuf (char *b, int s) : buf(b), sz(s), solong(0){}
  31             char *buf;
  32             int sz;
  33             int solong; //amount of bytes written or read so far
  34         };
  35
  36         connection(chanmgr *m1, int f1, int lossytest=0);
  37         ~connection();
  38
  39         int channo() { return fd_; }
  40         bool isdead();
  41         void closeconn();
  42
  43         bool send(char *b, int sz);
  44         void write_cb(int s);
  45         void read_cb(int s);
  46
  47         void incref();
  48         void decref();
  49         int ref();
  50
  51         int compare(connection *another);
  52     private:
  53
  54         bool readpdu();
  55         bool writepdu();
  56
  57         chanmgr *mgr_;
  58         const int fd_;
  59         bool dead_;
  60
  61         charbuf wpdu_;
  62         charbuf rpdu_;
  63
  64         struct timeval create_time_;
  65
  66         int waiters_;
  67         int refno_;
  68         const int lossy_;
  69
  70         std::mutex m_;
  71         std::mutex ref_m_;
  72         std::condition_variable send_complete_;
  73         std::condition_variable send_wait_;
  74 };
  75
  76 class tcpsconn {
  77     public:
  78         tcpsconn(chanmgr *m1, int port, int lossytest=0);
  79         ~tcpsconn();
  80         inline int port() { return port_; }
  81         void accept_conn();
  82     private:
  83         int port_;
  84         std::mutex m_;
  85         std::thread th_;
  86         int pipe_[2];
  87
  88         int tcp_; //file desciptor for accepting connection
  89         chanmgr *mgr_;
  90         int lossy_;
  91         std::map<int, connection *> conns_;
  92
  93         void process_accept();
  94 };
  95
  96 struct bundle {
  97     bundle(chanmgr *m, int s, int l):mgr(m),tcp(s),lossy(l) {}
  98     chanmgr *mgr;
  99     int tcp;
 100     int lossy;
 101 };
 102
 103 connection *connect_to_dst(const sockaddr_in &dst, chanmgr *mgr, int lossy=0);
 104 #endif