fixed a lot for the tests and added extra tetts

.gitignore

@@ -25,4 +25,4 @@ distributed_system_coursework-*.tar
 # Temporary files, for example, from tests.
 /tmp/
 
-*.bream
+*.beam

lib/paxos.ex

@@ -20,15 +20,16 @@ defmodule Paxos do
       leader: nil,
       instmap: %{},
       other_values: %{},
-      decided: %{},
+      decided: %{}
-      aborted: MapSet.new(),
-      timeout: MapSet.new()
     }
 
     run(state)
   end
 
-  def add_inst_map(state, inst, value, pid_to_inform) do
+  def add_inst_map(state, inst, value, pid_to_inform, action) do
+
+    IO.puts("#{state.name} SET BALLOT VALUE 3 nil")
+
     instmap =
       Map.put(state.instmap, inst, %{
         value: value,
@@ -36,17 +37,26 @@ defmodule Paxos do
         ballot_value: nil,
         prepared_values: [],
         accepted: 0,
-        running_ballot: 0,
         accepted_ballot: nil,
         accepted_value: nil,
-        pid_to_inform: pid_to_inform
+        pid_to_inform: pid_to_inform,
+        has_sent_accept: false,
+        action: action,
       })
 
     %{state | instmap: instmap}
   end
 
+  def has_or_create(state, inst) do
+    if Map.has_key?(state.instmap, inst) do
+      state
+    else
+      add_inst_map(state, inst, nil, nil, nil)
+    end
+  end
+
   def has_finished(state, inst) do
-    Map.has_key?(state.decided, inst) or inst in state.timeout or inst in state.aborted
+    Map.has_key?(state.decided, inst)
   end
 
   def run(state) do
@@ -60,57 +70,66 @@ defmodule Paxos do
             prepare(st, inst)
           end)
 
-        {:propose, inst, value, t, pid_to_inform} ->
+        {:propose, inst, value, t, pid_to_inform, action} ->
-          IO.puts("#{state.name} - Propose #{inst}")
+          IO.puts("#{state.name} - Propose #{inspect(inst)} with action #{inspect(action)}")
 
           cond do
             has_finished(state, inst) ->
-              send(pid_to_inform, {:abort, inst})
+              IO.puts("#{state.name} - Has already decided for #{inspect(inst)} sending #{inspect(state.decided[inst])}")
+              send(pid_to_inform, {:decision, inst, state.decided[inst]})
               state
 
             not Map.has_key?(state.instmap, inst) ->
               EagerReliableBroadcast.broadcast(state.name, {:other_propose, inst, value})
-              state = add_inst_map(state, inst, value, pid_to_inform)
+              state = add_inst_map(state, inst, value, pid_to_inform, action)
               Process.send_after(self(), {:timeout, inst}, t)
               prepare(state, inst)
 
             state.instmap[inst].value == nil ->
+              EagerReliableBroadcast.broadcast(state.name, {:other_propose, inst, value})
               Process.send_after(self(), {:timeout, inst}, t)
 
               prepare(
                 set_instmap(state, inst, %{
                   state.instmap[inst]
                   | value: value,
-                    pid_to_inform: pid_to_inform
+                    pid_to_inform: pid_to_inform,
+                    action: action,
                 }),
                 inst
               )
 
             true ->
-              state
+              prepare(state, inst)
           end
 
-        {:rb_deliver, _proc, {:other_propose, inst, value}} ->
+        {:rb_deliver, proc, {:other_propose, inst, value}} ->
-          state =
+          state = %{state | other_values: Map.put(state.other_values, inst, value)}
-            if Map.has_key?(state.instmap, inst) do
+
+          cond do
+            Map.has_key?(state.decided, inst) ->
+              EagerReliableBroadcast.broadcast(
+                state.name,
+                {:decide, inst, state.decided[inst]}
+              )
               state
-            else
+
-              add_inst_map(state, inst, nil, nil)
+            true -> 
+              state = has_or_create(state, inst)
+              prepare(state, inst)
           end
 
-          %{state | other_values: Map.put(state.other_values, inst, value)}
+        {:rb_deliver, proc, {:prepare, proc, inst, ballot}} ->
-
+          IO.puts("#{state.name} - prepare from #{proc}")
-        {:rb_deliver, _proc, {:prepare, proc, inst, ballot}} ->
-          IO.puts("#{state.name} - prepare")
 
           cond do
             has_finished(state, inst) ->
               state
 
             not Map.has_key?(state.instmap, inst) ->
-              state
+              IO.puts("I think that is the cause")
+              state = has_or_create(state, inst)
 
-            ballot > state.instmap[inst].running_ballot ->
               Utils.unicast(
                 {:prepared, inst, ballot, state.instmap[inst].accepted_ballot,
                  state.instmap[inst].accepted_value},
@@ -119,7 +138,19 @@ defmodule Paxos do
               
               set_instmap(state, inst, %{
                 state.instmap[inst]
-                | running_ballot: ballot
+                | ballot: ballot
+              })
+
+            ballot > state.instmap[inst].ballot ->
+              Utils.unicast(
+                {:prepared, inst, ballot, state.instmap[inst].accepted_ballot,
+                 state.instmap[inst].accepted_value},
+                proc
+              )
+
+              set_instmap(state, inst, %{
+                state.instmap[inst]
+                | ballot: ballot
               })
 
             true ->
@@ -128,71 +159,42 @@ defmodule Paxos do
           end
 
         {:timeout, inst} ->
-          EagerReliableBroadcast.broadcast(state.name, {:timeout, inst})
+          if not has_finished(state, inst) do
-          state
-
-        {:rb_deliver, _proc, {:timeout, inst}} ->
-          IO.puts("#{state.name}- timeout")
-
-          if has_finished(state, inst) do
-            state
-          else
-            if Map.has_key?(state.instmap, inst) do
-              if state.instmap[inst].pid_to_inform do
             send(state.instmap[inst].pid_to_inform, {:timeout, inst})
           end
-            end
 
-            %{
           state
-              | instmap: Map.delete(state.instmap, inst),
-                timeout: MapSet.put(state.timeout, inst)
-            }
-          end
-
-        # {:rb_deliver, _proc, {:abort, inst}} ->
-        #   IO.puts("#{state.name}- abort")
-
-        #   if has_finished(state, inst) do
-        #     state
-        #   else
-        #     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
-        #       | aborted: MapSet.put(state.aborted, inst),
-        #         instmap: Map.delete(state.instmap, inst)
-        #     }
-        #   end
 
         {:nack, inst, ballot} ->
-          IO.puts("#{state.name}- nack")
+          IO.puts("#{state.name} - nack #{inspect(inst)} #{inspect(ballot)}")
 
-          if has_finished(state, inst) do
+          cond do
+            has_finished(state, inst) ->
               state
-          else
-            if state.leader == state.name and state.instmap[inst].ballot == ballot do
-              # EagerReliableBroadcast.broadcast(state.name, {:abort, inst})
 
+            state.leader == state.name and state.instmap[inst].ballot == 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
 
+              set_instmap(state, inst, %{
+                state.instmap[inst] | has_sent_accept: false
+              })
+
+            true ->
               state
-            else
-              state
-            end
           end
 
         {:prepared, inst, ballot, accepted_ballot, accepted_value} ->
-          IO.puts("#{state.name}- prepared")
+          IO.puts(
+            "#{state.name} - prepared #{inspect(inst)} #{inspect(ballot)} #{inspect(accepted_ballot)} #{inspect(accepted_value)}"
+          )
 
-          if has_finished(state, inst) do
+          cond do
+            has_finished(state, inst) ->
               state
-          else
+
-            if ballot == state.instmap[inst].ballot do
+            ballot == state.instmap[inst].ballot ->
               state =
                 set_instmap(state, inst, %{
                   state.instmap[inst]
@@ -200,35 +202,47 @@ defmodule Paxos do
                       state.instmap[inst].prepared_values ++ [{accepted_ballot, accepted_value}]
                 })
 
+              IO.puts("#{state.name} Try to run prepared")
+
               prepared(state, inst)
-            else
+
+            ballot > state.instmap[inst].ballot ->
+              IO.puts("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
           end
 
         {:rb_deliver, proc, {:accept, inst, ballot, value}} ->
-          IO.puts("#{state.name} accept")
+          cond do
-
+            has_finished(state, inst) ->
-          if has_finished(state, inst) do
               state
-          else
+
-            if ballot >= state.instmap[inst].running_ballot do
+            true ->
+              state = has_or_create(state, inst)
+
+              if ballot >= state.instmap[inst].ballot do
+                IO.puts("#{state.name} - accept #{inspect(inst)} #{inspect(ballot)} #{inspect(value)}")
+            
                 Utils.unicast({:accepted, inst, ballot}, proc)
 
                 set_instmap(state, inst, %{
                   state.instmap[inst]
-                | running_ballot: ballot,
+                  | ballot: ballot,
                     accepted_value: value,
                     accepted_ballot: ballot
                 })
               else
+                IO.puts("#{state.name} -> #{proc} nack")
                 Utils.unicast({:nack, inst, ballot}, proc)
                 state
               end
           end
 
         {:accepted, inst, ballot} ->
-          IO.puts("#{state.name} accepted")
+          IO.puts("#{state.name} accepted #{inspect(inst)} #{inspect(ballot)}")
 
           if has_finished(state, inst) do
             state
@@ -254,16 +268,7 @@ defmodule Paxos do
               send(pid_to_inform, {:get_value_res, inst})
 
             has_finished(state, inst) ->
-              cond do
-                inst in state.aborted ->
-                  send(pid_to_inform, {:get_value_res, inst})
-
-                inst in state.timeout ->
-                  send(pid_to_inform, {:get_value_res, inst})
-
-                Map.has_key?(state.decided, inst) ->
               send(pid_to_inform, {:get_value_res_actual, inst, state.decided[inst]})
-              end
 
             true ->
               Process.send_after(self(), {:get_value, inst, pid_to_inform, t - 500}, 500)
@@ -272,7 +277,7 @@ defmodule Paxos do
           state
 
         {:rb_deliver, _, {:decide, inst, value}} ->
-          IO.puts("#{state.name} decided")
+          IO.puts("#{state.name} decided #{inspect(inst)} #{inspect(value)}")
 
           if has_finished(state, inst) do
             state
@@ -304,18 +309,25 @@ defmodule Paxos do
   def prepare(state, _) when state.leader != state.name, do: state
 
   def prepare(state, inst) do
-    if Map.get(state.instmap, inst) == nil and Map.get(state.other_values, inst) == nil do
+    cond do
+      Map.get(state.instmap, inst) == nil and Map.get(state.other_values, inst) == nil ->
         state
-    else
+
+      Map.get(state.instmap, inst) != nil and state.instmap[inst].has_sent_accept ->
+        state
+
+      true ->
         ballot = state.instmap[inst].ballot + 1
+        IO.puts("#{state.name} sending all prepare #{inst} #{ballot}")
         EagerReliableBroadcast.broadcast(state.name, {:prepare, state.name, inst, ballot})
 
+        IO.puts("#{state.name} SET BALLOT VALUE 2 nil")
         set_instmap(state, inst, %{
           state.instmap[inst]
-        | ballot: ballot,
+          |  prepared_values: [],
-          prepared_values: [],
             accepted: 0,
-          ballot_value: nil
+            ballot_value: nil,
+            has_sent_accept: false
         })
     end
   end
@@ -326,7 +338,9 @@ defmodule Paxos do
   def prepared(state, _) when state.leader != state.name, do: state
 
   def prepared(state, inst) do
-    if length(state.instmap[inst].prepared_values) >= floor(length(state.processes) / 2) + 1 do
+    IO.puts("#{state.name} #{length(state.processes)} #{length(state.instmap[inst].prepared_values)} #{state.instmap[inst].has_sent_accept}")
+    if length(state.instmap[inst].prepared_values) >= floor(length(state.processes) / 2) + 1 and
+         not state.instmap[inst].has_sent_accept do
       {_, a_val} =
         Enum.reduce(state.instmap[inst].prepared_values, {0, nil}, fn {bal, val},
                                                                       {a_bal, a_val} ->
@@ -353,9 +367,12 @@ defmodule Paxos do
         {:accept, inst, state.instmap[inst].ballot, a_val}
       )
 
+      IO.puts("#{state.name} SET BALLOT VALUE #{inspect(a_val)}")
+
       set_instmap(state, inst, %{
         state.instmap[inst]
-        | ballot_value: a_val
+        | ballot_value: a_val,
+          has_sent_accept: true
       })
     else
       state
@@ -371,13 +388,17 @@ defmodule Paxos do
     if state.instmap[inst].accepted >= floor(length(state.processes) / 2) + 1 do
       value = state.instmap[inst].ballot_value
 
+      if state.instmap[inst].action == :kill_before_decision do
+        IO.puts("#{state.name} - Leader has action to die before decision #{inspect({:decide, inst, value})}")
+        Process.exit(self(), :kill)
+      end
+
       EagerReliableBroadcast.broadcast(
         state.name,
         {:decide, inst, value}
       )
 
-      if Map.has_key?(state.instmap, inst) != nil and
+      if state.instmap[inst].pid_to_inform != nil do
-           state.instmap[inst].pid_to_inform != nil do
         send(state.instmap[inst].pid_to_inform, {:decision, inst, value})
       end
 
@@ -391,10 +412,18 @@ defmodule Paxos do
     end
   end
 
+  def propose_action(pid, inst, value, t, action) do
+    # Utils.unicast({:propose, value}, name)
+
+    send(pid, {:propose, inst, value, t, self(), action})
+
+    propose_loop(inst)
+  end
+
   def propose(pid, inst, value, t) do
     # Utils.unicast({:propose, value}, name)
 
-    send(pid, {:propose, inst, value, t, self()})
+    send(pid, {:propose, inst, value, t, self(), nil})
 
     propose_loop(inst)
   end
@@ -422,7 +451,8 @@ defmodule Paxos do
           propose_loop(inInst)
         end
 
-      _ ->
+      x ->
+        Process.send_after(self(), x, 500)
         propose_loop(inInst)
     end
   end
@@ -448,7 +478,8 @@ defmodule Paxos do
           get_decision_loop(inInst)
         end
 
-      _ ->
+      x ->
+        Process.send_after(self(), x, 500)
         get_decision_loop(inInst)
     end
   end

lib/paxos_test.ex

@@ -1,7 +1,7 @@
 defmodule PaxosTest do
   # The functions implement 
   # the module specific testing logic
-  defp init(name, participants, all \\ false) do
+  def init(name, participants, all \\ false) do
     cpid = TestHarness.wait_to_register(:coord, :global.whereis_name(:coord))
 
     try do
@@ -24,15 +24,15 @@ defmodule PaxosTest do
     end
   end
 
-  defp kill_paxos(pid, name) do
+  def kill_paxos(pid, name) do
     Process.exit(pid, :kill)
     :global.unregister_name(name)
     pid
   end
 
-  defp wait_for_decision(_, _, timeout) when timeout <= 0, do: {:none, :none}
+  def wait_for_decision(_, _, timeout) when timeout <= 0, do: {:none, :none}
 
-  defp wait_for_decision(pid, inst, timeout) do
+  def wait_for_decision(pid, inst, timeout) do
     Process.sleep(100)
     v = Paxos.get_decision(pid, inst, 1)
 
@@ -42,7 +42,7 @@ defmodule PaxosTest do
     end
   end
 
-  defp propose_until_commit(pid, inst, val) do
+  def propose_until_commit(pid, inst, val) do
     status = Paxos.propose(pid, inst, val, 10000)
 
     case status do
@@ -365,7 +365,8 @@ defmodule PaxosTest do
             [new_leader | _] = spare
 
             if name == leader do
-              Paxos.propose(pid, 1, val, 10000)
+              # Put the time out at 1000 since that is the window of my leader elector
+              Paxos.propose(pid, 1, val, 1000)
               Process.sleep(Enum.random(1..5))
               Process.exit(pid, :kill)
             end
@@ -423,7 +424,8 @@ defmodule PaxosTest do
           leader = (fn [h | _] -> h end).(participants)
 
           if name == leader do
-            Paxos.propose(pid, 1, val, 10000)
+            # Put the time out at 1000 since that is the window of my leader elector
+            Paxos.propose(pid, 1, val, 1000)
             Process.sleep(Enum.random(1..5))
             Process.exit(pid, :kill)
           end
@@ -501,7 +503,8 @@ defmodule PaxosTest do
           # IO.puts "kill: leaders, followers = #{inspect leaders}, #{inspect followers}"
 
           if name in leaders do
-            Paxos.propose(pid, 1, val, 10000)
+            # Put the time out at 1000 since that is the window of my leader elector
+            Paxos.propose(pid, 1, val, 1000)
             Process.sleep(Enum.random(1..5))
             Process.exit(pid, :kill)
           end
@@ -573,7 +576,8 @@ defmodule PaxosTest do
             IO.puts("kill: leaders, followers = #{inspect(leaders)}, #{inspect(followers)}")
 
             if name in leaders do
-              Paxos.propose(pid, 1, val, 10000)
+              # Put the time out at 1000 since that is the window of my leader elector
+              Paxos.propose(pid, 1, val, 1000)
               Process.sleep(Enum.random(1..5))
               Process.exit(pid, :kill)
             end

lib/paxos_test_aditional.ex

@@ -0,0 +1,64 @@
+defmodule PaxosTestAditional do
+
+  # Leader crashes, no concurrent ballots
+  def run_leader_crash_simple_before_decision(name, participants, val) do
+    {cpid, pid} = PaxosTest.init(name, participants, true)
+    send(cpid, :ready)
+
+    {status, val, a, spare} =
+      try do
+        receive do
+          :start ->
+            IO.puts("#{inspect(name)}: started")
+
+            [leader | spare] = Enum.sort(participants)
+            [new_leader | _] = spare
+
+            if name == leader do
+              # Propose with action when passed with :kill_before_decision will die right before a decision is selected
+              Paxos.propose_action(pid, 1, val, 1000, :kill_before_decision)
+              # Process.sleep(Enum.random(1..5))
+              # Process.exit(pid, :kill)
+            end
+
+            if name == new_leader do
+              Process.sleep(10)
+              PaxosTest.propose_until_commit(pid, 1, val)
+            end
+
+            if name in spare do
+              {status, val} = PaxosTest.wait_for_decision(pid, 1, 10000)
+
+              if status != :none,
+                do: IO.puts("#{name}: decided #{inspect(val)}"),
+                else: IO.puts("#{name}: No decision after 10 seconds")
+
+              {status, val, 10, spare}
+            else
+              {:killed, :none, -1, spare}
+            end
+        end
+      rescue
+        _ -> {:none, :none, 10, []}
+      end
+
+    send(cpid, :done)
+
+    receive do
+      :all_done ->
+        Process.sleep(100)
+
+        ql =
+          if name in spare do
+            IO.puts("#{name}: #{inspect(ql = Process.info(pid, :message_queue_len))}")
+            ql
+          else
+            {:message_queue_len, -1}
+          end
+
+        PaxosTest.kill_paxos(pid, name)
+        send(cpid, {:finished, name, pid, status, val, a, ql})
+    end
+  end
+end
+

lib/test_script.exs

@@ -7,6 +7,7 @@ IEx.Helpers.c("paxos.ex", ".")
 # Do not modify the following ##########
 IEx.Helpers.c("test_harness.ex", ".")
 IEx.Helpers.c("paxos_test.ex", ".")
+IEx.Helpers.c("paxos_test_aditional.ex", ".")
 IEx.Helpers.c("uuid.ex", ".")
 IEx.Helpers.c("test_util.ex", ".")
 
@@ -64,7 +65,14 @@ test_suite = [
   {&PaxosTest.run_leader_crash_complex_2/3, TestUtil.get_dist_config(host, 11), 10,
    "Cascading failures of leaders and non-leaders, random delays, 7 nodes"},
   {&PaxosTest.run_leader_crash_complex_2/3, TestUtil.get_local_config(11), 10,
-   "Cascading failures of leaders and non-leaders, random delays, 7 local procs"}
+   "Cascading failures of leaders and non-leaders, random delays, 7 local procs"},
+
+  # Aditional Test functions
+  
+  {&PaxosTestAditional.run_leader_crash_simple_before_decision/3, TestUtil.get_dist_config(host, 5), 10,
+   "Leader crashes right before decision, no concurrent ballots, 5 nodes"},
+  {&PaxosTestAditional.run_leader_crash_simple_before_decision/3, TestUtil.get_local_config(5), 10,
+   "Leader crashes right before decision, no concurrent ballots, 5 local procs"},
 ]
 
 Node.stop()
@@ -116,6 +124,9 @@ Enum.reduce(test_suite, length(test_suite), fn {func, config, n, doc}, acc ->
       # IO.puts(:stderr, "#{inspect res}")
     else
       IO.puts(:stderr, "FAIL\n\t#{inspect(res)}")
+
+      # Stop tests on a fail
+      Process.exit(self(), :kill)
     end
   end