-module(nhttp_acceptor_core). -moduledoc false. %%%----------------------------------------------------------------------------- %% INCLUDES %%%----------------------------------------------------------------------------- -include("nhttp_proc_lib.hrl"). %%%----------------------------------------------------------------------------- %% API EXPORTS %%%----------------------------------------------------------------------------- -export([ get_listen_port/1, start_link/3, stop_accepting/1 ]). %%%----------------------------------------------------------------------------- %% PROC_LIB / SYS EXPORTS %%%----------------------------------------------------------------------------- -export([ init/1, system_code_change/4, system_continue/3, system_terminate/4 ]). %%%----------------------------------------------------------------------------- %% TYPES %%%----------------------------------------------------------------------------- -export_type([spawn_ctx/0]). -type spawn_ctx() :: #{ name := term(), counter := nhttp_listener_counter:counter(), opts := nhttp:opts(), conn_sup := pid(), tracker := pid() }. %%%----------------------------------------------------------------------------- %% LOCAL MACROS %%%----------------------------------------------------------------------------- -define(GET_PORT_REPLY_TIMEOUT, 5000). -define(ACCEPT_BACKOFF_INITIAL, 100). -define(ACCEPT_BACKOFF_MAX, 1000). -define(ACCEPT_ERROR_LOG_INTERVAL, 5000). %%%----------------------------------------------------------------------------- %% BEHAVIOUR CALLBACKS %%%----------------------------------------------------------------------------- -doc """ Build per-transport state from the listener Opts (e.g. the listen socket for TCP/TLS, the QUIC listener handle for HTTP/3). Returned value is stored opaquely by the core and threaded through `do_accept/1`. """. -callback init_sub(nhttp:opts()) -> term(). -doc """ Block on the transport-specific accept primitive. The 1000 ms timeout is shared across transports so the core loop can interleave system messages cheaply. """. -callback do_accept(SubState :: term()) -> {ok, term()} | {error, term()}. -doc """ Release the transport resource for an accepted connection that the core has decided NOT to spawn (at-capacity). Send any application- visible refusal here (e.g. HTTP/1.1 503). """. -callback reject(Accepted :: term()) -> ok. -doc """ Spawn the connection process for an accepted resource. Owns the full lifecycle: counter release on `nhttp_conn_sup:start_conn/3` failure, ownership handoff (e.g. `controlling_process` + `{socket_ready, _}` for sockets), and any transport-specific cleanup. Returns `ok` once the work is done. """. -callback spawn_conn(Accepted :: term(), spawn_ctx()) -> ok. %%%----------------------------------------------------------------------------- %% INTERNAL RECORDS %%%----------------------------------------------------------------------------- -record(state, { module :: module(), sub_state :: term(), name :: term(), counter :: nhttp_listener_counter:counter(), opts :: nhttp:opts(), port :: inet:port_number(), parent :: pid(), debug :: [sys:debug_option()], conn_sup :: pid(), tracker :: pid(), backoff = 0 :: non_neg_integer(), last_error_log = undefined :: integer() | undefined }). %%%----------------------------------------------------------------------------- %% API %%%----------------------------------------------------------------------------- -doc "Get the port the acceptor is listening on.". -spec get_listen_port(pid()) -> {ok, inet:port_number()} | nhttp_error:t(). get_listen_port(AcceptorPid) -> Ref = make_ref(), AcceptorPid ! {get_port, self(), Ref}, receive {port_reply, Ref, Port} -> {ok, Port} after ?GET_PORT_REPLY_TIMEOUT -> nhttp_error:accept_timeout() end. -spec start_link(module(), nhttp_registry:tab(), nhttp:opts()) -> {ok, pid()}. start_link(Mod, Tab, Opts) -> Args = {Mod, Tab, Opts, self()}, proc_lib:start_link(?MODULE, init, [Args]). -doc """ Signal the acceptor to stop accepting new connections. The acceptor exits normally. In-flight connections are unaffected. """. -spec stop_accepting(pid()) -> ok. stop_accepting(AcceptorPid) -> AcceptorPid ! stop_accepting, ok. %%%----------------------------------------------------------------------------- %% PROC_LIB CALLBACKS %%%----------------------------------------------------------------------------- -spec init({module(), nhttp_registry:tab(), nhttp:opts(), pid()}) -> no_return(). init({Mod, Tab, Opts, Parent}) -> Port = maps:get(actual_port, Opts), Debug = sys:debug_options([]), ConnSupPid = nhttp_registry:lookup_conn_sup(Tab), TrackerPid = nhttp_registry:lookup_conn_tracker(Tab), Counter = nhttp_registry:lookup_counter(Tab), Name = nhttp_registry:lookup_name(Tab), true = is_pid(ConnSupPid), true = is_pid(TrackerPid), true = Counter =/= undefined, SubState = Mod:init_sub(Opts), proc_lib:init_ack(Parent, {ok, self()}), State = #state{ module = Mod, sub_state = SubState, name = Name, counter = Counter, opts = Opts, port = Port, parent = Parent, debug = Debug, conn_sup = ConnSupPid, tracker = TrackerPid }, accept_loop(State). -spec system_code_change(#state{}, module(), term(), term()) -> {ok, #state{}}. ?NHTTP_SYSTEM_CODE_CHANGE_NOOP. -spec system_continue(pid(), [sys:debug_option()], #state{}) -> no_return(). system_continue(_Parent, Debug, State) -> accept_loop(State#state{debug = Debug}). -spec system_terminate(term(), pid(), [sys:debug_option()], #state{}) -> no_return(). system_terminate(Reason, _Parent, _Debug, _State) -> exit(Reason). %%%----------------------------------------------------------------------------- %% ACCEPT LOOP %%%----------------------------------------------------------------------------- -spec accept_loop(#state{}) -> no_return(). accept_loop(#state{parent = Parent, debug = Debug} = State) -> case receive_system_msg(Parent, Debug, State, 0) of {ok, NewDebug, NewState} -> do_accept(NewState#state{debug = NewDebug}); {stop, Reason} -> exit(Reason) end. -spec do_accept(#state{}) -> no_return(). do_accept(#state{module = Mod, sub_state = SubState} = State) -> case Mod:do_accept(SubState) of {ok, Accepted} -> handle_accepted(Accepted, State), accept_loop(State#state{backoff = 0}); {error, timeout} -> accept_loop(State); {error, closed} -> exit(normal); {error, Reason} -> accept_loop(handle_accept_error(Reason, State)) end. -spec handle_accept_error(term(), #state{}) -> #state{}. handle_accept_error(Reason, State) -> State1 = maybe_log_accept_error(Reason, State), Backoff = next_backoff(State1#state.backoff), State2 = State1#state{backoff = Backoff}, case receive_system_msg(State2#state.parent, State2#state.debug, State2, Backoff) of {ok, NewDebug, NewState} -> NewState#state{debug = NewDebug}; {stop, StopReason} -> exit(StopReason) end. -spec maybe_log_accept_error(term(), #state{}) -> #state{}. maybe_log_accept_error(Reason, #state{last_error_log = Last, name = Name} = State) -> Now = erlang:monotonic_time(millisecond), case Last =:= undefined orelse (Now - Last) >= ?ACCEPT_ERROR_LOG_INTERVAL of true -> nhttp_log:accept_error(#{listener_name => Name}, Reason), State#state{last_error_log = Now}; false -> State end. -spec next_backoff(non_neg_integer()) -> pos_integer(). next_backoff(0) -> ?ACCEPT_BACKOFF_INITIAL; next_backoff(Backoff) -> min(Backoff * 2, ?ACCEPT_BACKOFF_MAX). -spec handle_accepted(term(), #state{}) -> ok. handle_accepted( Accepted, #state{module = Mod, name = Name, counter = Counter, opts = Opts} = State ) -> case nhttp_listener_counter:acquire(Counter) of ok -> ok = Mod:spawn_conn(Accepted, spawn_ctx(State)); {error, {server, #{type := at_capacity}}} -> ok = Mod:reject(Accepted), OtelConfig = nhttp_otel:config(Opts), nhttp_otel:connection_rejected(OtelConfig, #{ listener_name => Name, reason => at_capacity }), ok end. -spec receive_system_msg(pid(), [sys:debug_option()], #state{}, timeout()) -> {ok, [sys:debug_option()], #state{}} | {stop, term()}. receive_system_msg(Parent, Debug, State, Timeout) -> receive {system, From, Request} -> sys:handle_system_msg(Request, From, Parent, ?MODULE, Debug, State), {ok, Debug, State}; {'EXIT', Parent, Reason} -> {stop, Reason}; {get_port, From, Ref} -> From ! {port_reply, Ref, State#state.port}, {ok, Debug, State}; stop_accepting -> {stop, normal} after Timeout -> {ok, Debug, State} end. -spec spawn_ctx(#state{}) -> spawn_ctx(). spawn_ctx(#state{ name = Name, counter = Counter, opts = Opts, conn_sup = ConnSup, tracker = Tracker }) -> #{ name => Name, counter => Counter, opts => Opts, conn_sup => ConnSup, tracker => Tracker }.