defprotocol Shard.Manifest do @moduledoc""" A shard manifest is a data structure that uniquely defines the identity of the shard. The hash of the manifest is the unique identifier of that shard on the network. The Manifest protocol is a protocol implemented by the manifest structs for the different shard types. It contains an operation start() that is able to launch the correct process for this shard and connect to other peers that use it. """ @doc""" Start the corresponding Shard process """ def start(manifest) end defmodule Shard.Manager do @moduledoc""" Maintains several important tables : - :shard_db (persistent with DETS) List of { id, manifest, pid | nil } - :shard_state (persistent with DETS) List of { id, state } - :peer_db (persistent with DETS) Mult-list of { shard_id, peer_info } # TODO: add health info (last seen, ping, etc) peer_info := {:inet4, ip, port} | {:inet6, ip, port} | {:onion, name} - :shard_procs (not persistent) List of { {id, path}, pid } - :connections (not persistent) List of { peer_info, pid, nil | {my_pk, his_pk} } And an internal table : - :outbox (not persistent) Multi-list of { dest_peer_info, message, time_inserted } dest := peer_info No support for messages on authenticated channels """ use GenServer require Logger @shard_db [Application.get_env(:shard, :data_path), "shard_db"] |> Path.join |> String.to_atom @shard_state [Application.get_env(:shard, :data_path), "shard_state"] |> Path.join |> String.to_atom @peer_db [Application.get_env(:shard, :data_path), "peer_db"] |> Path.join |> String.to_atom def start_link(my_port) do GenServer.start_link(__MODULE__, my_port, name: __MODULE__) end def init(my_port) do :dets.open_file(@shard_db, [type: :set]) for [{id, manifest, _pid}] <- :dets.match @shard_db, :"$1" do :dets.insert @shard_db, {id, manifest, nil} spawn fn -> Shard.Manifest.start manifest end end :dets.open_file(@shard_state, [type: :set]) :dets.open_file(@peer_db, [type: :bag]) :ets.new(:shard_procs, [:set, :protected, :named_table]) :ets.new(:connections, [:bag, :protected, :named_table]) outbox = :ets.new(:outbox, [:bag, :private]) {:ok, %{my_port: my_port, outbox: outbox} } end def handle_call({:register, shard_id, manifest, pid}, _from, state) do will_live = case :dets.lookup(@shard_db, shard_id) do [{ ^shard_id, _, pid }] when pid != nil -> not Process.alive?(pid) _ -> true end reply = if will_live do Process.monitor(pid) :dets.insert(@shard_db, {shard_id, manifest, pid}) :ok else :redundant end {:reply, reply, state} end def handle_cast({:dispatch_to, shard_id, path, pid}, state) do :ets.insert(:shard_procs, { {shard_id, path}, pid }) Process.monitor(pid) {:noreply, state} end def handle_cast({:interested, conn_pid, peer_info, auth, shards}, state) do for shard_id <- shards do case :dets.lookup(@shard_db, shard_id) do [{ ^shard_id, _, pid }] -> :dets.insert(@peer_db, {shard_id, peer_info}) GenServer.cast(pid, {:interested, conn_pid, auth}) [] -> nil end end {:noreply, state} end def handle_cast({:not_interested, peer_info, shard_id}, state) do :dets.match_delete(@peer_db, {shard_id, peer_info}) {:noreply, state} end def handle_cast({:shard_peer_db_insert, shard_id, peer_info}, state) do :dets.insert(@peer_db, {shard_id, peer_info}) {:noreply, state} end def handle_cast({:peer_up, pid, peer_info, auth}, state) do :ets.insert(:connections, {peer_info, pid, auth}) # Send interested message for all our shards id_list = (for [{id, _, _}] <- :dets.match(@shard_db, :"$1"), do: id) GenServer.cast(pid, {:send_msg, {:interested, id_list}}) # Send queued messages for {_, msg, _} <- :ets.lookup(state.outbox, peer_info) do GenServer.cast(pid, {:send_msg, msg}) end :ets.delete(state.outbox, peer_info) {:noreply, state} end def handle_cast({:peer_down, peer_pid, peer_info, auth}, state) do :ets.match_delete(:connections, {peer_info, peer_pid, auth}) {:noreply, state} end def handle_cast({:connect_and_send, peer_info, msg}, state) do case peer_info do {:tcp4, ip, port} -> add_peer(ip, port, state) currtime = System.os_time :second :ets.insert(state.outbox, {peer_info, msg, currtime}) outbox_cleanup = [ {{:_, :_, :'$1'}, [{:<, :'$1', currtime - 60}], [true]} ] :ets.select_delete(state.outbox, outbox_cleanup) end {:noreply, state} end def handle_cast({:try_connect, pk_list}, state) do for pk <- pk_list do case :dets.lookup(@peer_db, pk) do [{^pk, nil, ip, port}] -> add_peer(ip, port, state) _ -> nil end end {:noreply, state} end def handle_cast({:add_peer, ip, port}, state) do add_peer(ip, port, state) {:noreply, state} end def handle_info({:DOWN, _, :process, pid, _}, state) do :ets.match_delete(:shard_procs, {:_, pid}) {:noreply, state} end defp add_peer(ip, port, state) do spawn fn -> case :gen_tcp.connect(ip, port, [:binary, packet: 2, active: false]) do {:ok, client} -> {:ok, pid} = DynamicSupervisor.start_child(Shard.DynamicSupervisor, {SNet.TCPConn, %{socket: client, my_port: state.my_port, is_client: true, auth: nil}}) :ok = :gen_tcp.controlling_process(client, pid) _ -> Logger.info "Could not connect to #{inspect ip}:#{port}, some messages may be dropped" end end end # ====================== # CALLED BY SNet.TcpConn # ====================== @doc""" Dispatch incoming message to correct shard process """ def incoming(conn_pid, peer_info, auth, {:interested, shards}) do GenServer.cast(__MODULE__, {:interested, conn_pid, peer_info, auth, shards}) end def incoming(_conn_pid, peer_info, _auth, {:not_interested, shard}) do GenServer.cast(__MODULE__, {:not_interested, peer_info, shard}) end def incoming(conn_pid, peer_info, auth, {shard_id, path, msg}) do case :dets.lookup(@shard_db, shard_id) do [] -> GenServer.cast(conn_pid, {:send_msg, {:not_interested, shard_id}}) [_] -> case :dets.match(@peer_db, {shard_id, peer_info}) do [] -> GenServer.cast(__MODULE__, {:shard_peer_db_insert, shard_id, peer_info}) _ -> nil end case :ets.lookup(:shard_procs, {shard_id, path}) do [{ {^shard_id, ^path}, pid }] -> GenServer.cast(pid, {:msg, conn_pid, auth, shard_id, path, msg}) [] -> Logger.info("Warning: dropping message for #{inspect shard_id}/#{inspect path}, no handler running.\n\t#{inspect msg}") end end end # ================ # CALLED BY Sapp.* # ================ @doc""" Send message to a peer specified by peer id """ def send_pid(pid, msg) do GenServer.cast(pid, {:send_msg, msg}) end @doc""" Send message to a peer specified by peer info. Opens a connection if necessary. """ def send(peer_info, msg) do case :ets.lookup(:connections, peer_info) do [{^peer_info, pid, _auth}|_] -> GenServer.cast(pid, {:send_msg, msg}) [] -> GenServer.cast(__MODULE__, {:connect_and_send, peer_info, msg}) end end @doc""" Register a process as the main process for a shard. Returns either :ok or :redundant, in which case the process must exit. """ def register(shard_id, manifest, pid) do GenServer.call(__MODULE__, {:register, shard_id, manifest, pid}) end @doc""" Register a process as the handler for shard packets for a given path. """ def dispatch_to(shard_id, path, pid) do GenServer.cast(__MODULE__, {:dispatch_to, shard_id, path, pid}) end @doc""" Return the list of all peer info for peers that are interested in a certain shard """ def get_shard_peers(shard_id) do for {_, peer_info} <- :dets.lookup(@peer_db, shard_id), do: peer_info end @doc""" Return the saved state value for a shard """ def load_state(shard_id) do case :dets.lookup(@shard_state, shard_id) do [{^shard_id, state}] -> state _ -> nil end end @doc""" Save a state value for a shard """ def save_state(shard_id, state) do :dets.insert(@shard_state, {shard_id, state}) end # ================ # CALLED BY ANYONE # ================ @doc""" Connect to a peer specified by ip address and port """ def add_peer(ip, port) do GenServer.cast(__MODULE__, {:add_peer, ip, port}) end @doc""" Returns the pid for a shard if it exists """ def find_proc(shard_id) do case :dets.lookup(@shard_db, shard_id) do [{^shard_id, _, pid}] -> pid _ -> nil end end @doc""" Return the list of all shards. """ def list_shards() do for [x] <- :dets.match(@shard_db, :"$1"), do: x end @doc""" Return the list of all connected peers """ def list_connections() do for [x] <- :ets.match(:connections, :"$1"), do: x end end