Compare commits
2 Commits
0dc9215672
...
33281da43f
Author | SHA1 | Date | |
---|---|---|---|
33281da43f | |||
bf9d0cd34a |
616
lib/paxos.ex
616
lib/paxos.ex
@ -1,4 +1,8 @@
|
|||||||
defmodule Utils do
|
defmodule Utils do
|
||||||
|
|
||||||
|
@min_print_level 0
|
||||||
|
|
||||||
|
|
||||||
def safecast(p, m) when p == nil, do: IO.puts('Trying to safecast #{m} with p as nil')
|
def safecast(p, m) when p == nil, do: IO.puts('Trying to safecast #{m} with p as nil')
|
||||||
def safecast(p, m) when is_pid(p), do: send(p, m)
|
def safecast(p, m) when is_pid(p), do: send(p, m)
|
||||||
def safecast(p, m) do
|
def safecast(p, m) do
|
||||||
@ -8,6 +12,10 @@ defmodule Utils do
|
|||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
def alter_name(name, part) do
|
||||||
|
String.to_atom(Atom.to_string(name) <> part)
|
||||||
|
end
|
||||||
|
|
||||||
def beb_broadcast(m, dest), do: for(p <- dest, do: safecast(p, m))
|
def beb_broadcast(m, dest), do: for(p <- dest, do: safecast(p, m))
|
||||||
|
|
||||||
def register_name(name, pid, link \\ true) do
|
def register_name(name, pid, link \\ true) do
|
||||||
@ -26,6 +34,21 @@ defmodule Utils do
|
|||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
defmacro create_log(level) do
|
||||||
|
quote do
|
||||||
|
def log(msg) do
|
||||||
|
Utils._log(msg, unquote(level))
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
def _log(msg, level) do
|
||||||
|
if (@min_print_level < level) do
|
||||||
|
IO.puts(msg)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
|
||||||
defmacro or_state(val, do: expr) do
|
defmacro or_state(val, do: expr) do
|
||||||
quote do
|
quote do
|
||||||
case unquote(val) do
|
case unquote(val) do
|
||||||
@ -34,6 +57,17 @@ defmodule Utils do
|
|||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
defmacro runfn(do: expr) do
|
||||||
|
quote do
|
||||||
|
def run(s) do
|
||||||
|
var!(state) = s
|
||||||
|
run(receive do
|
||||||
|
unquote(expr)
|
||||||
|
end)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
#
|
#
|
||||||
@ -47,6 +81,8 @@ defmodule Paxos do
|
|||||||
require Utils
|
require Utils
|
||||||
import Utils
|
import Utils
|
||||||
|
|
||||||
|
create_log 0
|
||||||
|
|
||||||
defmacro set_instmap(do: expr) do
|
defmacro set_instmap(do: expr) do
|
||||||
quote do
|
quote do
|
||||||
var!(map) = var!(state).instmap[var!(inst)]
|
var!(map) = var!(state).instmap[var!(inst)]
|
||||||
@ -57,22 +93,21 @@ defmodule Paxos do
|
|||||||
|
|
||||||
|
|
||||||
# Starts the Paxos replica with a specific name and some processes
|
# Starts the Paxos replica with a specific name and some processes
|
||||||
def start(name, processes) do
|
def start(name, processes, link \\ false) do
|
||||||
IO.puts("Starting paxos for #{name}")
|
log("Starting paxos for #{name}")
|
||||||
|
|
||||||
pid = spawn(Paxos, :init, [name, name, processes])
|
pid = spawn(Paxos, :init, [name, processes])
|
||||||
register_name(name, pid, false)
|
register_name(name, pid, link)
|
||||||
end
|
end
|
||||||
|
|
||||||
# Init event must be the first
|
# Init event must be the first
|
||||||
# one after the component is created
|
# one after the component is created
|
||||||
def init(name, parent, processes) do
|
def init(name, processes) do
|
||||||
EventualLeaderElector.start(name, processes)
|
EventualLeaderElector.start(name, processes)
|
||||||
EagerReliableBroadcast.start(name, processes)
|
EagerReliableBroadcast.start(name, processes)
|
||||||
|
|
||||||
state = %{
|
state = %{
|
||||||
name: name,
|
name: name,
|
||||||
parent: parent,
|
|
||||||
processes: processes,
|
processes: processes,
|
||||||
leader: nil,
|
leader: nil,
|
||||||
instmap: %{},
|
instmap: %{},
|
||||||
@ -115,246 +150,242 @@ defmodule Paxos do
|
|||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
def run(state) do
|
runfn do
|
||||||
run(
|
{:ele_trust, proc} ->
|
||||||
receive do
|
log("#{state.name} - #{proc} is leader")
|
||||||
{:ele_trust, proc} ->
|
|
||||||
IO.puts("#{state.name} - #{proc} is leader")
|
|
||||||
|
|
||||||
Enum.reduce(Map.keys(state.instmap), %{state | leader: proc}, fn inst, st ->
|
Enum.reduce(Map.keys(state.instmap), %{state | leader: proc}, fn inst, st ->
|
||||||
prepare(st, inst)
|
prepare(st, inst)
|
||||||
end)
|
end)
|
||||||
|
|
||||||
{:propose, inst, value, pid_to_inform, action} ->
|
{:propose, inst, value, pid_to_inform, action} ->
|
||||||
IO.puts("#{state.name} - Propose #{inspect(inst)} with action #{inspect(action)}")
|
log("#{state.name} - Propose #{inspect(inst)} with action #{inspect(action)}")
|
||||||
|
|
||||||
cond do
|
cond do
|
||||||
has_finished(state, inst, true) ->
|
has_finished(state, inst, true) ->
|
||||||
IO.puts("#{state.name} - Has already decided for #{inspect(inst)} sending #{inspect(state.decided[inst])}")
|
log("#{state.name} - Has already decided for #{inspect(inst)} sending #{inspect(state.decided[inst])}")
|
||||||
send(pid_to_inform, {:decision, inst, state.decided[inst]})
|
send(pid_to_inform, {:decision, inst, state.decided[inst]})
|
||||||
state
|
|
||||||
|
|
||||||
action == :increase_ballot_number ->
|
|
||||||
IO.puts("#{state.name} - Got request to increase ballot number for inst #{inst}")
|
|
||||||
state = has_or_create(state, inst)
|
|
||||||
|
|
||||||
set_instmap do
|
|
||||||
%{map| ballot: Ballot.inc(map.ballot)}
|
|
||||||
end
|
|
||||||
|
|
||||||
not Map.has_key?(state.instmap, inst) ->
|
|
||||||
EagerReliableBroadcast.broadcast(state.name, {:other_propose, inst, value})
|
|
||||||
state = has_or_create(state, inst, value, pid_to_inform, action)
|
|
||||||
prepare(state, inst)
|
|
||||||
|
|
||||||
state.instmap[inst].value == nil ->
|
|
||||||
EagerReliableBroadcast.broadcast(state.name, {:other_propose, inst, value})
|
|
||||||
|
|
||||||
set_instmap do
|
|
||||||
%{ map |
|
|
||||||
value: value,
|
|
||||||
pid_to_inform: pid_to_inform,
|
|
||||||
action: action,
|
|
||||||
}
|
|
||||||
end
|
|
||||||
|
|
||||||
prepare(state, inst)
|
|
||||||
|
|
||||||
true ->
|
|
||||||
EagerReliableBroadcast.broadcast(state.name, {:other_propose, inst, value})
|
|
||||||
prepare(state, inst)
|
|
||||||
end
|
|
||||||
|
|
||||||
{:rb_deliver, proc, {:other_propose, inst, value}} ->
|
|
||||||
cond do
|
|
||||||
has_finished(state, inst, true) ->
|
|
||||||
EagerReliableBroadcast.broadcast(
|
|
||||||
state.name,
|
|
||||||
{:decide, inst, state.decided[inst]}
|
|
||||||
)
|
|
||||||
state
|
|
||||||
|
|
||||||
true ->
|
|
||||||
state = has_or_create(state, inst)
|
|
||||||
set_instmap do
|
|
||||||
%{map | other_value: value}
|
|
||||||
end
|
|
||||||
prepare(state, inst)
|
|
||||||
end
|
|
||||||
|
|
||||||
{:rb_deliver, proc, {:prepare, proc, inst, ballot}} ->
|
|
||||||
IO.puts("#{state.name} - prepare from #{proc}")
|
|
||||||
|
|
||||||
cond do
|
|
||||||
has_finished(state, inst) ->
|
|
||||||
state
|
|
||||||
|
|
||||||
not Map.has_key?(state.instmap, inst) ->
|
|
||||||
state = has_or_create(state, inst)
|
|
||||||
|
|
||||||
safecast(proc, {:prepared, inst, ballot, state.instmap[inst].accepted_ballot, state.instmap[inst].accepted_value});
|
|
||||||
|
|
||||||
set_instmap do
|
|
||||||
%{ map
|
|
||||||
| ballot: ballot
|
|
||||||
}
|
|
||||||
end
|
|
||||||
|
|
||||||
Ballot.compare(ballot, &>/2, state.instmap[inst].ballot) ->
|
|
||||||
safecast(proc,
|
|
||||||
{:prepared, inst, ballot, state.instmap[inst].accepted_ballot,
|
|
||||||
state.instmap[inst].accepted_value}
|
|
||||||
)
|
|
||||||
|
|
||||||
set_instmap do
|
|
||||||
%{ map
|
|
||||||
| ballot: ballot
|
|
||||||
}
|
|
||||||
end
|
|
||||||
|
|
||||||
true ->
|
|
||||||
safecast(proc, {:nack, inst, ballot})
|
|
||||||
state
|
|
||||||
end
|
|
||||||
|
|
||||||
{:nack, inst, ballot} ->
|
|
||||||
IO.puts("#{state.name} - nack #{inspect(inst)} #{inspect(ballot)}")
|
|
||||||
|
|
||||||
cond do
|
|
||||||
has_finished(state, inst) ->
|
|
||||||
state
|
|
||||||
|
|
||||||
state.leader == state.name and Ballot.compare(state.instmap[inst].ballot, &==/2, ballot) ->
|
|
||||||
if Map.has_key?(state.instmap, inst) and state.instmap[inst].pid_to_inform != nil do
|
|
||||||
send(state.instmap[inst].pid_to_inform, {:abort, inst})
|
|
||||||
end
|
|
||||||
|
|
||||||
EagerReliableBroadcast.broadcast(state.name, {:abort, inst, ballot})
|
|
||||||
|
|
||||||
set_instmap do
|
|
||||||
%{ map | has_sent_accept: false,
|
|
||||||
has_sent_prepare: false,
|
|
||||||
ballot: Ballot.inc(map.ballot),
|
|
||||||
aborted: true,
|
|
||||||
}
|
|
||||||
end
|
|
||||||
|
|
||||||
true ->
|
|
||||||
state
|
|
||||||
end
|
|
||||||
|
|
||||||
{:rb_deliver, _proc, {:abort, inst, ballot}} ->
|
|
||||||
cond do
|
|
||||||
has_finished(state, inst) ->
|
|
||||||
state
|
|
||||||
|
|
||||||
true ->
|
|
||||||
IO.puts("#{state.name} - got information to send abort")
|
|
||||||
|
|
||||||
if Map.has_key?(state.instmap, inst) and state.instmap[inst].pid_to_inform != nil do
|
|
||||||
send(state.instmap[inst].pid_to_inform, {:abort, inst})
|
|
||||||
end
|
|
||||||
|
|
||||||
state
|
|
||||||
end
|
|
||||||
|
|
||||||
{:prepared, inst, ballot, accepted_ballot, accepted_value} ->
|
|
||||||
IO.puts(
|
|
||||||
"#{state.name} - prepared #{inspect(inst)} #{inspect(ballot)} #{inspect(accepted_ballot)} #{inspect(accepted_value)}"
|
|
||||||
)
|
|
||||||
|
|
||||||
cond do
|
|
||||||
has_finished(state, inst) ->
|
|
||||||
state
|
|
||||||
|
|
||||||
Ballot.compare(ballot, &==/2, state.instmap[inst].ballot) ->
|
|
||||||
set_instmap do
|
|
||||||
%{
|
|
||||||
map
|
|
||||||
| prepared_values: map.prepared_values ++ [{accepted_ballot, accepted_value}]
|
|
||||||
}
|
|
||||||
end
|
|
||||||
prepared(state, inst)
|
|
||||||
|
|
||||||
Ballot.compare(ballot, &>/2, state.instmap[inst].ballot) ->
|
|
||||||
IO.puts("#{state.name} - Probably recieved this before preare came to self sending with delay")
|
|
||||||
Process.send_after(self(), {:prepared, inst, ballot, accepted_ballot, accepted_value}, 100)
|
|
||||||
state
|
|
||||||
|
|
||||||
true ->
|
|
||||||
state
|
|
||||||
end
|
|
||||||
|
|
||||||
{:rb_deliver, proc, {:accept, inst, ballot, value}} ->
|
|
||||||
cond do
|
|
||||||
has_finished(state, inst) ->
|
|
||||||
state
|
|
||||||
|
|
||||||
true ->
|
|
||||||
state = has_or_create(state, inst)
|
|
||||||
|
|
||||||
if Ballot.compare(ballot, &>=/2, state.instmap[inst].ballot) do
|
|
||||||
IO.puts("#{state.name} - accept #{inspect(inst)} #{inspect(ballot)} #{inspect(value)}")
|
|
||||||
|
|
||||||
safecast(proc, {:accepted, inst, ballot})
|
|
||||||
|
|
||||||
set_instmap do
|
|
||||||
%{ map
|
|
||||||
| ballot: ballot,
|
|
||||||
accepted_value: value,
|
|
||||||
accepted_ballot: ballot
|
|
||||||
}
|
|
||||||
end
|
|
||||||
else
|
|
||||||
IO.puts("#{state.name} -> #{proc} nack")
|
|
||||||
safecast(proc, {:nack, inst, ballot})
|
|
||||||
state
|
|
||||||
end
|
|
||||||
end
|
|
||||||
|
|
||||||
{:accepted, inst, ballot} ->
|
|
||||||
IO.puts("#{state.name} - accepted #{inspect(inst)} #{inspect(ballot)}")
|
|
||||||
|
|
||||||
cond do
|
|
||||||
has_finished(state, inst) ->
|
|
||||||
state
|
|
||||||
|
|
||||||
state.leader == state.name and state.instmap[inst].ballot == ballot ->
|
|
||||||
set_instmap do
|
|
||||||
%{ map |
|
|
||||||
accepted: map.accepted + 1
|
|
||||||
}
|
|
||||||
end
|
|
||||||
|
|
||||||
accepted( state, inst)
|
|
||||||
|
|
||||||
true ->
|
|
||||||
state
|
|
||||||
end
|
|
||||||
|
|
||||||
{:get_value, inst, pid_to_inform} ->
|
|
||||||
# IO.puts("#{state.name} get_value")
|
|
||||||
if has_finished(state, inst, true) do
|
|
||||||
send(pid_to_inform, {:get_value_res_actual, inst, state.decided[inst]})
|
|
||||||
end
|
|
||||||
state
|
state
|
||||||
|
|
||||||
{:rb_deliver, _, {:decide, inst, value}} ->
|
action == :increase_ballot_number ->
|
||||||
IO.puts("#{state.name} - decided #{inspect(inst)} #{inspect(value)}")
|
log("#{state.name} - Got request to increase ballot number for inst #{inst}")
|
||||||
|
state = has_or_create(state, inst)
|
||||||
|
|
||||||
or_state not has_finished(state, inst) do
|
set_instmap do
|
||||||
if Map.has_key?(state.instmap, inst) != nil and
|
%{map| ballot: Ballot.inc(map.ballot)}
|
||||||
state.instmap[inst].pid_to_inform != nil do
|
end
|
||||||
send(state.instmap[inst].pid_to_inform, {:decision, inst, value})
|
|
||||||
end
|
|
||||||
|
|
||||||
%{ state |
|
not Map.has_key?(state.instmap, inst) ->
|
||||||
decided: Map.put(state.decided, inst, value),
|
EagerReliableBroadcast.broadcast(state.name, {:other_propose, inst, value})
|
||||||
instmap: Map.delete(state.instmap, inst)
|
state = has_or_create(state, inst, value, pid_to_inform, action)
|
||||||
|
prepare(state, inst)
|
||||||
|
|
||||||
|
state.instmap[inst].value == nil ->
|
||||||
|
EagerReliableBroadcast.broadcast(state.name, {:other_propose, inst, value})
|
||||||
|
|
||||||
|
set_instmap do
|
||||||
|
%{ map |
|
||||||
|
value: value,
|
||||||
|
pid_to_inform: pid_to_inform,
|
||||||
|
action: action,
|
||||||
}
|
}
|
||||||
end
|
end
|
||||||
|
|
||||||
|
prepare(state, inst)
|
||||||
|
|
||||||
|
true ->
|
||||||
|
EagerReliableBroadcast.broadcast(state.name, {:other_propose, inst, value})
|
||||||
|
prepare(state, inst)
|
||||||
|
end
|
||||||
|
|
||||||
|
{:rb_deliver, proc, {:other_propose, inst, value}} ->
|
||||||
|
cond do
|
||||||
|
has_finished(state, inst, true) ->
|
||||||
|
EagerReliableBroadcast.broadcast(
|
||||||
|
state.name,
|
||||||
|
{:decide, inst, state.decided[inst]}
|
||||||
|
)
|
||||||
|
state
|
||||||
|
|
||||||
|
true ->
|
||||||
|
state = has_or_create(state, inst)
|
||||||
|
set_instmap do
|
||||||
|
%{map | other_value: value}
|
||||||
|
end
|
||||||
|
prepare(state, inst)
|
||||||
|
end
|
||||||
|
|
||||||
|
{:rb_deliver, proc, {:prepare, proc, inst, ballot}} ->
|
||||||
|
log("#{state.name} - prepare from #{proc}")
|
||||||
|
|
||||||
|
cond do
|
||||||
|
has_finished(state, inst) ->
|
||||||
|
state
|
||||||
|
|
||||||
|
not Map.has_key?(state.instmap, inst) ->
|
||||||
|
state = has_or_create(state, inst)
|
||||||
|
|
||||||
|
safecast(proc, {:prepared, inst, ballot, state.instmap[inst].accepted_ballot, state.instmap[inst].accepted_value});
|
||||||
|
|
||||||
|
set_instmap do
|
||||||
|
%{ map
|
||||||
|
| ballot: ballot
|
||||||
|
}
|
||||||
|
end
|
||||||
|
|
||||||
|
Ballot.compare(ballot, &>/2, state.instmap[inst].ballot) ->
|
||||||
|
safecast(proc,
|
||||||
|
{:prepared, inst, ballot, state.instmap[inst].accepted_ballot,
|
||||||
|
state.instmap[inst].accepted_value}
|
||||||
|
)
|
||||||
|
|
||||||
|
set_instmap do
|
||||||
|
%{ map
|
||||||
|
| ballot: ballot
|
||||||
|
}
|
||||||
|
end
|
||||||
|
|
||||||
|
true ->
|
||||||
|
safecast(proc, {:nack, inst, ballot})
|
||||||
|
state
|
||||||
|
end
|
||||||
|
|
||||||
|
{:nack, inst, ballot} ->
|
||||||
|
log("#{state.name} - nack #{inspect(inst)} #{inspect(ballot)}")
|
||||||
|
|
||||||
|
cond do
|
||||||
|
has_finished(state, inst) ->
|
||||||
|
state
|
||||||
|
|
||||||
|
state.leader == state.name and Ballot.compare(state.instmap[inst].ballot, &==/2, ballot) ->
|
||||||
|
if Map.has_key?(state.instmap, inst) and state.instmap[inst].pid_to_inform != nil do
|
||||||
|
send(state.instmap[inst].pid_to_inform, {:abort, inst})
|
||||||
|
end
|
||||||
|
|
||||||
|
EagerReliableBroadcast.broadcast(state.name, {:abort, inst, ballot})
|
||||||
|
|
||||||
|
set_instmap do
|
||||||
|
%{ map | has_sent_accept: false,
|
||||||
|
has_sent_prepare: false,
|
||||||
|
ballot: Ballot.inc(map.ballot),
|
||||||
|
aborted: true,
|
||||||
|
}
|
||||||
|
end
|
||||||
|
|
||||||
|
true ->
|
||||||
|
state
|
||||||
|
end
|
||||||
|
|
||||||
|
{:rb_deliver, _proc, {:abort, inst, ballot}} ->
|
||||||
|
cond do
|
||||||
|
has_finished(state, inst) ->
|
||||||
|
state
|
||||||
|
|
||||||
|
true ->
|
||||||
|
log("#{state.name} - got information to send abort")
|
||||||
|
|
||||||
|
if Map.has_key?(state.instmap, inst) and state.instmap[inst].pid_to_inform != nil do
|
||||||
|
send(state.instmap[inst].pid_to_inform, {:abort, inst})
|
||||||
|
end
|
||||||
|
|
||||||
|
state
|
||||||
|
end
|
||||||
|
|
||||||
|
{:prepared, inst, ballot, accepted_ballot, accepted_value} ->
|
||||||
|
log(
|
||||||
|
"#{state.name} - prepared #{inspect(inst)} #{inspect(ballot)} #{inspect(accepted_ballot)} #{inspect(accepted_value)}"
|
||||||
|
)
|
||||||
|
|
||||||
|
cond do
|
||||||
|
has_finished(state, inst) ->
|
||||||
|
state
|
||||||
|
|
||||||
|
Ballot.compare(ballot, &==/2, state.instmap[inst].ballot) ->
|
||||||
|
set_instmap do
|
||||||
|
%{
|
||||||
|
map
|
||||||
|
| prepared_values: map.prepared_values ++ [{accepted_ballot, accepted_value}]
|
||||||
|
}
|
||||||
|
end
|
||||||
|
prepared(state, inst)
|
||||||
|
|
||||||
|
Ballot.compare(ballot, &>/2, state.instmap[inst].ballot) ->
|
||||||
|
log("#{state.name} - Probably recieved this before preare came to self sending with delay")
|
||||||
|
Process.send_after(self(), {:prepared, inst, ballot, accepted_ballot, accepted_value}, 100)
|
||||||
|
state
|
||||||
|
|
||||||
|
true ->
|
||||||
|
state
|
||||||
|
end
|
||||||
|
|
||||||
|
{:rb_deliver, proc, {:accept, inst, ballot, value}} ->
|
||||||
|
cond do
|
||||||
|
has_finished(state, inst) ->
|
||||||
|
state
|
||||||
|
|
||||||
|
true ->
|
||||||
|
state = has_or_create(state, inst)
|
||||||
|
|
||||||
|
if Ballot.compare(ballot, &>=/2, state.instmap[inst].ballot) do
|
||||||
|
log("#{state.name} - accept #{inspect(inst)} #{inspect(ballot)} #{inspect(value)}")
|
||||||
|
|
||||||
|
safecast(proc, {:accepted, inst, ballot})
|
||||||
|
|
||||||
|
set_instmap do
|
||||||
|
%{ map
|
||||||
|
| ballot: ballot,
|
||||||
|
accepted_value: value,
|
||||||
|
accepted_ballot: ballot
|
||||||
|
}
|
||||||
|
end
|
||||||
|
else
|
||||||
|
log("#{state.name} -> #{proc} nack")
|
||||||
|
safecast(proc, {:nack, inst, ballot})
|
||||||
|
state
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
{:accepted, inst, ballot} ->
|
||||||
|
log("#{state.name} - accepted #{inspect(inst)} #{inspect(ballot)}")
|
||||||
|
|
||||||
|
cond do
|
||||||
|
has_finished(state, inst) ->
|
||||||
|
state
|
||||||
|
|
||||||
|
state.leader == state.name and state.instmap[inst].ballot == ballot ->
|
||||||
|
set_instmap do
|
||||||
|
%{ map |
|
||||||
|
accepted: map.accepted + 1
|
||||||
|
}
|
||||||
|
end
|
||||||
|
|
||||||
|
accepted( state, inst)
|
||||||
|
|
||||||
|
true ->
|
||||||
|
state
|
||||||
|
end
|
||||||
|
|
||||||
|
{:get_value, inst, pid_to_inform} ->
|
||||||
|
# log("#{state.name} get_value")
|
||||||
|
if has_finished(state, inst, true) do
|
||||||
|
safecast(pid_to_inform, {:get_value_res, inst, state.decided[inst]})
|
||||||
|
end
|
||||||
|
state
|
||||||
|
|
||||||
|
{:rb_deliver, _, {:decide, inst, value}} ->
|
||||||
|
log("#{state.name} - decided #{inspect(inst)} #{inspect(value)}")
|
||||||
|
|
||||||
|
or_state not has_finished(state, inst) do
|
||||||
|
if Map.has_key?(state.instmap, inst) != nil and
|
||||||
|
state.instmap[inst].pid_to_inform != nil do
|
||||||
|
safecast(state.instmap[inst].pid_to_inform, {:decision, inst, value})
|
||||||
|
end
|
||||||
|
|
||||||
|
%{ state |
|
||||||
|
decided: Map.put(state.decided, inst, value),
|
||||||
|
instmap: Map.delete(state.instmap, inst)
|
||||||
|
}
|
||||||
end
|
end
|
||||||
)
|
|
||||||
end
|
end
|
||||||
|
|
||||||
#
|
#
|
||||||
@ -379,7 +410,7 @@ defmodule Paxos do
|
|||||||
|
|
||||||
true ->
|
true ->
|
||||||
ballot = Ballot.inc(state.instmap[inst].ballot)
|
ballot = Ballot.inc(state.instmap[inst].ballot)
|
||||||
IO.puts("#{state.name} - sending all prepare #{inst} #{inspect(ballot)}")
|
log("#{state.name} - sending all prepare #{inst} #{inspect(ballot)}")
|
||||||
EagerReliableBroadcast.broadcast(state.name, {:prepare, state.name, inst, ballot})
|
EagerReliableBroadcast.broadcast(state.name, {:prepare, state.name, inst, ballot})
|
||||||
|
|
||||||
set_instmap do
|
set_instmap do
|
||||||
@ -453,7 +484,7 @@ defmodule Paxos do
|
|||||||
value = state.instmap[inst].ballot_value
|
value = state.instmap[inst].ballot_value
|
||||||
|
|
||||||
if state.instmap[inst].action == :kill_before_decision do
|
if state.instmap[inst].action == :kill_before_decision do
|
||||||
IO.puts("#{state.name} - Leader has action to die before decision #{inspect({:decide, inst, value})}")
|
log("#{state.name} - Leader has action to die before decision #{inspect({:decide, inst, value})}")
|
||||||
Process.exit(self(), :kill)
|
Process.exit(self(), :kill)
|
||||||
end
|
end
|
||||||
|
|
||||||
@ -510,10 +541,7 @@ defmodule Paxos do
|
|||||||
def get_decision_loop(input) do
|
def get_decision_loop(input) do
|
||||||
{_, t} = input
|
{_, t} = input
|
||||||
receive do
|
receive do
|
||||||
{:get_value_res, inst} ->
|
{:get_value_res, inst, v} ->
|
||||||
check_and_apply(nil, inst, input, &get_decision_loop/1)
|
|
||||||
|
|
||||||
{:get_value_res_actual, inst, v} ->
|
|
||||||
check_and_apply(v, inst, input, &get_decision_loop/1)
|
check_and_apply(v, inst, input, &get_decision_loop/1)
|
||||||
|
|
||||||
x ->
|
x ->
|
||||||
@ -571,26 +599,22 @@ defmodule EagerReliableBroadcast do
|
|||||||
require Utils
|
require Utils
|
||||||
import Utils
|
import Utils
|
||||||
|
|
||||||
def get_rb_name(name) do
|
|
||||||
String.to_atom(Atom.to_string(name) <> "_rb")
|
|
||||||
end
|
|
||||||
|
|
||||||
def get_non_rb_name(name) do
|
def get_non_rb_name(name) do
|
||||||
String.to_atom(String.replace(Atom.to_string(name), "_rb", ""))
|
String.to_atom(String.replace(Atom.to_string(name), "_rb", ""))
|
||||||
end
|
end
|
||||||
|
|
||||||
def start(name, processes) do
|
def start(name, processes) do
|
||||||
pid = spawn(EagerReliableBroadcast, :init, [name, processes])
|
pid = spawn(EagerReliableBroadcast, :init, [name, processes])
|
||||||
register_name(get_rb_name(name), pid)
|
register_name(alter_name(name, "_rb"), pid)
|
||||||
end
|
end
|
||||||
|
|
||||||
# Init event must be the first
|
# Init event must be the first
|
||||||
# one after the component is created
|
# one after the component is created
|
||||||
def init(parent, processes) do
|
def init(parent, processes) do
|
||||||
state = %{
|
state = %{
|
||||||
name: get_rb_name(parent),
|
name: alter_name(parent, "_rb"),
|
||||||
parent: parent,
|
parent: parent,
|
||||||
processes: Enum.map(processes, fn name -> get_rb_name(name) end),
|
processes: Enum.map(processes, fn name -> alter_name(name, "_rb") end),
|
||||||
# Use this data structure to remember IDs of the delivered messages
|
# Use this data structure to remember IDs of the delivered messages
|
||||||
delivered: %{},
|
delivered: %{},
|
||||||
# Use this variable to remember the last sequence number used to identify a message
|
# Use this variable to remember the last sequence number used to identify a message
|
||||||
@ -600,39 +624,35 @@ defmodule EagerReliableBroadcast do
|
|||||||
run(state)
|
run(state)
|
||||||
end
|
end
|
||||||
|
|
||||||
def run(state) do
|
runfn do
|
||||||
run(
|
# Handle the broadcast request event
|
||||||
receive do
|
{:broadcast, m} ->
|
||||||
# Handle the broadcast request event
|
data_msg = {:data, state.name, state.seq_no, m}
|
||||||
{:broadcast, m} ->
|
beb_broadcast(data_msg, state.processes)
|
||||||
data_msg = {:data, state.name, state.seq_no, m}
|
%{state | seq_no: state.seq_no + 1}
|
||||||
beb_broadcast(data_msg, state.processes)
|
|
||||||
%{state | seq_no: state.seq_no + 1}
|
|
||||||
|
|
||||||
{:data, proc, seq_no, m} ->
|
{:data, proc, seq_no, m} ->
|
||||||
if not Map.has_key?(state.delivered, {proc, seq_no, m}) do
|
if not Map.has_key?(state.delivered, {proc, seq_no, m}) do
|
||||||
data_msg = {:data, proc, seq_no, m}
|
data_msg = {:data, proc, seq_no, m}
|
||||||
beb_broadcast(data_msg, state.processes)
|
beb_broadcast(data_msg, state.processes)
|
||||||
|
|
||||||
safecast(state.parent, {:rb_deliver, get_non_rb_name(proc), m})
|
safecast(state.parent, {:rb_deliver, get_non_rb_name(proc), m})
|
||||||
%{state | delivered: Map.put(state.delivered, {proc, seq_no, m}, 1)}
|
%{state | delivered: Map.put(state.delivered, {proc, seq_no, m}, 1)}
|
||||||
else
|
else
|
||||||
val = Map.get(state.delivered, {proc, seq_no, m})
|
val = Map.get(state.delivered, {proc, seq_no, m})
|
||||||
|
|
||||||
if val < Enum.count(state.processes) do
|
if val < Enum.count(state.processes) do
|
||||||
%{state | delivered: Map.put(state.delivered, {proc, seq_no, m}, val + 1)}
|
%{state | delivered: Map.put(state.delivered, {proc, seq_no, m}, val + 1)}
|
||||||
else
|
else
|
||||||
%{state | delivered: Map.delete(state.delivered, {proc, seq_no, m})}
|
%{state | delivered: Map.delete(state.delivered, {proc, seq_no, m})}
|
||||||
end
|
end
|
||||||
end
|
|
||||||
end
|
end
|
||||||
)
|
|
||||||
end
|
end
|
||||||
|
|
||||||
#############
|
#############
|
||||||
# Interface #
|
# Interface #
|
||||||
#############
|
#############
|
||||||
def broadcast(name, m), do: safecast(get_rb_name(name), {:broadcast, m})
|
def broadcast(name, m), do: safecast(alter_name(name, "_rb"), {:broadcast, m})
|
||||||
end
|
end
|
||||||
|
|
||||||
#
|
#
|
||||||
@ -643,12 +663,8 @@ defmodule EventualLeaderElector do
|
|||||||
require Utils
|
require Utils
|
||||||
import Utils
|
import Utils
|
||||||
|
|
||||||
def getEleName(name) do
|
|
||||||
String.to_atom(Atom.to_string(name) <> "_ele")
|
|
||||||
end
|
|
||||||
|
|
||||||
def start(name, processes) do
|
def start(name, processes) do
|
||||||
new_name = getEleName(name)
|
new_name = alter_name(name, "_ele")
|
||||||
pid = spawn(EventualLeaderElector, :init, [new_name, name, processes])
|
pid = spawn(EventualLeaderElector, :init, [new_name, name, processes])
|
||||||
|
|
||||||
register_name(new_name, pid)
|
register_name(new_name, pid)
|
||||||
@ -657,7 +673,7 @@ defmodule EventualLeaderElector do
|
|||||||
# Init event must be the first
|
# Init event must be the first
|
||||||
# one after the component is created
|
# one after the component is created
|
||||||
def init(name, parent, processes) do
|
def init(name, parent, processes) do
|
||||||
processes = Enum.map(processes, fn name -> getEleName(name) end)
|
processes = Enum.map(processes, fn name -> alter_name(name, "_ele") end)
|
||||||
|
|
||||||
state = %{
|
state = %{
|
||||||
name: name,
|
name: name,
|
||||||
@ -680,30 +696,26 @@ defmodule EventualLeaderElector do
|
|||||||
state
|
state
|
||||||
end
|
end
|
||||||
|
|
||||||
def run(state) do
|
runfn do
|
||||||
run(
|
{:heartbeat_request, name, seq} ->
|
||||||
receive do
|
safecast(name, {:heartbeat, state.parent, seq})
|
||||||
{:heartbeat_request, name, seq} ->
|
state
|
||||||
safecast(name, {:heartbeat, state.parent, seq})
|
|
||||||
state
|
|
||||||
|
|
||||||
{:heartbeat, name, seq} ->
|
{:heartbeat, name, seq} ->
|
||||||
or_state seq == state.seq do
|
or_state seq == state.seq do
|
||||||
%{state | heard_back: MapSet.put(state.heard_back, name)}
|
%{state | heard_back: MapSet.put(state.heard_back, name)}
|
||||||
end
|
|
||||||
|
|
||||||
{:timeout} ->
|
|
||||||
state = or_state MapSet.size(state.heard_back) >= floor(length(state.processes)/2) + 1 do
|
|
||||||
to_trust = Enum.at(Enum.sort(MapSet.to_list(state.heard_back)), 0)
|
|
||||||
|
|
||||||
or_state state.last_trust != to_trust do
|
|
||||||
safecast(state.parent, {:ele_trust, to_trust})
|
|
||||||
%{state | last_trust: to_trust}
|
|
||||||
end
|
|
||||||
end
|
|
||||||
|
|
||||||
request_heartbeats(state)
|
|
||||||
end
|
end
|
||||||
)
|
|
||||||
|
{:timeout} ->
|
||||||
|
state = or_state MapSet.size(state.heard_back) >= floor(length(state.processes)/2) + 1 do
|
||||||
|
to_trust = Enum.at(Enum.sort(MapSet.to_list(state.heard_back)), 0)
|
||||||
|
|
||||||
|
or_state state.last_trust != to_trust do
|
||||||
|
safecast(state.parent, {:ele_trust, to_trust})
|
||||||
|
%{state | last_trust: to_trust}
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
request_heartbeats(state)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
161
lib/server.ex
Normal file
161
lib/server.ex
Normal file
@ -0,0 +1,161 @@
|
|||||||
|
defmodule ServerMacros do
|
||||||
|
|
||||||
|
def create_create_loop(name, do: match_exp, else: process_exp) do
|
||||||
|
function_name = :"#{name}_loop"
|
||||||
|
|
||||||
|
ast1 = quote do
|
||||||
|
{:timeout} -> :timeout
|
||||||
|
end
|
||||||
|
ast2 = quote do
|
||||||
|
value ->
|
||||||
|
Process.send_after(self(), value, 100)
|
||||||
|
unquote(function_name)(v)
|
||||||
|
end
|
||||||
|
|
||||||
|
ast3 = ast1 ++ match_exp ++ ast2
|
||||||
|
|
||||||
|
quote do
|
||||||
|
def unquote(function_name)(v) do
|
||||||
|
var!(v) = v
|
||||||
|
unquote(process_exp)
|
||||||
|
receive do
|
||||||
|
unquote(ast3)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
def create_create_loop(name, do: exp) do
|
||||||
|
create_create_loop(name, do: exp, else: nil)
|
||||||
|
end
|
||||||
|
|
||||||
|
defmacro create_loop(name, clauses) do
|
||||||
|
create_create_loop(name, clauses)
|
||||||
|
end
|
||||||
|
|
||||||
|
defmacro try_propose(val, do: ready, else: recal_do) do
|
||||||
|
ast1 = quote do
|
||||||
|
{:timeout} -> unquote(recal_do)
|
||||||
|
{:abort} -> unquote(recal_do)
|
||||||
|
end
|
||||||
|
ast2 = quote do
|
||||||
|
{:decision, v} ->
|
||||||
|
var!(state) = apply_game(var!(state), v)
|
||||||
|
unquote(recal_do)
|
||||||
|
|
||||||
|
v ->
|
||||||
|
raise "Unknown message on try_propose #{inspect(v)}"
|
||||||
|
end
|
||||||
|
|
||||||
|
ast3 = ast1 ++ ready ++ ast2
|
||||||
|
|
||||||
|
quote do
|
||||||
|
v = Paxos.propose(var!(state).paxos, var!(state).instance, unquote(val), 1000)
|
||||||
|
case v do
|
||||||
|
unquote(ast3)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
end
|
||||||
|
|
||||||
|
defmodule Server do
|
||||||
|
require ServerMacros
|
||||||
|
import ServerMacros
|
||||||
|
require Utils
|
||||||
|
import Utils
|
||||||
|
|
||||||
|
create_log 2
|
||||||
|
|
||||||
|
def start(name, participants) do
|
||||||
|
log("starting server")
|
||||||
|
|
||||||
|
pid = spawn(Server, :init, [name, participants])
|
||||||
|
register_name(name, pid, false)
|
||||||
|
end
|
||||||
|
|
||||||
|
def init(name, participants) do
|
||||||
|
paxos = Paxos.start(alter_name(name, "_paxos"), Enum.map(participants, fn name -> alter_name(name, "_paxos") end))
|
||||||
|
state = %{
|
||||||
|
name: name,
|
||||||
|
procs: participants,
|
||||||
|
games: %{},
|
||||||
|
paxos: paxos,
|
||||||
|
instance: 0,
|
||||||
|
}
|
||||||
|
|
||||||
|
run(state)
|
||||||
|
end
|
||||||
|
|
||||||
|
runfn do
|
||||||
|
{:start_game, participants, pid_to_inform} ->
|
||||||
|
{state, game_id} = try_to_create_game(state, participants)
|
||||||
|
safecast(pid_to_inform, {:start_game_ans, game_id})
|
||||||
|
state
|
||||||
|
end
|
||||||
|
|
||||||
|
def try_to_create_game(state, participants) do
|
||||||
|
game_ids = Map.keys(state.games)
|
||||||
|
new_game_id = Enum.at(Enum.sort(game_ids), length(game_ids) - 1) + 1
|
||||||
|
|
||||||
|
# TODO: randomize game state
|
||||||
|
new_game_state = [1, 1]
|
||||||
|
# TODO: randomize Initial hand value
|
||||||
|
hand = Enum.reduce(participants, %{}, fn p, acc -> Map.put(acc, p, 1) end)
|
||||||
|
|
||||||
|
try_propose {:start_game, new_game_id, participants, new_game_state, hand}
|
||||||
|
do
|
||||||
|
{:decision, {:start_game, ^new_game_id, ^participants, ^new_game_state, ^hand}} ->
|
||||||
|
state = apply_game(state, {:start_game, new_game_id, participants, new_game_state, hand})
|
||||||
|
{state, new_game_id}
|
||||||
|
else
|
||||||
|
try_to_create_game(state, participants)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
#
|
||||||
|
# Apply Game States
|
||||||
|
#
|
||||||
|
|
||||||
|
def apply_game(state, {:start_game, game_id, participants, new_game_state, hand}) do
|
||||||
|
if state.name in participants do
|
||||||
|
%{state | games: Map.put(state.games, game_id, %{
|
||||||
|
game_state: new_game_state,
|
||||||
|
participants: participants,
|
||||||
|
hand: hand[state.name]
|
||||||
|
})}
|
||||||
|
else
|
||||||
|
%{state | games: Map.put(state.games, game_id, :not_playing_in_it)}
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
def apply_game(state, _), do: raise :do_not_know_how_to_apply_game_state
|
||||||
|
|
||||||
|
############
|
||||||
|
# Interface
|
||||||
|
############
|
||||||
|
|
||||||
|
create_loop :start_game do
|
||||||
|
{:start_game, game_id} ->
|
||||||
|
log("Started a game #{game_id}")
|
||||||
|
end
|
||||||
|
|
||||||
|
def start_game(name, participants) do
|
||||||
|
safecast(name, {:start_game, participants, self()})
|
||||||
|
start_game_loop(nil)
|
||||||
|
end
|
||||||
|
|
||||||
|
############
|
||||||
|
# Debug
|
||||||
|
############
|
||||||
|
|
||||||
|
def spinup(number_of_participants) do
|
||||||
|
procs = Enum.to_list(0..number_of_participants) |> Enum.map(fn n -> :"p#{n}" end)
|
||||||
|
Enum.map(procs, fn proc -> Server.start(proc, procs) end)
|
||||||
|
end
|
||||||
|
|
||||||
|
def kill (pids) do
|
||||||
|
pids |> Enum.map(fn m -> Process.exit(m, :kill) end)
|
||||||
|
end
|
||||||
|
|
||||||
|
end
|
26
lib/test.ex
26
lib/test.ex
@ -1,9 +1,23 @@
|
|||||||
defmodule Test do
|
defmodule Test do
|
||||||
|
|
||||||
defmacro createfuncBase(name) do
|
defmacro createfuncBase(name, do: do_exp, else: else_exp) do
|
||||||
quote do
|
b1 = quote do
|
||||||
def unquote(name)(true), do: true
|
false -> unquote(else_exp)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
b2 = b1 ++ do_exp
|
||||||
|
|
||||||
|
t = quote do
|
||||||
|
def test(v) do
|
||||||
|
case v do
|
||||||
|
unquote(b2)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
IO.puts("test #{inspect(t)}")
|
||||||
|
|
||||||
|
t
|
||||||
end
|
end
|
||||||
|
|
||||||
end
|
end
|
||||||
@ -11,7 +25,7 @@ end
|
|||||||
defmodule Test2 do
|
defmodule Test2 do
|
||||||
require Test
|
require Test
|
||||||
|
|
||||||
def test(), do: false
|
Test.createfuncBase :lol do
|
||||||
|
true -> :test
|
||||||
Test.createfuncBase(:test)
|
else :test1 end
|
||||||
end
|
end
|
||||||
|
Reference in New Issue
Block a user