defmodule Hyparview.PeerManager do @moduledoc """ Hyparview PeerManager module """ import Logger, only: [debug: 1] alias :gen_statem, as: GenStatem alias :rand, as: Rand alias Hyparview.Config alias Hyparview.View alias Hyparview.Utils alias Hyparview.Messages.Join alias Hyparview.Messages.JoinAccepted alias Hyparview.Messages.JoinFailed alias Hyparview.Messages.ForwardJoin alias Hyparview.Messages.Shuffle alias Hyparview.Messages.ShuffleReply alias Hyparview.Messages.Neighbor alias Hyparview.Messages.NeighborAccepted alias Hyparview.Messages.NeighborRejected alias Hyparview.Messages.Disconnect alias Hyparview.Messages.Connect @behaviour GenStatem defmodule Data do @moduledoc false defstruct view: %View{}, join_interval: Config.join_interval(), join_timeout: Config.join_timeout(), shuffle_interval: Config.shuffle_interval(), neighbor_interval: Config.neighbor_interval(), node: Node.self() def new do _ = Rand.seed(:exsplus) %Data{} end end # API functions @spec fanout(MapSet.t(), term()) :: :ok def fanout(nodes, msg) do :ok = nodes |> MapSet.delete(Node.self()) |> Enum.each(&send_message(&1, msg)) end @spec send_message(nil | Node.t(), term()) :: :ok def send_message(nil, _msg), do: :ok def send_message(node, msg) do :ok = Process.send({__MODULE__, node}, msg, []) end @spec send_message(term()) :: :ok def send_message(msg) do :ok = Process.send(__MODULE__, msg, []) end @spec send_after(Node.t(), term(), non_neg_integer()) :: reference() def send_after(node, msg, after_msec) do _timer_ref = Process.send_after(__MODULE__, {:send_after, node, msg}, after_msec) end @spec send_after(term(), non_neg_integer()) :: reference() def send_after(msg, after_msec) do _timer_ref = Process.send_after(__MODULE__, msg, after_msec) end @spec get_active_view() :: MapSet.t(Node.t()) | {:error, :timeout} def get_active_view, do: get_active_view(Node.self()) @spec get_active_view(Node.t()) :: MapSet.t(Node.t()) | {:error, :timeout} def get_active_view(node), do: safe_call(node, {:get, :active}) @spec get_passive_view() :: MapSet.t(Node.t()) | {:error, :timeout} def get_passive_view, do: get_passive_view(Node.self()) @spec get_passive_view(Node.t()) :: MapSet.t(Node.t()) | {:error, :timeout} def get_passive_view(node), do: safe_call(node, {:get, :passive}) # gen_statem callback functions def callback_mode do [:state_enter, :handle_event_function] end def start_link do GenStatem.start_link({:local, __MODULE__}, __MODULE__, [], []) end def init(_args) do {:ok, INIT, Data.new()} end # Handler for common state def handle_event({:call, from}, {:get, :active}, _state, data) do {:keep_state_and_data, [{:reply, from, data.view.active}]} end def handle_event({:call, from}, {:get, :passive}, _state, data) do {:keep_state_and_data, [{:reply, from, data.view.passive}]} end def handle_event(:info, {:send_after, _node, %Join{}}, JOINED, _data) do # Drop :keep_state_and_data end def handle_event(:info, {:send_after, node, msg}, _state, _data) do :ok = send_message(node, msg) :keep_state_and_data end # handler for INIT state def handle_event(type, msg, INIT, data) do handle_INIT(type, msg, data) catch _, _ -> System.stacktrace() |> inspect() |> Logger.error() end # handler for JOINED state def handle_event(type, msg, JOINED, data) do handle_JOINED(type, msg, data) catch _, _ -> System.stacktrace() |> inspect() |> Logger.error() end # handler for unknown state def handle_event(type, msg, _state, _data) do debug(fn -> "Unhandled message received (type: #{type} msg: #{inspect(msg)})" end) :keep_state_and_data end # private functions defp handle_INIT(:enter, _old_state, data) do :ok = Join.send!(data.view) timeouts = [ {:state_timeout, 1000, StartJoin}, {:state_timeout, 50_000, StartNeighbor} ] {:keep_state, data, timeouts} end defp handle_INIT(:state_timeout, StartJoin, _data) do :repeat_state_and_data end defp handle_INIT(:state_timeout, StartNeighbor, data) do :ok = Neighbor.send!(data.view) neighbor_timeout = Utils.random_delay(data.neighbor_interval) {:keep_state, data, [{:state_timeout, neighbor_timeout, StartNeighbor}]} end defp handle_INIT(:info, %Join{sender: sender} = join, data) do :ok = debug("JOIN request received from #{sender} on #{Node.self()}") view = Join.handle(join, data.view) {:next_state, JOINED, %{data | view: view}} end defp handle_INIT(:info, %Neighbor{sender: sender} = neighbor, data) do :ok = debug("NEIGHBOR received from #{sender} on #{Node.self()}") view = Neighbor.handle(neighbor, data.view) {:next_state, %{data | view: view}} end defp handle_INIT(:info, %NeighborAccepted{sender: sender} = neighbor_accepted, data) do :ok = debug("NEIGHBOR ACCEPTED by #{sender} on #{Node.self()}") view = NeighborAccepted.handle(neighbor_accepted, data.view) {:next_state, JOINED, %{data | view: view}} end defp handle_INIT(:info, %NeighborRejected{sender: sender} = neighbor_rejected, data) do :ok = debug("NEIGHBOR REJECTED by #{sender} on #{Node.self()}") view = NeighborRejected.handle(neighbor_rejected, data.view) {:keep_state, %{data | view: view}} end defp handle_INIT(:info, %JoinAccepted{sender: sender} = join_accepted, data) do :ok = debug("JOIN accepted by #{sender} on #{Node.self()}") view = JoinAccepted.handle(join_accepted, data.view) {:next_state, JOINED, %{data | view: view}} end defp handle_INIT(:info, %JoinFailed{sender: sender} = join_failed, data) do :ok = debug("JOIN rejected by #{sender} on #{Node.self()}") _tref = JoinFailed.handle(join_failed, data.view) :keep_state_and_data end defp handle_INIT(:info, %ForwardJoin{}, _data) do {:keep_state_and_data, [:postpone]} end defp handle_INIT(:info, %Neighbor{}, _data) do {:keep_state_and_data, [:postpone]} end defp handle_INIT(:info, %Disconnect{}, _data) do {:keep_state_and_data, [:postpone]} end defp handle_INIT(type, msg, _data) do :ok = debug(fn -> "Unhandled message received (type: #{type} msg: #{inspect(msg)}) on INIT" end) :keep_state_and_data end defp handle_JOINED(:enter, _old, data) do shuffle_inval_delay = Utils.random_delay(data.shuffle_interval) _ = send_after(StartShuffle, shuffle_inval_delay) neigh_inval_delay = Utils.random_delay(data.neighbor_interval) _ = send_after(StartNeighbor, neigh_inval_delay) :ok = debug("Schedule SHUFFLE after #{shuffle_inval_delay} msec") :ok = debug("Schedule NEIGHBOR after #{neigh_inval_delay} msec") :keep_state_and_data end defp handle_JOINED(:info, StartShuffle, data) do :ok = Shuffle.send!(data.view) shuffle_inval_delay = Utils.random_delay(data.shuffle_interval) _ = send_after(StartShuffle, shuffle_inval_delay) :ok = debug("Schedule SHUFFLE after #{shuffle_inval_delay} msec") :keep_state_and_data end defp handle_JOINED(:info, StartNeighbor, data) do if View.has_free_slot_in_active_view?(data.view), do: :ok = Neighbor.send!(data.view) neigh_inval_delay = Utils.random_delay(data.neighbor_interval) _ = send_after(StartNeighbor, neigh_inval_delay) :ok = debug("Schedule NEIGHBOR after #{neigh_inval_delay} msec") :keep_state_and_data end defp handle_JOINED(:info, %Connect{sender: sender} = connect, data) do :ok = debug("CONNECT by #{sender} on #{Node.self()}") view = Connect.handle(connect, data.view) {:keep_state, %{data | view: view}} end defp handle_JOINED(:info, %Join{sender: sender} = join, data) do :ok = debug("JOIN request received from #{sender} on #{Node.self()}") view = Join.handle(join, data.view) {:keep_state, %{data | view: view}} end defp handle_JOINED(:info, %JoinAccepted{sender: sender} = join_accepted, data) do :ok = debug("JOIN accepted by #{sender} on #{Node.self()}") view = JoinAccepted.handle(join_accepted, data.view) {:keep_state, %{data | view: view}} end defp handle_JOINED(:info, %ForwardJoin{sender: sender} = forward_join, data) do :ok = debug("FORWARDJOIN received from #{sender} on #{Node.self()}") {_result, view} = ForwardJoin.handle(forward_join, data.view) {:keep_state, %{data | view: view}} end defp handle_JOINED(:info, %Shuffle{sender: sender} = shuffle, data) do :ok = debug("SHUFFLE request received from #{sender} on #{Node.self()}") view = Shuffle.handle(shuffle, data.view) {:keep_state, %{data | view: view}} end defp handle_JOINED(:info, %ShuffleReply{sender: sender} = shuffle_reply, data) do :ok = debug("SHUFFLEREPLY received from #{sender} on #{Node.self()}") view = ShuffleReply.handle(shuffle_reply, data.view) {:keep_state, %{data | view: view}} end defp handle_JOINED(:info, %Neighbor{sender: sender} = neighbor, data) do :ok = debug("NEIGHBOR received from #{sender} on #{Node.self()}") view = Neighbor.handle(neighbor, data.view) {:keep_state, %{data | view: view}} end defp handle_JOINED(:info, %NeighborAccepted{sender: sender} = neighbor_accepted, data) do :ok = debug("NEIGHBOR ACCEPTED by #{sender} on #{Node.self()}") view = NeighborAccepted.handle(neighbor_accepted, data.view) {:keep_state, %{data | view: view}} end defp handle_JOINED(:info, %NeighborRejected{sender: sender} = neighbor_rejected, data) do :ok = debug("NEIGHBOR REJECTED by #{sender} on #{Node.self()}") view = NeighborRejected.handle(neighbor_rejected, data.view) {:keep_state, %{data | view: view}} end defp handle_JOINED(:info, %Disconnect{sender: sender} = disconnect, data) do :ok = debug("DISCONNECTED #{sender} on #{Node.self()}") view = Disconnect.handle(disconnect, data.view) {:repeat_state, %{data | view: view}} end defp handle_JOINED(type, msg, _data) do debug(fn -> "Unhandled message received (type: #{type} msg: #{inspect(msg)}) on JOINED" end) :keep_state_and_data end @call_timeout 1000 @spec safe_call(Node.t(), term()) :: MapSet.t(Node.t()) | {:error, :timeout} defp safe_call(node, msg) do GenStatem.call({__MODULE__, node}, msg, @call_timeout) catch :exit, _ -> {:error, :timeout} end end