%% Connection pool FFI for Gleam distribute library %% Uses ETS for atomic counters and process-based pool management %% module declaration -module(connection_pool_ffi). -export([new_pool/2, get_connection/1, release_connection/1, pool_stats/1, destroy_pool/1, is_ok/1, get_error/1, stress_test_pool/3]). %% Create a new connection pool backed by ETS %% Returns {ok, PoolId} or {error, Reason} new_pool(TargetNode, MaxConnections) when is_binary(TargetNode), is_integer(MaxConnections) -> PoolId = erlang:unique_integer([positive]), TableName = list_to_atom("pool_" ++ integer_to_list(PoolId)), try ets:new(TableName, [named_table, public, set]), ets:insert(TableName, {target_node, TargetNode}), ets:insert(TableName, {max_connections, MaxConnections}), ets:insert(TableName, {active_count, 0}), ets:insert(TableName, {available_count, MaxConnections}), {ok, {pool, PoolId, TableName}} catch _:Reason -> {error, iolist_to_binary(io_lib:format("~p", [Reason]))} end; new_pool(_, _) -> {error, <<"invalid_arguments">>}. %% Get a connection from the pool (atomic increment) %% Returns {ok, ConnectionId} or {error, pool_exhausted} get_connection({pool, _PoolId, TableName}) -> try [{max_connections, Max}] = ets:lookup(TableName, max_connections), case ets:update_counter(TableName, active_count, {2, 1, Max, Max}) of Max -> %% We hit the limit, decrement back ets:update_counter(TableName, active_count, {2, -1, 0, 0}), {error, pool_exhausted}; Active when Active < Max -> ets:update_counter(TableName, available_count, {2, -1, 0, 0}), ConnId = erlang:unique_integer([positive]), {ok, {connection, ConnId, TableName}} end catch _:_ -> {error, pool_exhausted} end; get_connection(_) -> {error, <<"invalid_pool">>}. %% Release a connection back to the pool release_connection({connection, _ConnId, TableName}) -> try ets:update_counter(TableName, active_count, {2, -1, 0, 0}), ets:update_counter(TableName, available_count, {2, 1}), ok catch _:_ -> {error, <<"release_failed">>} end; release_connection(_) -> {error, <<"invalid_connection">>}. %% Get pool statistics pool_stats({pool, _PoolId, TableName}) -> try [{target_node, TargetNode}] = ets:lookup(TableName, target_node), [{max_connections, Max}] = ets:lookup(TableName, max_connections), [{active_count, Active}] = ets:lookup(TableName, active_count), [{available_count, Available}] = ets:lookup(TableName, available_count), {ok, #{ target_node => TargetNode, max_connections => Max, active_connections => Active, available_connections => Available }} catch _:Reason -> {error, iolist_to_binary(io_lib:format("~p", [Reason]))} end; pool_stats(_) -> {error, <<"invalid_pool">>}. %% Destroy a pool and clean up ETS table destroy_pool({pool, _PoolId, TableName}) -> try ets:delete(TableName), ok catch _:_ -> {error, <<"destroy_failed">>} end; destroy_pool(_) -> {error, <<"invalid_pool">>}. %% Run a concurrent stress test: spawn Workers processes each performing OpsPerWorker stress_test_pool(Pool = {pool, _PoolId, _TableName}, OpsPerWorker, Workers) when is_integer(OpsPerWorker), is_integer(Workers), OpsPerWorker >= 0, Workers >= 0 -> Parent = self(), spawn_workers(Workers, Parent, Pool, OpsPerWorker), wait_for_workers(Workers, 0), % Return pool stats at end pool_stats(Pool); stress_test_pool(_, _, _) -> {error, <<"invalid_args">>}. spawn_workers(0, _Parent, _Pool, _Ops) -> ok; spawn_workers(N, Parent, Pool, Ops) when N > 0 -> spawn(fun() -> worker_loop(Pool, Ops, Parent) end), spawn_workers(N - 1, Parent, Pool, Ops). worker_loop(Pool, Ops, Parent) -> case Ops of 0 -> Parent ! done; _ -> % Simulate getting a connection (atomic increment) and releasing it case get_connection(Pool) of {ok, Conn} -> % Do a small amount of work ok, release_connection(Conn), worker_loop(Pool, Ops - 1, Parent); {error, _} -> % If pool exhausted or error, continue worker_loop(Pool, Ops - 1, Parent) end end. wait_for_workers(0, _CountDone) -> ok; wait_for_workers(N, _CountDone) -> receive done -> wait_for_workers(N - 1, _CountDone + 1) end. %% Helpers is_ok({ok, _}) -> true; is_ok(ok) -> true; is_ok(_) -> false. get_error({error, Reason}) when is_binary(Reason) -> Reason; get_error({error, Reason}) when is_atom(Reason) -> atom_to_binary(Reason, utf8); get_error({error, Reason}) -> iolist_to_binary(io_lib:format("~p", [Reason])); get_error(_) -> <<"unknown_error">>.