More clean-ups
[invirt/third/libt4.git] / paxos.cc
1 #include "paxos.h"
2
3 using namespace std::placeholders;
4
5 paxos_change::~paxos_change() {}
6
7 bool isamember(const node_t & m, const nodes_t & nodes) {
8     return std::find(nodes.begin(), nodes.end(), m) != nodes.end();
9 }
10
11 // check if l2 contains a majority of the elements of l1
12 bool majority(const nodes_t & l1, const nodes_t & l2) {
13     auto overlap = (size_t)std::count_if(
14             l1.begin(), l1.end(), std::bind(isamember, _1, l2));
15     return overlap >= (l1.size() >> 1) + 1;
16 }
17
18 // This module implements the proposer and acceptor of the Paxos
19 // distributed algorithm as described by Lamport's "Paxos Made
20 // Simple".  To kick off an instance of Paxos, the caller supplies a
21 // list of nodes, a proposed value, and invokes the proposer.  If the
22 // majority of the nodes agree on the proposed value after running
23 // this instance of Paxos, the acceptor invokes the upcall
24 // paxos_commit to inform higher layers of the agreed value for this
25 // instance.
26
27 proposer_acceptor::proposer_acceptor(paxos_change *_delegate,
28         bool _first, const node_t & _me, const value_t & _value)
29     : delegate(_delegate), me (_me)
30 {
31     // at this point, the log has already been replayed
32     if (instance_h == 0 && _first) {
33         values[1] = _value;
34         l.loginstance(1, _value);
35         instance_h = 1;
36     }
37
38     pxs.reg(paxos_protocol::preparereq, &proposer_acceptor::preparereq, this);
39     pxs.reg(paxos_protocol::acceptreq, &proposer_acceptor::acceptreq, this);
40     pxs.reg(paxos_protocol::decidereq, &proposer_acceptor::decidereq, this);
41 }
42
43 bool proposer_acceptor::run(unsigned instance, const nodes_t & cur_nodes, const value_t & newv)
44 {
45     lock ml(proposer_mutex);
46     LOG << "initiate paxos for " << cur_nodes << " w. i=" << instance << " v=\"" << newv << "\" stable=" << stable;
47     if (!stable) {  // already running proposer?
48         LOG << "paxos proposer already running";
49         return false;
50     }
51     stable = false;
52     bool r = false;
53     proposal.n = std::max(promise.n, proposal.n) + 1;
54     nodes_t accepts;
55     value_t v;
56     if (prepare(instance, accepts, cur_nodes, v)) {
57
58         if (majority(cur_nodes, accepts)) {
59             LOG << "received a majority of prepare responses";
60
61             if (!v.size())
62                 v = newv;
63
64             breakpoint1();
65
66             nodes_t nodes;
67             nodes.swap(accepts);
68             accept(instance, accepts, nodes, v);
69
70             if (majority(cur_nodes, accepts)) {
71                 LOG << "received a majority of accept responses";
72
73                 breakpoint2();
74
75                 decide(instance, accepts, v);
76                 r = true;
77             } else {
78                 LOG << "no majority of accept responses";
79             }
80         } else {
81             LOG << "no majority of prepare responses";
82         }
83     } else {
84         LOG << "prepare is rejected " << stable;
85     }
86     stable = true;
87     return r;
88 }
89
90 bool proposer_acceptor::prepare(unsigned instance, nodes_t & accepts,
91         const nodes_t & nodes, value_t & v) {
92     LOG << "sending prepare messages (" << proposal.n << ", " << proposal.m << ", \"" << v << "\")";
93     prepareres res;
94     prop_t highest_n_a{0, ""};
95     for (auto i : nodes) {
96         auto cl = rpcc::bind_cached(i);
97         if (!cl)
98             continue;
99         int status = cl->call_timeout(paxos_protocol::preparereq, milliseconds(100),
100                 res, me, instance, proposal);
101         if (status == paxos_protocol::OK) {
102             LOG << "preparereq response type=" << res.type << " n_a=(" << res.n_a.n
103                 << ", " << res.n_a.m << ") " << "v_a=\"" << res.v_a << "\"";
104             if (res.type == prepareres::oldinstance) {
105                 LOG << "commiting old instance!";
106                 lock ml(acceptor_mutex);
107                 commit(instance, res.v_a, ml);
108                 return false;
109             } else if (res.type == prepareres::accept) {
110                 accepts.push_back(i);
111                 if (res.n_a >= highest_n_a) {
112                     LOG << "found a newer accepted proposal, \"" << res.v_a << "\", with number (" << res.n_a.n << ", " << res.n_a.m << ")";
113                     v = res.v_a;
114                     highest_n_a = res.n_a;
115                 }
116             }
117         }
118     }
119     return true;
120 }
121
122 void proposer_acceptor::accept(unsigned instance, nodes_t & accepts,
123         const nodes_t & nodes, const value_t & v) {
124     bool accept = false;
125     for (auto i : nodes) {
126         if (auto cl = rpcc::bind_cached(i)) {
127             int status = cl->call_timeout(paxos_protocol::acceptreq, milliseconds(100),
128                     accept, me, instance, proposal, v);
129             if (status == paxos_protocol::OK && accept)
130                 accepts.push_back(i);
131         }
132     }
133 }
134
135 void proposer_acceptor::decide(unsigned instance, const nodes_t & accepts, const value_t & v) {
136     int res = 0;
137     for (auto i : accepts)
138         if (auto cl = rpcc::bind_cached(i))
139             cl->call_timeout(paxos_protocol::decidereq, milliseconds(100), res, me, instance, v);
140 }
141
142 paxos_protocol::status
143 proposer_acceptor::preparereq(prepareres & r, const node_t &, unsigned instance, prop_t n) {
144     LOG << "instance " << instance << " proposal (" << n.n << ", " << n.m << ")";
145     lock ml(acceptor_mutex);
146     if (instance <= instance_h) {
147         LOG << "old instance " << instance << " has value " << values[instance];
148         r = prepareres{prepareres::oldinstance, accepted, values[instance]};
149     } else if (n > promise) {
150         LOG << "looks good to me";
151         promise = n;
152         l.logprop(promise);
153         r = prepareres{prepareres::accept, accepted, accepted_value};
154     } else {
155         LOG << "I totally rejected this request.  Ha.";
156         r = prepareres{prepareres::reject, accepted, accepted_value};
157     }
158     LOG << "preparereq is responding with oldinstance=" << r.oldinstance << " accept=" << r.accept
159         << " n_a=(" << r.n_a.n << ", " << r.n_a.m << ") " << "v_a=\"" << r.v_a << "\"";
160     return paxos_protocol::OK;
161 }
162
163 paxos_protocol::status
164 proposer_acceptor::acceptreq(bool & r, const node_t &, unsigned instance, prop_t n, const value_t & v) {
165     lock ml(acceptor_mutex);
166     r = false;
167     if (instance != instance_h + 1)
168         return paxos_protocol::ERR;
169     if (n >= promise) {
170         accepted = n;
171         accepted_value = v;
172         l.logaccept(accepted, accepted_value);
173         r = true;
174     }
175     return paxos_protocol::OK;
176 }
177
178 paxos_protocol::status
179 proposer_acceptor::decidereq(int &, const node_t &, unsigned instance, const value_t & v) {
180     lock ml(acceptor_mutex);
181     LOG << "decidereq for accepted instance " << instance << " (my instance " << instance_h << ") v=" << accepted_value;
182     if (instance == instance_h + 1) {
183         VERIFY(accepted_value == v);
184         commit(instance, v, ml);
185     } else if (instance <= instance_h) {
186         // we are ahead; ignore.
187     } else {
188         // we are behind.
189         VERIFY(0);
190     }
191     return paxos_protocol::OK;
192 }
193
194 void proposer_acceptor::commit(unsigned instance, const value_t & value, lock & acceptor_mutex_lock) {
195     VERIFY(&value != &accepted_value); // eited by aliasing?
196     VERIFY(acceptor_mutex_lock);
197     LOG << "instance=" << instance << " has v=" << value;
198     if (instance > instance_h) {
199         LOG << "highestacceptedinstance = " << instance;
200         values[instance] = value;
201         l.loginstance(instance, value);
202         instance_h = instance;
203         accepted = promise = {0, me};
204         accepted_value.clear();
205         if (delegate) {
206             acceptor_mutex_lock.unlock();
207             delegate->paxos_commit(instance, value);
208             acceptor_mutex_lock.lock();
209         }
210     }
211 }
212
213 // For testing purposes
214 void proposer_acceptor::breakpoint1() {
215     if (break1) {
216         LOG << "Dying at breakpoint 1!";
217         exit(1);
218     }
219 }
220
221 void proposer_acceptor::breakpoint2() {
222     if (break2) {
223         LOG << "Dying at breakpoint 2!";
224         exit(1);
225     }
226 }
227
228 void proposer_acceptor::breakpoint(int b) {
229     if (b == 3) {
230         LOG << "breakpoint 1";
231         break1 = true;
232     } else if (b == 4) {
233         LOG << "breakpoint 2";
234         break2 = true;
235     }
236 }