fix quic的状态消息处理

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}
       ]}

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}
       ]}

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}),
+    case quicer:async_accept_stream(Conn, #{active => StreamActiveN}) of
-    {next_state, waiting_stream, State};
+        {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}) ->
+handle_event(info, {quic, new_stream, Stream, Opts}, _StateName, State) ->
-    {stop, connection_closed, 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}) ->
+handle_event(info, {quic, passive, Stream, _Props}, _StateName, State = #state{stream = Stream, stream_active_n = StreamActiveN}) ->
-    {stop, transport_shutdown, State};
+    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(<<Buf/binary, Data/binary>>, 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, <<?PACKET_REGISTER_SUPER, Body/binary>>}, 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, <<?PACKET_UNREGISTER>>}, 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}) ->
+handle_event(info, {timeout, TimerRef, ping_ticker}, _, State = #state{ping_timer = TimerRef, client_id = ClientId, ping_counter = PingCounter}) ->
-    %% 等待下一次的心跳检测
-    erlang:start_timer(?PING_TICKER, self(), ping_ticker),
     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, <<?PACKET_EVENT, Event/binary>>),
@@ -371,6 +421,9 @@ handle_event(cast, {command, Ref, ReceiverPid, SubCommand}, registered, State=#s
     quic_send(Stream, <<?PACKET_COMMAND, CommandPkt/binary>>),
     {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, <<Len:16, Packet/binary>>) 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.

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);
+                    loop_accept(L, Limits, AcceptorId);
-                {error, _} ->
+                {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]),
+            logger:debug("[sdlan_quic_server] acceptor: ~p, accept failed: ~p", [AcceptorId, Reason]),
-            loop_accept(L, Limits)
+            loop_accept(L, Limits, AcceptorId)
     end.

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)).