From 0ed802601812acd6ad764e6ffd5aacfd7e674553 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Fri, 12 Oct 2018 22:39:38 +0200 Subject: Connection management improvement, stuff --- shard/lib/net/group.ex | 6 +++--- shard/lib/net/manager.ex | 26 ++++++++++++++++---------- shard/lib/net/tcpconn.ex | 5 +++++ 3 files changed, 24 insertions(+), 13 deletions(-) (limited to 'shard/lib/net') diff --git a/shard/lib/net/group.ex b/shard/lib/net/group.ex index 3cce22a..a5f0867 100644 --- a/shard/lib/net/group.ex +++ b/shard/lib/net/group.ex @@ -36,7 +36,7 @@ defmodule SNet.PubShardGroup do def init_lookup(%SNet.PubShardGroup{id: id}, notify_to) do # For now: ask all currently connected peers and connect to new peers we know of spawn fn -> - for {_, pid, _} <- SNet.Manager.list_connections do + for {_, pid, _, _} <- SNet.Manager.list_connections do GenServer.cast(notify_to, {:peer_connected, pid}) end for peer_info <- Shard.Manager.get_shard_peers id do @@ -91,7 +91,7 @@ defmodule SNet.PrivGroup do def init_lookup(%SNet.PrivGroup{pk_list: pk_list}, notify_to) do spawn fn -> # 1. We might already have some connections to these guys - for {_, pid, %SNet.Auth{my_pk: my_pk, his_pk: his_pk}} <- SNet.Manager.list_connections do + for {_, pid, %SNet.Auth{my_pk: my_pk, his_pk: his_pk}, _} <- SNet.Manager.list_connections do if (my_pk in pk_list) and (his_pk in pk_list) do GenServer.cast(notify_to, {:peer_connected, pid}) end @@ -115,7 +115,7 @@ defmodule SNet.PrivGroup do end def get_connections(%SNet.PrivGroup{pk_list: pk_list}) do - for {_, pid, %SNet.Auth{my_pk: my_pk, his_pk: his_pk}} <- SNet.Manager.list_connections, + for {_, pid, %SNet.Auth{my_pk: my_pk, his_pk: his_pk}, _} <- SNet.Manager.list_connections, (my_pk in pk_list) and (his_pk in pk_list), do: pid end diff --git a/shard/lib/net/manager.ex b/shard/lib/net/manager.ex index dd80347..759c5f0 100644 --- a/shard/lib/net/manager.ex +++ b/shard/lib/net/manager.ex @@ -37,12 +37,18 @@ defmodule SNet.Manager do end def handle_call({:peer_up, pid, peer_info, auth}, _from, state) do - case :ets.match(:connections, {peer_info, :'$1', auth}) do + case :ets.match(:connections, {peer_info, :'$1', (if auth != nil do auth else :_ end), :_}) do [[pid2]|_] when pid2 != pid -> {:reply, :redundant, state} _ -> - :ets.match_delete(:connections, {peer_info, pid, :_}) - :ets.insert(:connections, {peer_info, pid, auth}) + :ets.match_delete(:connections, {peer_info, pid, :_, :_}) + :ets.insert(:connections, {peer_info, pid, auth, :established}) + + if auth != nil do + for [pid3] <- :ets.match(:connections, {peer_info, :'$1', nil, :_}) do + GenServer.cast(pid3, :close) + end + end # Send interested message for all our shards id_list = (for {id, _, _} <- Shard.Manager.list_shards(), do: id) @@ -59,7 +65,7 @@ defmodule SNet.Manager do end def handle_info({:EXIT, pid, _reason}, state) do - :ets.match_delete(:connections, {:_, pid, :_}) + :ets.match_delete(:connections, {:_, pid, :_, :_}) {:noreply, state} end @@ -67,12 +73,12 @@ defmodule SNet.Manager do if SNet.Addr.is_local? peer_info do nil else - case :ets.match(:connections, {peer_info, :'$1', (if auth != nil do auth else :_ end)}) do + case :ets.match(:connections, {peer_info, :'$1', (if auth != nil do auth else :_ end), :_}) do [[pid]|_] -> pid [] -> my_port = Application.get_env(:shard, :port) {:ok, pid} = SNet.TCPConn.start_link(%{connect_to: peer_info, my_port: my_port, auth: auth}) - :ets.insert(:connections, {peer_info, pid, auth}) + :ets.insert(:connections, {peer_info, pid, auth, :establishing}) pid end end @@ -100,14 +106,14 @@ defmodule SNet.Manager do Return the list of connections to a given peer, possibly with different auth """ def get_connections_to(peer_info) do - for {^peer_info, pid, auth} <- :ets.lookup(:connections, peer_info), do: {pid, auth} + for {^peer_info, pid, auth, _} <- :ets.lookup(:connections, peer_info), do: {pid, auth} end @doc""" Return the list of connections to a given peer that match a given auth spec """ def get_auth_connections_to(peer_info, my_auth, his_auth) do - for {^peer_info, pid, %SNet.Auth{my_pk: my_pk, his_pk: his_pk}} <- :ets.lookup(:connections, peer_info), + for {^peer_info, pid, %SNet.Auth{my_pk: my_pk, his_pk: his_pk}, _} <- :ets.lookup(:connections, peer_info), my_pk == my_auth or my_pk in my_auth, his_pk == his_auth or his_pk in his_auth, do: pid @@ -119,7 +125,7 @@ defmodule SNet.Manager do """ def send(peer_info, msg) do case :ets.lookup(:connections, peer_info) do - [{^peer_info, pid, _auth}|_] -> + [{^peer_info, pid, _auth, _}|_] -> GenServer.cast(pid, {:send_msg, msg}) [] -> GenServer.cast(__MODULE__, {:connect_and_send, peer_info, nil, msg}) @@ -127,7 +133,7 @@ defmodule SNet.Manager do end def send_auth(peer_info, auth, msg) do - case :ets.match(:connections, {peer_info, :'$1', auth}) do + case :ets.match(:connections, {peer_info, :'$1', auth, :_}) do [[pid]|_] -> GenServer.cast(pid, {:send_msg, msg}) [] -> diff --git a/shard/lib/net/tcpconn.ex b/shard/lib/net/tcpconn.ex index bd169aa..7025ffe 100644 --- a/shard/lib/net/tcpconn.ex +++ b/shard/lib/net/tcpconn.ex @@ -296,6 +296,11 @@ defmodule SNet.TCPConn do {:noreply, %{state | nonce_send: next_nonce(state.nonce_send) }} end + def handle_cast(:close, state) do + Logger.info "Closing: #{print_id state} at #{inspect state.peer_info}" + exit(:normal) + end + def handle_info({:tcp, _socket, raw_data}, state) do {:ok, msgbin} = :enacl.secretbox_open(raw_data, state.nonce_recv, state.secret_recv) msg_data = :erlang.binary_to_term(msgbin, [:safe]) -- cgit v1.2.3