Packages
macula
0.24.1
7.1.0
7.0.0
6.0.0
5.2.2
5.2.1
5.2.0
5.1.0
5.0.0
4.8.0
4.7.1
4.7.0
4.6.0
4.5.0
4.4.10
4.4.9
4.4.8
4.4.7
4.4.6
4.4.5
4.4.4
4.4.3
4.4.2
4.4.1
4.4.0
4.3.1
4.3.0
4.2.9
4.2.8
4.2.7
4.2.6
4.2.5
4.2.4
4.2.3
4.2.2
4.2.1
4.2.0
4.1.1
4.1.0
4.0.0
3.16.0
3.15.3
3.15.2
3.15.1
3.14.0
3.13.0
3.12.1
3.12.0
3.11.1
3.11.0
3.10.3
3.10.2
3.10.1
3.9.0
3.8.0
3.7.0
3.5.0
3.4.0
3.3.0
3.2.0
3.1.0
3.0.0
2.1.1
2.1.0
2.0.0
1.5.2
1.5.1
1.4.30
1.4.29
1.4.28
1.4.27
1.4.26
1.4.25
1.4.24
1.4.23
1.4.22
1.4.21
1.4.20
1.4.19
1.4.18
1.4.17
1.4.16
1.4.15
1.4.14
1.4.13
1.4.11
1.4.10
1.4.9
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.1
1.3.0
1.2.0
1.1.0
1.0.10
1.0.9
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.48.6
0.48.5
0.48.4
0.48.3
0.48.2
0.48.1
0.48.0
0.47.1
0.47.0
0.46.3
0.46.1
0.46.0
0.45.3
0.45.2
0.45.1
0.45.0
0.44.2
0.44.1
0.44.0
0.43.3
0.43.2
0.43.1
0.43.0
0.42.9
0.42.8
0.42.7
0.42.6
0.42.5
0.42.4
0.42.3
0.42.2
0.42.1
0.42.0
0.41.1
0.41.0
0.40.1
0.40.0
0.39.9
0.39.8
0.39.7
0.39.6
0.39.5
0.39.4
0.39.3
0.39.2
0.39.1
0.39.0
0.38.8
0.38.7
0.38.6
0.38.5
0.38.4
0.38.3
0.38.2
0.38.1
0.38.0
0.37.7
0.37.6
0.37.5
0.37.4
0.37.3
0.37.2
0.37.1
0.37.0
0.36.6
0.36.5
0.36.4
0.36.3
0.36.2
0.36.1
0.36.0
0.35.4
0.35.3
0.35.2
0.35.1
0.35.0
0.34.1
0.34.0
0.33.1
0.33.0
0.32.5
0.32.4
0.32.3
0.32.2
0.32.1
0.32.0
0.31.9
0.31.8
0.31.7
0.31.6
0.31.5
0.31.4
0.31.3
0.31.2
0.31.1
0.31.0
0.30.10
0.30.9
0.30.8
0.30.7
0.30.6
0.30.5
0.30.4
0.30.3
0.30.2
0.30.1
0.30.0
0.29.0
0.28.3
0.28.2
0.28.1
0.28.0
0.27.1
0.27.0
0.26.1
0.26.0
0.25.6
0.25.5
0.25.4
0.25.3
0.25.2
0.25.1
0.25.0
0.24.6
0.24.5
0.24.4
0.24.3
0.24.2
0.24.1
0.24.0
0.23.3
0.23.2
0.23.1
0.23.0
0.22.12
0.22.11
0.22.10
0.22.9
0.22.8
0.22.7
0.22.6
0.22.5
0.22.4
0.22.3
0.22.2
0.22.1
0.22.0
0.21.7
0.21.6
0.21.5
0.21.4
0.21.2
0.21.1
0.21.0
0.20.25
0.20.24
0.20.23
0.20.22
0.20.21
0.20.20
0.20.19
0.20.18
0.20.17
0.20.16
0.20.15
0.20.14
0.20.13
0.20.12
0.20.11
0.20.10
0.20.9
0.20.8
0.20.7
0.20.6
0.20.5
0.20.3
0.20.2
0.20.1
0.20.0
0.19.2
0.19.1
0.19.0
0.18.1
0.18.0
0.17.4
0.17.3
0.17.2
0.17.1
0.17.0
0.16.6
0.16.5
0.16.4
0.16.3
0.16.2
0.16.1
0.16.0
0.15.1
0.15.0
0.14.3
0.14.2
0.14.1
0.14.0
0.12.6
0.12.5
0.12.3
0.11.3
0.10.2
0.10.1
0.10.0
0.9.2
0.9.1
0.9.0
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.10
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.30
0.7.29
0.7.28
0.7.27
0.7.26
0.7.25
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.0
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.4
0.3.3
0.3.2
0.3.1
Macula HTTP/3 Mesh SDK — connect, subscribe, publish, call, advertise
Current section
Files
Jump to
Current section
Files
src/macula_peer_system/macula_peer_connection_pool.erl
%%%-------------------------------------------------------------------
%%% @doc
%%% Peer Connection Pool - Reuses QUIC connections to remote peers.
%%%
%%% Stores one QUIC connection per endpoint. Callers open fresh streams
%%% on the reused connection (QUIC multiplexing). Connections are
%%% health-checked, LRU-evicted, and cleaned up on idle timeout.
%%%
%%% @end
%%%-------------------------------------------------------------------
-module(macula_peer_connection_pool).
-behaviour(gen_server).
-include_lib("quicer/include/quicer.hrl").
-include_lib("kernel/include/logger.hrl").
%% API
-export([
start_link/0,
start_link/1,
get_connection/1,
return_connection/2,
invalidate/1,
stats/0,
put/2,
get_connected_peers/0
]).
%% gen_server callbacks
-export([
init/1,
handle_call/3,
handle_cast/2,
handle_info/2,
terminate/2
]).
-define(SERVER, ?MODULE).
-define(TABLE, macula_peer_connection_pool_table).
-define(DEFAULT_MAX_CONNECTIONS, 100).
-define(DEFAULT_IDLE_TIMEOUT_MS, 60000).
-define(CLEANUP_INTERVAL_MS, 10000).
-record(state, {
table :: ets:tid(),
max_connections :: pos_integer(),
idle_timeout_ms :: pos_integer(),
hits :: non_neg_integer(),
misses :: non_neg_integer(),
evictions :: non_neg_integer()
}).
-record(pooled_conn, {
endpoint :: binary(),
connection :: term(),
last_used :: integer(),
created :: integer()
}).
%%%===================================================================
%%% API
%%%===================================================================
start_link() ->
start_link(#{}).
start_link(Opts) ->
gen_server:start_link({local, ?SERVER}, ?MODULE, Opts, []).
%% @doc Get a connection to an endpoint. Returns the QUIC connection handle.
%% Caller must open their own stream via macula_quic:open_stream(Conn).
-spec get_connection(binary()) -> {ok, term()} | {error, term()}.
get_connection(Endpoint) ->
%% Direct ETS read for fast path (no gen_server call)
case ets:info(?TABLE) of
undefined -> create_and_pool(Endpoint);
_ ->
case ets:lookup(?TABLE, Endpoint) of
[#pooled_conn{connection = Conn}] ->
case is_connection_alive(Conn) of
true ->
ets:update_element(?TABLE, Endpoint,
{#pooled_conn.last_used, erlang:system_time(millisecond)}),
gen_server:cast(?SERVER, hit),
{ok, Conn};
false ->
ets:delete(?TABLE, Endpoint),
create_and_pool(Endpoint)
end;
[] ->
create_and_pool(Endpoint)
end
end.
%% @doc Return a connection to the pool (store/update).
-spec return_connection(binary(), term()) -> ok.
return_connection(Endpoint, Connection) ->
gen_server:cast(?SERVER, {return_connection, Endpoint, Connection}).
%% @doc Invalidate and close a pooled connection.
-spec invalidate(binary()) -> ok.
invalidate(Endpoint) ->
gen_server:cast(?SERVER, {invalidate, Endpoint}).
%% @doc Get pool statistics.
-spec stats() -> map().
stats() ->
gen_server:call(?SERVER, stats).
%% @doc Put a connection directly (NAT system API).
-spec put(binary(), term()) -> ok.
put(PeerId, Connection) ->
gen_server:cast(?SERVER, {return_connection, PeerId, Connection}).
%% @doc Get list of peer endpoints with active connections.
-spec get_connected_peers() -> [binary()].
get_connected_peers() ->
case ets:info(?TABLE) of
undefined -> [];
_ -> ets:match(?TABLE, #pooled_conn{endpoint = '$1', _ = '_'})
end.
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
init(Opts) ->
MaxConnections = maps:get(max_connections, Opts, ?DEFAULT_MAX_CONNECTIONS),
IdleTimeoutMs = maps:get(idle_timeout_ms, Opts, ?DEFAULT_IDLE_TIMEOUT_MS),
Table = ets:new(?TABLE, [
named_table, set, public,
{keypos, #pooled_conn.endpoint},
{read_concurrency, true},
{write_concurrency, true}
]),
erlang:send_after(?CLEANUP_INTERVAL_MS, self(), cleanup_idle_connections),
{ok, #state{
table = Table,
max_connections = MaxConnections,
idle_timeout_ms = IdleTimeoutMs,
hits = 0,
misses = 0,
evictions = 0
}}.
handle_call(stats, _From, #state{hits = Hits, misses = Misses, evictions = Evictions} = State) ->
Total = Hits + Misses,
HitRate = case Total of 0 -> 0.0; _ -> Hits / Total * 100 end,
Stats = #{
hits => Hits, misses => Misses, evictions => Evictions,
total => Total, hit_rate => HitRate,
pool_size => ets:info(?TABLE, size)
},
{reply, Stats, State};
handle_call(_Request, _From, State) ->
{reply, {error, unknown_request}, State}.
handle_cast(hit, State) ->
{noreply, State#state{hits = State#state.hits + 1}};
handle_cast(miss, State) ->
{noreply, State#state{misses = State#state.misses + 1}};
handle_cast({return_connection, Endpoint, Connection}, State) ->
CurrentSize = ets:info(?TABLE, size),
NewState = maybe_evict(CurrentSize, State),
Now = erlang:system_time(millisecond),
ets:insert(?TABLE, #pooled_conn{
endpoint = Endpoint,
connection = Connection,
last_used = Now,
created = Now
}),
{noreply, NewState};
handle_cast({invalidate, Endpoint}, State) ->
case ets:lookup(?TABLE, Endpoint) of
[#pooled_conn{connection = Conn}] ->
catch quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0),
ets:delete(?TABLE, Endpoint);
[] -> ok
end,
{noreply, State};
handle_cast(_Msg, State) ->
{noreply, State}.
handle_info(cleanup_idle_connections, State) ->
Now = erlang:system_time(millisecond),
Cutoff = Now - State#state.idle_timeout_ms,
Expired = ets:select(?TABLE, [{
#pooled_conn{endpoint = '$1', last_used = '$2', connection = '$3', _ = '_'},
[{'<', '$2', Cutoff}],
[{{'$1', '$3'}}]
}]),
lists:foreach(fun({Endpoint, Conn}) ->
catch quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0),
ets:delete(?TABLE, Endpoint)
end, Expired),
erlang:send_after(?CLEANUP_INTERVAL_MS, self(), cleanup_idle_connections),
{noreply, State};
handle_info(_Info, State) ->
{noreply, State}.
terminate(_Reason, _State) ->
lists:foreach(fun(#pooled_conn{connection = Conn}) ->
catch quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0)
end, ets:tab2list(?TABLE)),
ok.
%%%===================================================================
%%% Internal
%%%===================================================================
create_and_pool(Endpoint) ->
case create_connection(Endpoint) of
{ok, Conn} ->
gen_server:cast(?SERVER, miss),
gen_server:cast(?SERVER, {return_connection, Endpoint, Conn}),
{ok, Conn};
{error, _} = Err ->
gen_server:cast(?SERVER, miss),
Err
end.
create_connection(Endpoint) when is_binary(Endpoint) ->
create_connection(binary_to_list(Endpoint));
create_connection(Endpoint) when is_list(Endpoint) ->
Stripped = strip_protocol(Endpoint),
case string:split(Stripped, ":", trailing) of
[Host, PortStr] ->
case catch list_to_integer(PortStr) of
Port when is_integer(Port), Port > 0, Port < 65536 ->
do_connect(Host, Port);
_ -> {error, invalid_port}
end;
_ -> {error, invalid_format}
end.
do_connect(Host, Port) ->
ConnOpts = [
{alpn, ["macula"]},
{verify, none},
{idle_timeout_ms, 60000},
{keep_alive_interval_ms, 20000},
{handshake_idle_timeout_ms, 30000}
],
case macula_quic:connect(Host, Port, ConnOpts, 5000) of
{ok, Conn} -> {ok, Conn};
{error, transport_down, _} -> {error, {connect_failed, transport_down}};
{error, Reason} -> {error, {connect_failed, Reason}}
end.
strip_protocol("https://" ++ Rest) -> Rest;
strip_protocol("http://" ++ Rest) -> Rest;
strip_protocol(Endpoint) -> Endpoint.
is_connection_alive(Conn) ->
case quicer:getstat(Conn, [recv_cnt]) of
{ok, _} -> true;
_ -> false
end.
maybe_evict(CurrentSize, #state{max_connections = Max, evictions = Evictions} = State)
when CurrentSize >= Max ->
evict_oldest(),
State#state{evictions = Evictions + 1};
maybe_evict(_CurrentSize, State) ->
State.
evict_oldest() ->
case ets:first(?TABLE) of
'$end_of_table' -> ok;
_ ->
Oldest = ets:foldl(fun
(#pooled_conn{endpoint = E, last_used = LU}, none) -> {E, LU};
(#pooled_conn{endpoint = E, last_used = LU}, {_OE, OLU}) when LU < OLU -> {E, LU};
(_, Acc) -> Acc
end, none, ?TABLE),
case Oldest of
none -> ok;
{Endpoint, _} ->
case ets:lookup(?TABLE, Endpoint) of
[#pooled_conn{connection = Conn}] ->
catch quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0);
_ -> ok
end,
ets:delete(?TABLE, Endpoint)
end
end.