rpc/connection.h

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