From 48cf77b0d7613d4c957b4e2e541f6061bb873fe0 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Mon, 27 Apr 2026 11:22:30 +0800 Subject: [PATCH] =?UTF-8?q?fix=20quic=E7=9A=84=E7=8A=B6=E6=80=81=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/sys-dev.config | 2 + config/sys-prod.config | 2 + src/quic/sdlan_quic_channel.erl | 147 ++++++++++++++++++++++++++++---- src/quic/sdlan_quic_server.erl | 48 ++++++++--- src/sdlan_network.erl | 38 ++++++++- 5 files changed, 207 insertions(+), 30 deletions(-) diff --git a/config/sys-dev.config b/config/sys-dev.config index cb34008..c718564 100644 --- a/config/sys-dev.config +++ b/config/sys-dev.config @@ -10,11 +10,13 @@ {quic_server, [ {port, 443}, + {acceptors, 10}, {alpn, ["punchnet/1.0"]}, {certfile, "cert.pem"}, {keyfile, "key.pem"}, {limits, [ {max_packet_size, 16384}, + {stream_active_n, 100}, %% 单位为秒 {heartbeat_sec, 5} ]} diff --git a/config/sys-prod.config b/config/sys-prod.config index 4681e83..288e80b 100644 --- a/config/sys-prod.config +++ b/config/sys-prod.config @@ -10,11 +10,13 @@ {quic_server, [ {port, 443}, + {acceptors, 10}, {alpn, ["punchnet/1.0"]}, {certfile, "fullchain.cer"}, {keyfile, "root.punchsky.com.key"}, {limits, [ {max_packet_size, 16384}, + {stream_active_n, 100}, %% 单位为秒 {heartbeat_sec, 5} ]} diff --git a/src/quic/sdlan_quic_channel.erl b/src/quic/sdlan_quic_channel.erl index 90ee0d3..ad80fda 100644 --- a/src/quic/sdlan_quic_channel.erl +++ b/src/quic/sdlan_quic_channel.erl @@ -15,6 +15,8 @@ %% 心跳包监测机制 -define(PING_TICKER, 15000). +-define(STREAM_ACTIVE_N, 100). +-define(METRICS_TICKER, 60000). %% 注册失败的的错误码 @@ -25,7 +27,7 @@ %% API -export([start_link/2]). --export([send_event/2, command/4, stop/2]). +-export([accept_stream/1, send_event/2, command/4, stop/2, debug_info/1]). -export([test_rules/2]). %% gen_statem callbacks @@ -37,6 +39,9 @@ max_packet_size = 16384, %% 心跳间隔 heartbeat_sec = 10, + ping_timer :: undefined | reference(), + metrics_timer :: undefined | reference(), + stream_active_n = ?STREAM_ACTIVE_N, stream :: undefined | quicer:stream_handle(), %% 累积器，用于处理协议framing的解析 @@ -56,6 +61,8 @@ pending_commands = #{}, ping_counter = 0, + frames_recv = 0, + bytes_recv = 0, %% 离线回调函数 offline_cb :: undefined | fun() @@ -79,10 +86,16 @@ send_event(Pid, ProtobufEvent) when is_pid(Pid), is_binary(ProtobufEvent) -> command(Pid, Ref, ReceiverPid, SubCommand) when is_pid(Pid), is_pid(ReceiverPid) -> gen_statem:cast(Pid, {command, Ref, ReceiverPid, SubCommand}). +accept_stream(Pid) when is_pid(Pid) -> + gen_statem:cast(Pid, accept_stream). + -spec stop(Pid :: pid(), Reason :: term()) -> ok. stop(Pid, Reason) when is_pid(Pid) -> gen_statem:stop(Pid, Reason, 2000). +debug_info(Pid) when is_pid(Pid) -> + gen_statem:call(Pid, debug_info). + %% @doc Creates a gen_statem process which calls Module:init/1 to %% initialize. To ensure a synchronized start-up procedure, this %% function does not return until Module:init/1 has returned. @@ -100,7 +113,8 @@ start_link(Conn, Limits) when is_list(Limits) -> init([Conn, Limits]) -> MaxPacketSize = proplists:get_value(max_packet_size, Limits, 16384), HeartbeatSec = proplists:get_value(heartbeat_sec, Limits, 10), - {ok, initializing, #state{conn = Conn, max_packet_size = MaxPacketSize, heartbeat_sec = HeartbeatSec}, [{next_event, internal, do_init}]}. + StreamActiveN = proplists:get_value(stream_active_n, Limits, ?STREAM_ACTIVE_N), + {ok, initializing, #state{conn = Conn, max_packet_size = MaxPacketSize, heartbeat_sec = HeartbeatSec, stream_active_n = StreamActiveN}}. %% @private %% @doc This function is called by a gen_statem when it needs to find out @@ -113,10 +127,14 @@ callback_mode() -> %% gen_statem receives an event from call/2, cast/2, or as a normal %% process message, this function is called. -handle_event(internal, do_init, initializing, State=#state{conn = Conn}) -> +handle_event(cast, accept_stream, initializing, State=#state{conn = Conn, stream_active_n = StreamActiveN}) -> logger:debug("[sdlan_quic_channel] call do_init of conn: ~p", [Conn]), - {ok, _} = quicer:async_accept_stream(Conn, #{active => true}), - {next_state, waiting_stream, State}; + case quicer:async_accept_stream(Conn, #{active => StreamActiveN}) of + {ok, _} -> + {next_state, waiting_stream, schedule_metrics(State)}; + {error, Reason} -> + {stop, {accept_stream_failed, Reason}, State} + end; %% 处理收到的quic消息 handle_event(info, {quic, dgram_state_changed, Conn, Opts = #{dgram_send_enabled := true}}, _, State=#state{conn = Conn}) -> @@ -147,24 +165,48 @@ handle_event(info, {quic, new_stream, Stream, Opts}, waiting_stream, State=#stat {next_state, initialized, State#state{stream = Stream}}; -handle_event(info, {quic, closed, Stream, _Props}, _StateName, State = #state{stream = Stream}) -> - {stop, connection_closed, State}; +handle_event(info, {quic, new_stream, Stream, Opts}, _StateName, State) -> + logger:warning("[sdlan_quic_channel] reject unexpected stream: ~p, opts: ~p", [Stream, Opts]), + quicer:close_stream(Stream, 1000), + {keep_state, State}; + +handle_event(info, {quic, stream_closed, Stream, Props}, _StateName, State = #state{stream = Stream}) -> + {stop, {stream_closed, Props}, State}; + +handle_event(info, {quic, peer_send_shutdown, Stream, _Props}, _StateName, State = #state{stream = Stream}) -> + {stop, peer_send_shutdown, State}; + +handle_event(info, {quic, peer_send_aborted, Stream, ErrorCode}, _StateName, State = #state{stream = Stream}) -> + {stop, {peer_send_aborted, ErrorCode}, State}; + +handle_event(info, {quic, peer_receive_aborted, Stream, ErrorCode}, _StateName, State = #state{stream = Stream}) -> + {stop, {peer_receive_aborted, ErrorCode}, State}; handle_event(info, {quic, send_shutdown_complete, Stream, _Props}, _StateName, State = #state{stream = Stream}) -> {stop, connection_shutdown, State}; -handle_event(info, {quic, transport_shutdown, Stream, _Props}, _StateName, State = #state{stream = Stream}) -> - {stop, transport_shutdown, State}; +handle_event(info, {quic, passive, Stream, _Props}, _StateName, State = #state{stream = Stream, stream_active_n = StreamActiveN}) -> + ok = quicer:setopt(Stream, active, StreamActiveN), + {keep_state, State}; + +handle_event(info, {quic, closed, Conn, Props}, _StateName, State = #state{conn = Conn}) -> + {stop, {connection_closed, Props}, State}; + +handle_event(info, {quic, transport_shutdown, Conn, Props}, _StateName, State = #state{conn = Conn}) -> + {stop, {transport_shutdown, Props}, State}; + +handle_event(info, {quic, shutdown, Conn, ErrorCode}, _StateName, State = #state{conn = Conn}) -> + {stop, {connection_shutdown_by_peer, ErrorCode}, State}; %% 处理quicer相关的信息, 需要转换成内部能够识别的frame消息 -handle_event(info, {quic, Data, Stream, _Props}, _StateName, State = #state{stream = Stream, buf = Buf, max_packet_size = MaxPacketSize}) when is_binary(Data) -> +handle_event(info, {quic, Data, Stream, _Props}, _StateName, State = #state{stream = Stream, buf = Buf, max_packet_size = MaxPacketSize, bytes_recv = BytesRecv, frames_recv = FramesRecv}) when is_binary(Data) -> case decode_frames(<>, MaxPacketSize) of {error, Reason} -> {stop, Reason, State}; {ok, NBuf, Frames} -> Actions = [{next_event, internal, {frame, Frame}} || Frame <- Frames], %logger:debug("[sdlan_quic_channel] get frames: ~p", [Frames]), - {keep_state, State#state{buf = NBuf}, Actions} + {keep_state, State#state{buf = NBuf, bytes_recv = BytesRecv + byte_size(Data), frames_recv = FramesRecv + length(Frames)}, Actions} end; %% 处理内部的包消息 @@ -225,7 +267,7 @@ handle_event(internal, {frame, <>}, initial <<"status">> => 0 }) end, - {next_state, registered, State#state{network_id = NetworkId, network_pid = NetworkPid, client_id = ClientId, mac = Mac, ip = Ip, offline_cb = OfflineCb}}; + {next_state, registered, schedule_ping(State#state{network_id = NetworkId, network_pid = NetworkPid, client_id = ClientId, mac = Mac, ip = Ip, offline_cb = OfflineCb})}; undefined -> logger:warning("[sdlan_quic_channel] client_id: ~p, register get error: network not found", [ClientId]), quic_send(Stream, register_nak_reply(?NAK_INTERNAL_FAULT, <<"Internal Error">>)), @@ -344,17 +386,25 @@ handle_event(internal, {frame, <>}, registered, State=#state sdlan_network:unregister(NetworkPid, ClientId, Mac), {stop, normal, State}; -handle_event(info, {timeout, _, ping_ticker}, _, State = #state{client_id = ClientId, ping_counter = PingCounter}) -> - %% 等待下一次的心跳检测 - erlang:start_timer(?PING_TICKER, self(), ping_ticker), +handle_event(info, {timeout, TimerRef, ping_ticker}, _, State = #state{ping_timer = TimerRef, client_id = ClientId, ping_counter = PingCounter}) -> case PingCounter > 0 of true -> - {keep_state, State#state{ping_counter = 0}}; + {keep_state, schedule_ping(State#state{ping_counter = 0, ping_timer = undefined})}; false -> logger:debug("[sdlan_channel] client_id: ~p, ping losted", [ClientId]), - {stop, normal, State#state{ping_counter = 0}} + {stop, heartbeat_timeout, State#state{ping_counter = 0, ping_timer = undefined}} end; +handle_event(info, {timeout, _TimerRef, ping_ticker}, _, State) -> + {keep_state, State}; + +handle_event(info, {timeout, TimerRef, metrics_ticker}, StateName, State = #state{metrics_timer = TimerRef}) -> + log_metrics(StateName, State), + {keep_state, schedule_metrics(State#state{metrics_timer = undefined})}; + +handle_event(info, {timeout, _TimerRef, metrics_ticker}, _StateName, State) -> + {keep_state, State}; + %% 发送指令信息 handle_event(cast, {send_event, Event}, registered, #state{stream = Stream}) -> quic_send(Stream, <>), @@ -371,6 +421,9 @@ handle_event(cast, {command, Ref, ReceiverPid, SubCommand}, registered, State=#s quic_send(Stream, <>), {keep_state, State#state{pkt_id = PktId + 1, pending_commands = maps:put(PktId, {Ref, ReceiverPid}, PendingCommands)}}; +handle_event({call, From}, debug_info, StateName, State) -> + {keep_state, State, [{reply, From, debug_info(StateName, State)}]}; + handle_event(info, {'EXIT', _, _}, _StateName, State) -> {stop, connection_closed, State}; @@ -384,7 +437,7 @@ handle_event(EventType, Info, StateName, State) -> %% necessary cleaning up. When it returns, the gen_statem terminates with %% Reason. The return value is ignored. terminate(Reason, _StateName, _State = #state{conn = Conn, stream = Stream, offline_cb = OfflineCb}) -> - Stream /= undefined andalso quicer:close_stream(Stream), + Stream /= undefined andalso quicer:close_stream(Stream, 1000), quicer:close_connection(Conn), logger:warning("[sdlan_quic_conn] terminate closed with reason: ~p", [Reason]), %% 触发客户端的离线逻辑 @@ -427,8 +480,11 @@ rsa_encode(PlainText, RsaPubKey) when is_binary(PlainText) -> -spec quic_send(Stream :: quicer:stream_handle(), Packet :: binary()) -> no_return(). quic_send(Stream, Packet) when is_binary(Packet) -> Len = byte_size(Packet), + true = Len =< 65535, case quicer:send(Stream, <>) of {ok, _} -> + incr_counter(quic_frames_sent, 1), + incr_counter(quic_bytes_sent, Len + 2), ok; {error, Reason} -> exit({quic_send_failed, Reason}) @@ -439,3 +495,58 @@ get_rules(SrcIdentityId, DstIdentityId) when is_integer(SrcIdentityId), is_integ SrcPolicyIds = identity_policy_ets:get_policies(SrcIdentityId), DstPolicyIds = identity_policy_ets:get_policies(DstIdentityId), rule_ets:get_rules(SrcPolicyIds, DstPolicyIds). + +schedule_ping(State = #state{heartbeat_sec = HeartbeatSec}) -> + State#state{ping_timer = erlang:start_timer(heartbeat_ms(HeartbeatSec), self(), ping_ticker)}. + +heartbeat_ms(HeartbeatSec) when is_integer(HeartbeatSec), HeartbeatSec > 0 -> + HeartbeatSec * 1000; +heartbeat_ms(_) -> + ?PING_TICKER. + +schedule_metrics(State = #state{metrics_timer = undefined}) -> + State#state{metrics_timer = erlang:start_timer(?METRICS_TICKER, self(), metrics_ticker)}; +schedule_metrics(State) -> + State. + +log_metrics(StateName, State = #state{client_id = ClientId, network_id = NetworkId}) -> + Info = debug_info(StateName, State), + logger:debug("[sdlan_quic_channel] metrics client_id: ~p, network_id: ~p, metrics: ~p", [ClientId, NetworkId, Info]). + +debug_info(StateName, #state{ + client_id = ClientId, + network_id = NetworkId, + mac = Mac, + ip = Ip, + pending_commands = PendingCommands, + frames_recv = FramesRecv, + bytes_recv = BytesRecv, + stream_active_n = StreamActiveN, + heartbeat_sec = HeartbeatSec +}) -> + ProcInfo = maps:from_list(process_info(self(), [message_queue_len, memory, reductions])), + ProcInfo#{ + state => StateName, + client_id => ClientId, + network_id => NetworkId, + mac => Mac, + ip => Ip, + pending_commands => maps:size(PendingCommands), + frames_recv => FramesRecv, + frames_sent => get_counter(quic_frames_sent), + bytes_recv => BytesRecv, + bytes_sent => get_counter(quic_bytes_sent), + stream_active_n => StreamActiveN, + heartbeat_sec => HeartbeatSec + }. + +incr_counter(Key, Inc) -> + erlang:put(Key, get_counter(Key) + Inc). + +get_counter(Key) -> + case erlang:get(Key) of + undefined -> + 0; + Value when is_integer(Value) -> + Value + end. diff --git a/src/quic/sdlan_quic_server.erl b/src/quic/sdlan_quic_server.erl index c08cf6c..d1b5ad7 100644 --- a/src/quic/sdlan_quic_server.erl +++ b/src/quic/sdlan_quic_server.erl @@ -16,12 +16,14 @@ start_link() -> {ok, spawn_link(?MODULE, init, [])}. init() -> + process_flag(trap_exit, true), {ok, Props} = application:get_env(sdlan, quic_server), Port = proplists:get_value(port, Props), Alpn = proplists:get_value(alpn, Props), Limits = proplists:get_value(limits, Props), CertFile = proplists:get_value(certfile, Props), KeyFile = proplists:get_value(keyfile, Props), + AcceptorCount = proplists:get_value(acceptors, Props, 10), %% 获取环境变量 Path = os:getenv("QUIC_CERT_PATH", code:priv_dir(sdlan)), @@ -33,36 +35,60 @@ init() -> keyfile => Path ++ "/" ++ KeyFile, alpn => Alpn, peer_bidi_stream_count => 1, - conn_acceptors => 10 + conn_acceptors => AcceptorCount }, ListenAddr = "0.0.0.0:" ++ integer_to_list(Port), case quicer:listen(ListenAddr, LOptions) of {ok, L} -> - loop_accept(L, Limits); + logger:notice("[sdlan_quic_server] listen on ~s, acceptors: ~p", [ListenAddr, AcceptorCount]), + Pids = [spawn_link(fun() -> loop_accept(L, Limits, I) end) || I <- lists:seq(1, AcceptorCount)], + wait_acceptors(L, Pids); Error -> exit(Error) end. -loop_accept(L, Limits) -> +wait_acceptors(L, Pids) -> + receive + {'EXIT', Pid, Reason} -> + logger:warning("[sdlan_quic_server] acceptor ~p exit with reason: ~p", [Pid, Reason]), + quicer:close_listener(L), + exit({acceptor_exit, Pid, Reason, Pids}); + stop -> + quicer:close_listener(L), + ok; + Info -> + logger:notice("[sdlan_quic_server] get unexpected info: ~p", [Info]), + wait_acceptors(L, Pids) + end. + +loop_accept(L, Limits, AcceptorId) -> case quicer:accept(L, #{}, infinity) of {ok, Conn} -> - logger:debug("[sdlan_quic_server] accept a new connection: ~p", [Conn]), + logger:debug("[sdlan_quic_server] acceptor: ~p, accept a new connection: ~p", [AcceptorId, Conn]), case quicer:handshake(Conn) of {ok, NConn} -> case sdlan_quic_channel_sup:start_channel(NConn, Limits) of {ok, ChannelPid} -> logger:debug("[sdlan_quic_server] conn: ~p, handshake success, channel pid: ~p", [NConn, ChannelPid]), - quicer:controlling_process(NConn, ChannelPid); + case quicer:controlling_process(NConn, ChannelPid) of + ok -> + sdlan_quic_channel:accept_stream(ChannelPid); + {error, Reason} -> + logger:warning("[sdlan_quic_server] conn: ~p, controlling_process failed: ~p", [NConn, Reason]), + sdlan_quic_channel:stop(ChannelPid, {controlling_process_failed, Reason}), + quicer:close_connection(NConn) + end; Error -> quicer:close_connection(NConn), logger:notice("[sdlan_quic_server] start channel get error: ~p", [Error]) end, - loop_accept(L, Limits); - {error, _} -> + loop_accept(L, Limits, AcceptorId); + {error, Reason} -> + logger:debug("[sdlan_quic_server] acceptor: ~p, handshake failed: ~p", [AcceptorId, Reason]), quicer:close_connection(Conn), - loop_accept(L, Limits) + loop_accept(L, Limits, AcceptorId) end; {error, Reason} -> - logger:debug("[sdlan_quic_server] accept failed: ~p", [Reason]), - loop_accept(L, Limits) - end. \ No newline at end of file + logger:debug("[sdlan_quic_server] acceptor: ~p, accept failed: ~p", [AcceptorId, Reason]), + loop_accept(L, Limits, AcceptorId) + end. diff --git a/src/sdlan_network.erl b/src/sdlan_network.erl index 05c7069..fe67914 100644 --- a/src/sdlan_network.erl +++ b/src/sdlan_network.erl @@ -326,6 +326,7 @@ handle_call(debug_info, _From, State = #state{network_id = NetworkId, ipaddr = I <<"ipaddr">> => IpAddr, <<"mask_len">> => MaskLen, <<"owner_id">> => OwnerId, + <<"metrics">> => network_metrics(State), <<"used_ips">> => lists:map(fun format_endpoint/1, maps:to_list(Endpoints)) }, {reply, Reply, State}. @@ -436,6 +437,7 @@ handle_cast({update_hole, SessionToken, ClientId, Mac, Peer, NatType, V6Info}, S handle_info({timeout, _, flow_report_ticker}, State = #state{network_id = NetworkId, forward_bytes = ForwardBytes}) -> erlang:start_timer(?FLOW_REPORT_INTERVAL, self(), flow_report_ticker), catch sdlan_api:network_forward_report(NetworkId, ForwardBytes), + logger:debug("[sdlan_network] metrics: ~p", [network_metrics(State)]), {noreply, State#state{forward_bytes = 0}}; %% Channel进程退出, hole里面的数据也需要清理 @@ -541,8 +543,42 @@ format_endpoint({Mac, #endpoint{client_id = ClientId, ip = Ip, hole = #hole{peer ip => sdlan_util:int_to_ipv4(Ip), hole_map => HoleMap, v6_info => V6InfoMap + }; +format_endpoint({Mac, #endpoint{client_id = ClientId, ip = Ip, hole = undefined, v6_info = V6Info}}) -> + V6InfoMap = case V6Info of + undefined -> + #{}; + #'SDLV6Info'{v6 = V6, port = V6Port} -> + #{v6 => V6, port => V6Port} + end, + #{ + client_id => ClientId, + mac => sdlan_util:format_mac(Mac), + ip => sdlan_util:int_to_ipv4(Ip), + hole_map => undefined, + v6_info => V6InfoMap }. +network_metrics(#state{network_id = NetworkId, endpoints = Endpoints, forward_bytes = ForwardBytes}) -> + ProcInfo = maps:from_list(process_info(self(), [message_queue_len, memory, reductions])), + ProcInfo#{ + network_id => NetworkId, + endpoint_count => maps:size(Endpoints), + forward_bytes => ForwardBytes, + channel_metrics => channel_metrics(Endpoints) + }. + +channel_metrics(Endpoints) -> + maps:fold(fun(_Mac, #endpoint{channel_pid = ChannelPid}, Acc) -> + case is_pid(ChannelPid) andalso is_process_alive(ChannelPid) of + true -> + ProcInfo = maps:from_list(process_info(ChannelPid, [message_queue_len, memory])), + [ProcInfo#{pid => ChannelPid} | Acc]; + false -> + Acc + end + end, [], Endpoints). + -spec search_endpoint(F :: fun((term(), term()) -> boolean()), Endpoints :: map()) -> error | {ok, Key :: any(), Val :: any()}. search_endpoint(F, Endpoints) when is_function(F, 2), is_map(Endpoints) -> search_endpoint0(F, maps:iterator(Endpoints)). @@ -599,4 +635,4 @@ time33([], Hash) -> time33([C|Rest], Hash) -> %% hash = hash * 33 + char NewHash = ((Hash bsl 5) + Hash) + C, - time33(Rest, NewHash). \ No newline at end of file + time33(Rest, NewHash).