rsm.h

   1 // replicated state machine interface.
   2
   3 #ifndef rsm_h
   4 #define rsm_h
   5
   6 #include "types.h"
   7 #include "rsm_protocol.h"
   8 #include "rpc/rpc.h"
   9 #include <arpa/inet.h>
  10 #include "config.h"
  11
  12 class rsm_state_transfer {
  13     public:
  14         virtual string marshal_state() = 0;
  15         virtual void unmarshal_state(const string &) = 0;
  16         virtual ~rsm_state_transfer() {}
  17 };
  18
  19 class rsm : public config_view_change {
  20     private:
  21         void reg1(int proc, handler *);
  22     protected:
  23         map<int, handler *> procs;
  24         config *cfg;
  25         rsm_state_transfer *stf = nullptr;
  26         rpcs *rsmrpc;
  27         // On slave: expected viewstamp of next invoke request
  28         // On primary: viewstamp for the next request from rsm_client
  29         viewstamp last_myvs{0, 0};   // Viewstamp of the last executed request
  30         viewstamp myvs{0, 1};
  31         string primary;
  32         bool insync;
  33         bool inviewchange;
  34         unsigned vid_commit;  // Latest view id that is known to rsm layer
  35         unsigned vid_insync;  // The view id that this node is synchronizing for
  36         vector<string> backups;   // A list of unsynchronized backups
  37
  38         // For testing purposes
  39         rpcs *testsvr;
  40         bool partitioned;
  41         bool dopartition;
  42         bool break1;
  43         bool break2;
  44
  45         rsm_client_protocol::status client_members(vector<string> &r, int i);
  46         rsm_protocol::status invoke(int &, int proc, viewstamp vs, const string & mreq);
  47         rsm_protocol::status transferreq(rsm_protocol::transferres &r, const string & src,
  48                 viewstamp last, unsigned vid);
  49         rsm_protocol::status transferdonereq(int &, const string & m, unsigned vid);
  50         rsm_protocol::status joinreq(string & log, const string & src, viewstamp last);
  51         rsm_test_protocol::status test_net_repairreq(rsm_test_protocol::status &r, int heal);
  52         rsm_test_protocol::status breakpointreq(rsm_test_protocol::status &r, int b);
  53
  54         mutex rsm_mutex, invoke_mutex;
  55         cond recovery_cond, sync_cond;
  56
  57         void execute(int procno, const string & req, string & r);
  58         rsm_client_protocol::status client_invoke(string & r, int procno, const string & req);
  59         bool statetransfer(const string & m, lock & rsm_mutex_lock);
  60         bool statetransferdone(const string & m, lock & rsm_mutex_lock);
  61         bool join(const string & m, lock & rsm_mutex_lock);
  62         void set_primary(unsigned vid);
  63         bool sync_with_backups(lock & rsm_mutex_lock);
  64         bool sync_with_primary(lock & rsm_mutex_lock);
  65         void net_repair(bool heal, lock & rsm_mutex_lock);
  66         void breakpoint1();
  67         void breakpoint2();
  68         void partition1(lock & rsm_mutex_lock);
  69         void commit_change(unsigned vid, lock & rsm_mutex_lock);
  70     public:
  71         rsm (const string & _first, const string & _me);
  72
  73         bool amiprimary();
  74         void set_state_transfer(rsm_state_transfer *_stf) { stf = _stf; }
  75         void recovery();
  76         void commit_change(unsigned vid);
  77
  78         template<class F, class C=void> void reg(int proc, F f, C *c=nullptr);
  79 };
  80
  81 template<class F, class C> void rsm::reg(int proc, F f, C *c) {
  82     reg1(proc, marshalled_func<F>::wrap(f, c));
  83 }
  84
  85 #endif /* rsm_h */