Packages
macula
0.10.1
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 - Manages pooled QUIC connections to remote peers.
%%%
%%% Problem: Creating a new QUIC connection per message adds ~50-200ms latency.
%%% Solution: Pool connections and reuse them for subsequent messages.
%%%
%%% Design:
%%% - ETS-based connection pool for O(1) lookups
%%% - LRU eviction when pool is full
%%% - Automatic connection health monitoring
%%% - Configurable pool size and idle timeout
%%%
%%% Expected improvement: 1.5-2x latency reduction for repeated messaging.
%%%
%%% @end
%%%-------------------------------------------------------------------
-module(macula_peer_connection_pool).
-behaviour(gen_server).
-include_lib("quicer/include/quicer.hrl").
%% API
-export([
start_link/0,
start_link/1,
get_connection/1,
return_connection/2,
invalidate/1,
stats/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). % 60 seconds
-define(CLEANUP_INTERVAL_MS, 10000). % Cleanup every 10 seconds
-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(),
stream :: term(),
last_used :: integer(),
created :: integer()
}).
%%%===================================================================
%%% API
%%%===================================================================
%% @doc Start the connection pool with default options.
-spec start_link() -> {ok, pid()} | {error, term()}.
start_link() ->
start_link(#{}).
%% @doc Start the connection pool with options.
%% Options:
%% - max_connections: Maximum pooled connections (default: 100)
%% - idle_timeout_ms: Idle connection timeout (default: 60000)
-spec start_link(map()) -> {ok, pid()} | {error, term()}.
start_link(Opts) ->
gen_server:start_link({local, ?SERVER}, ?MODULE, Opts, []).
%% @doc Get a connection to an endpoint (from pool or create new).
%% Returns {ok, Connection, Stream} or {error, Reason}.
-spec get_connection(binary()) -> {ok, term(), term()} | {error, term()}.
get_connection(Endpoint) ->
gen_server:call(?SERVER, {get_connection, Endpoint}, 10000).
%% @doc Return a connection to the pool for reuse.
-spec return_connection(binary(), {term(), term()}) -> ok.
return_connection(Endpoint, {Connection, Stream}) ->
gen_server:cast(?SERVER, {return_connection, Endpoint, Connection, Stream}).
%% @doc Invalidate (remove) a connection from the pool.
%% Called when a connection fails or is no longer valid.
-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).
%%%===================================================================
%%% 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}
]),
%% Schedule periodic cleanup
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({get_connection, Endpoint}, _From, State) ->
case lookup_connection(Endpoint) of
{ok, Conn, Stream} ->
%% Pool hit - update last_used timestamp
update_last_used(Endpoint),
{reply, {ok, Conn, Stream}, State#state{hits = State#state.hits + 1}};
not_found ->
%% Pool miss - create new connection
case create_connection(Endpoint) of
{ok, Conn, Stream} ->
%% Don't store in pool yet - caller will return it after use
{reply, {ok, Conn, Stream}, State#state{misses = State#state.misses + 1}};
{error, Reason} ->
{reply, {error, Reason}, State#state{misses = State#state.misses + 1}}
end
end;
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,
PoolSize = ets:info(?TABLE, size),
Stats = #{
hits => Hits,
misses => Misses,
evictions => Evictions,
total => Total,
hit_rate => HitRate,
pool_size => PoolSize
},
{reply, Stats, State};
handle_call(_Request, _From, State) ->
{reply, {error, unknown_request}, State}.
handle_cast({return_connection, Endpoint, Connection, Stream}, State) ->
%% Check pool size limit
CurrentSize = ets:info(?TABLE, size),
NewState = case CurrentSize >= State#state.max_connections of
true ->
%% Pool full - evict oldest connection
evict_oldest_connection(),
State#state{evictions = State#state.evictions + 1};
false ->
State
end,
%% Store connection in pool
Now = erlang:system_time(millisecond),
PooledConn = #pooled_conn{
endpoint = Endpoint,
connection = Connection,
stream = Stream,
last_used = Now,
created = Now
},
ets:insert(?TABLE, PooledConn),
{noreply, NewState};
handle_cast({invalidate, Endpoint}, State) ->
case ets:lookup(?TABLE, Endpoint) of
[#pooled_conn{connection = Conn, stream = Stream}] ->
%% Close connection and stream
close_connection(Conn, Stream),
ets:delete(?TABLE, Endpoint);
[] ->
ok
end,
{noreply, State};
handle_cast(_Msg, State) ->
{noreply, State}.
handle_info(cleanup_idle_connections, State) ->
%% Remove idle connections that have timed out
Now = erlang:system_time(millisecond),
Cutoff = Now - State#state.idle_timeout_ms,
%% Find and remove expired connections
Expired = ets:select(?TABLE, [{
#pooled_conn{endpoint = '$1', last_used = '$2', connection = '$3', stream = '$4', _ = '_'},
[{'<', '$2', Cutoff}],
[{{'$1', '$3', '$4'}}]
}]),
lists:foreach(fun({Endpoint, Conn, Stream}) ->
close_connection(Conn, Stream),
ets:delete(?TABLE, Endpoint)
end, Expired),
%% Schedule next cleanup
erlang:send_after(?CLEANUP_INTERVAL_MS, self(), cleanup_idle_connections),
{noreply, State};
handle_info(_Info, State) ->
{noreply, State}.
terminate(_Reason, _State) ->
%% Close all pooled connections
AllConns = ets:tab2list(?TABLE),
lists:foreach(fun(#pooled_conn{connection = Conn, stream = Stream}) ->
close_connection(Conn, Stream)
end, AllConns),
ok.
%%%===================================================================
%%% Internal Functions
%%%===================================================================
%% @private
%% @doc Look up a connection in the pool.
lookup_connection(Endpoint) ->
case ets:lookup(?TABLE, Endpoint) of
[#pooled_conn{connection = Conn, stream = Stream}] ->
%% Check if connection is still alive
case is_connection_alive(Conn) of
true ->
{ok, Conn, Stream};
false ->
%% Connection died - remove from pool
ets:delete(?TABLE, Endpoint),
not_found
end;
[] ->
not_found
end.
%% @private
%% @doc Update last_used timestamp for a connection.
update_last_used(Endpoint) ->
Now = erlang:system_time(millisecond),
ets:update_element(?TABLE, Endpoint, {#pooled_conn.last_used, Now}).
%% @private
%% @doc Create a new QUIC connection to an endpoint.
create_connection(Endpoint) ->
case parse_endpoint(Endpoint) of
{ok, Host, Port} ->
ConnectOpts = [
{alpn, ["macula"]},
{verify, none},
{idle_timeout_ms, 60000},
{keep_alive_interval_ms, 20000},
{handshake_idle_timeout_ms, 30000}
],
case macula_quic:connect(Host, Port, ConnectOpts, 5000) of
{ok, Conn} ->
case macula_quic:open_stream(Conn) of
{ok, Stream} ->
{ok, Conn, Stream};
{error, Reason} ->
quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0),
{error, {stream_failed, Reason}}
end;
{error, transport_down, _Details} ->
{error, {connect_failed, transport_down}};
{error, Reason} ->
{error, {connect_failed, Reason}}
end;
{error, Reason} ->
{error, {invalid_endpoint, Reason}}
end.
%% @private
%% @doc Parse endpoint string into host and port.
parse_endpoint(Endpoint) when is_binary(Endpoint) ->
parse_endpoint(binary_to_list(Endpoint));
parse_endpoint(Endpoint) when is_list(Endpoint) ->
case string:split(Endpoint, ":") of
[Host, PortStr] ->
try
Port = list_to_integer(PortStr),
{ok, Host, Port}
catch
_:_ -> {error, invalid_port}
end;
_ ->
{error, invalid_format}
end.
%% @private
%% @doc Check if a QUIC connection is still alive.
is_connection_alive(Conn) ->
%% Try to get connection stats - if it fails, connection is dead
case quicer:getstat(Conn, [recv_cnt]) of
{ok, _} -> true;
_ -> false
end.
%% @private
%% @doc Evict the oldest (LRU) connection from the pool.
evict_oldest_connection() ->
%% Find oldest connection by last_used timestamp
case ets:first(?TABLE) of
'$end_of_table' ->
ok;
_ ->
%% Find connection with minimum last_used
Oldest = ets:foldl(fun(#pooled_conn{endpoint = E, last_used = LU} = PC, Acc) ->
case Acc of
none -> {E, LU, PC};
{_E, MinLU, _PC} when LU < MinLU -> {E, LU, PC};
_ -> Acc
end
end, none, ?TABLE),
case Oldest of
none ->
ok;
{Endpoint, _, #pooled_conn{connection = Conn, stream = Stream}} ->
close_connection(Conn, Stream),
ets:delete(?TABLE, Endpoint)
end
end.
%% @private
%% @doc Close a QUIC connection and stream gracefully.
close_connection(Conn, Stream) ->
catch quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 0),
catch quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0),
ok.