From 6c204d8c3627527adc1438c16449e59d82e41f69 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sat, 18 Apr 2026 17:18:31 +0800 Subject: [PATCH] fix ws_channel --- src/host/iot_host.erl | 25 +++++- src/transport/http/container_handler.erl | 16 ++-- src/transport/tcp/tcp_channel.erl | 104 +++++++++++++++++++---- 3 files changed, 117 insertions(+), 28 deletions(-) diff --git a/src/host/iot_host.erl b/src/host/iot_host.erl index e2b99c8..78bc763 100644 --- a/src/host/iot_host.erl +++ b/src/host/iot_host.erl @@ -26,7 +26,7 @@ -export([get_metric/1, get_status/1, kill/1]). %% 通讯相关 -export([pub/4, attach_channel/2, command/3]). --export([deploy_container/3, start_container/2, stop_container/2, remove_container/2, kill_container/2, config_container/3, get_containers/1, await_reply/2]). +-export([deploy_container/3, start_container/2, stop_container/2, remove_container/2, kill_container/2, config_container/3, get_containers/1, await_reply/3]). -export([heartbeat/1]). %% gen_statem callbacks @@ -127,8 +127,8 @@ remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) Params = #{<<"container_name">> => ContainerName}, gen_statem:call(Pid, {jsonrpc_call, self(), {<<"remove_container">>, Params}}). --spec await_reply(Ref :: reference(), Timeout :: integer()) -> {ok, Result :: binary()} | {error, Reason :: binary()}. -await_reply(Ref, Timeout) when is_reference(Ref), is_integer(Timeout) -> +-spec await_reply(Pid :: pid(), Ref :: reference(), Timeout :: integer()) -> {ok, Result :: binary()} | {error, Reason :: binary()}. +await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_reference(Ref), is_integer(Timeout) -> receive {jsonrpc_reply, Ref, #'JsonRpcReply'{result = ResultBin, error = <<>>}} -> {ok, erlang:binary_to_term(iolist_to_binary(ResultBin))}; @@ -136,6 +136,8 @@ await_reply(Ref, Timeout) when is_reference(Ref), is_integer(Timeout) -> #{<<"message">> := Message} = erlang:binary_to_term(iolist_to_binary(ErrorBin)), {error, Message} after Timeout -> + ok = gen_statem:call(Pid, {cancel_jsonrpc_call, Ref}), + flush_reply(Ref), {error, <<"timeout">>} end. @@ -225,6 +227,15 @@ handle_event({call, From}, {jsonrpc_call, ReceiverPid, RpcCall}, _, State = #sta {keep_state, State, [{reply, From, {error, <<"主机离线，发送请求失败"/utf8>>}}]} end; +handle_event({call, From}, {cancel_jsonrpc_call, Ref}, _, State = #state{channel_pid = ChannelPid}) -> + case is_pid(ChannelPid) of + true -> + ok = tcp_channel:cancel_jsonrpc_call(ChannelPid, Ref), + {keep_state, State, [{reply, From, ok}]}; + false -> + {keep_state, State, [{reply, From, ok}]} + end; + %% 发送指令时, pub/sub handle_event({call, From}, {pub, Topic, Qos, Content}, ?STATE_ACTIVATED, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) -> case HasSession andalso is_pid(ChannelPid) of @@ -384,3 +395,11 @@ state_map(#state{host_id = HostId, uuid = UUID, has_session = HasSession, heartb channel_pid => ChannelPid, metrics => Metrics }. + +flush_reply(Ref) -> + receive + {jsonrpc_reply, Ref, _Reply} -> + ok + after 0 -> + ok + end. diff --git a/src/transport/http/container_handler.erl b/src/transport/http/container_handler.erl index 13e3e79..1ea7980 100644 --- a/src/transport/http/container_handler.erl +++ b/src/transport/http/container_handler.erl @@ -23,7 +23,7 @@ handle_request("GET", "/container/get_all", #{<<"uuid">> := UUID}, _) when is_bi Pid when is_pid(Pid) -> case iot_host:get_containers(Pid) of {ok, Ref} -> - case iot_host:await_reply(Ref, ?REQ_TIMEOUT) of + case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, iot_util:json_data(Result)}; {error, Reason} -> @@ -48,7 +48,7 @@ handle_request("POST", "/container/push_config", _, Timeout = Timeout0 * 1000, case iot_host:config_container(Pid, ContainerName, Config) of {ok, Ref} -> - case iot_host:await_reply(Ref, Timeout) of + case iot_host:await_reply(Pid, Ref, Timeout) of {ok, Result} -> {ok, 200, iot_util:json_data(Result)}; {error, Reason} -> @@ -71,7 +71,7 @@ handle_request("POST", "/container/deploy", _, #{<<"uuid">> := UUID, <<"task_id" Pid when is_pid(Pid) -> case iot_host:deploy_container(Pid, TaskId, Config) of {ok, Ref} -> - case iot_host:await_reply(Ref, ?REQ_TIMEOUT) of + case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, iot_util:json_data(Result)}; {error, Reason} -> @@ -94,7 +94,7 @@ handle_request("POST", "/container/start", _, #{<<"uuid">> := UUID, <<"container Pid when is_pid(Pid) -> case iot_host:start_container(Pid, ContainerName) of {ok, Ref} -> - case iot_host:await_reply(Ref, ?REQ_TIMEOUT) of + case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, iot_util:json_data(Result)}; {error, Reason} -> @@ -113,7 +113,7 @@ handle_request("POST", "/container/stop", _, #{<<"uuid">> := UUID, <<"container_ Pid when is_pid(Pid) -> case iot_host:stop_container(Pid, ContainerName) of {ok, Ref} -> - case iot_host:await_reply(Ref, ?REQ_TIMEOUT) of + case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, iot_util:json_data(Result)}; {error, Reason} -> @@ -131,7 +131,7 @@ handle_request("POST", "/container/kill", _, #{<<"uuid">> := UUID, <<"container_ Pid when is_pid(Pid) -> case iot_host:kill_container(Pid, ContainerName) of {ok, Ref} -> - case iot_host:await_reply(Ref, ?REQ_TIMEOUT) of + case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, iot_util:json_data(Result)}; {error, Reason} -> @@ -150,7 +150,7 @@ handle_request("POST", "/container/remove", _, #{<<"uuid">> := UUID, <<"containe Pid when is_pid(Pid) -> case iot_host:remove_container(Pid, ContainerName) of {ok, Ref} -> - case iot_host:await_reply(Ref, ?REQ_TIMEOUT) of + case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, iot_util:json_data(Result)}; {error, Reason} -> @@ -306,4 +306,4 @@ check_type(Value, {map, {binary, any}}) when is_map(Value) -> check_type(Value, boolean) -> is_boolean(Value); check_type(_, _) -> - false. \ No newline at end of file + false. diff --git a/src/transport/tcp/tcp_channel.erl b/src/transport/tcp/tcp_channel.erl index cfb1a19..8de4115 100644 --- a/src/transport/tcp/tcp_channel.erl +++ b/src/transport/tcp/tcp_channel.erl @@ -12,8 +12,11 @@ -include("message_pb.hrl"). -behaviour(ranch_protocol). +-define(MAX_PACKET_ID, 16#FFFFFFFF). +-define(INFLIGHT_TIMEOUT, 60000). + %% API --export([pub/4, jsonrpc_call/3, command/3]). +-export([pub/4, jsonrpc_call/3, cancel_jsonrpc_call/2, command/3]). -export([start_link/3, stop/2]). %% gen_server callbacks @@ -50,6 +53,10 @@ jsonrpc_call(Pid, ReceiverPid, Request = {Method, _Params}) when is_pid(Pid), is gen_server:cast(Pid, {jsonrpc_call, ReceiverPid, Ref, Request}), Ref. +-spec cancel_jsonrpc_call(Pid :: pid(), Ref :: reference()) -> ok. +cancel_jsonrpc_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) -> + gen_server:call(Pid, {cancel_jsonrpc_call, Ref}). + %% 关闭方法 -spec stop(Pid :: pid(), Reason :: any()) -> no_return(). stop(undefined, _Reason) -> @@ -72,19 +79,27 @@ init(Ref, Transport, _Opts = []) -> % erlang:start_timer(?PING_TICKER, self(), ping_ticker), gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket}). +handle_call({cancel_jsonrpc_call, Ref}, _From, State = #state{inflight = Inflight}) -> + case take_inflight_by_ref(Ref, Inflight) of + {ok, _PacketId, {_ReceiverPid, _Ref, TimerRef}, NInflight} -> + erlang:cancel_timer(TimerRef), + {reply, ok, State#state{inflight = NInflight}}; + error -> + {reply, ok, State} + end; handle_call(_Request, _From, State) -> {reply, ok, State}. %% 发送消息, 基于pub/sub机制 handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) -> - Encoded = message_pb:encode_msg(#'Pub'{topic = Topic, qos = Qos, content = Content}, 'Pub'), + Encoded = message_pb:encode_msg(#'Pub'{topic = Topic, qos = Qos, content = Content}), EncPub = <>, Transport:send(Socket, <>), {noreply, State}; %% 发送Command消息 handle_cast({command, CommandType, Command}, State = #state{transport = Transport, socket = Socket}) -> - Encoded = message_pb:encode_msg(#'Command'{command_type = CommandType, command = Command}, 'Command'), + Encoded = message_pb:encode_msg(#'Command'{command_type = CommandType, command = Command}), EncCommand = <>, Transport:send(Socket, <>), {noreply, State}; @@ -92,20 +107,29 @@ handle_cast({command, CommandType, Command}, State = #state{transport = Transpor %% 推送消息 handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) when is_binary(Method) -> - Request = #'JsonRpcRequest'{method = Method, params = erlang:term_to_binary(Params)}, - Encoded = message_pb:encode_msg(Request, 'JsonRpcRequest'), - EncRequest = <>, - Transport:send(Socket, <>), - {noreply, State#state{packet_id = PacketId + 1, inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}}. + case next_packet_id(PacketId, Inflight) of + {ok, NPacketId, NextPacketId} -> + Encoded = message_pb:encode_msg(#'JsonRpcRequest'{ + method = Method, + params = erlang:term_to_binary(Params) + }), + + EncRequest = <>, + TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {jsonrpc_timeout, NPacketId}), + Transport:send(Socket, <>), + {noreply, State#state{ + packet_id = NextPacketId, + inflight = maps:put(NPacketId, {ReceiverPid, Ref, TimerRef}, Inflight) + }}; + {error, inflight_full} -> + logger:warning("[ws_channel] uuid: ~p, inflight requests exhausted", [State#state.uuid]), + {noreply, State} + end. %% auth验证 handle_info({tcp, Socket, <>}, State = #state{transport = Transport, socket = Socket}) -> - #'AuthRequest'{uuid = UUID0, username = Username0, token = Token0, salt = Salt0, timestamp = Timestamp} = - message_pb:decode_msg(RequestBin, 'AuthRequest'), - UUID = iolist_to_binary(UUID0), - Username = iolist_to_binary(Username0), - Token = iolist_to_binary(Token0), - Salt = iolist_to_binary(Salt0), + #'AuthRequest'{uuid = UUID, username = Username, token = Token, salt = Salt, timestamp = Timestamp} = message_pb:decode_msg(RequestBin, 'AuthRequest'), + logger:debug("[ws_channel] auth uuid: ~p", [UUID]), case iot_auth:check(Username, Token, UUID, Salt, Timestamp) of true -> @@ -180,13 +204,14 @@ handle_info({tcp, Socket, <>}, State = #state % {noreply, State}; %% 主机端的消息响应 -handle_info({tcp, Socket, <>}, State = #state{socket = Socket, inflight = Inflight}) when PacketId > 0 -> - <> = ResponseBin, +handle_info({tcp, Socket, <>}, State = #state{socket = Socket, inflight = Inflight}) when PacketId > 0 -> RpcReply = message_pb:decode_msg(ReplyBin, 'JsonRpcReply'), + case maps:take(PacketId, Inflight) of error -> {noreply, State}; - {{ReceiverPid, Ref}, NInflight} -> + {{ReceiverPid, Ref, TimerRef}, NInflight} -> + erlang:cancel_timer(TimerRef), case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of true -> ReceiverPid ! {jsonrpc_reply, Ref, RpcReply}; @@ -196,6 +221,15 @@ handle_info({tcp, Socket, <>} {noreply, State#state{inflight = NInflight}} end; +handle_info({timeout, TimerRef, {jsonrpc_timeout, PacketId}}, State = #state{inflight = Inflight}) -> + case maps:get(PacketId, Inflight, undefined) of + {_ReceiverPid, Ref, TimerRef} -> + logger:warning("[ws_channel] jsonrpc request timeout, packet_id: ~p, ref: ~p", [PacketId, Ref]), + {noreply, State#state{inflight = maps:remove(PacketId, Inflight)}}; + _ -> + {noreply, State} + end; + handle_info({tcp_error, Sock, Reason}, State = #state{socket = Sock}) -> logger:notice("[sdlan_channel] tcp_error: ~p", [Reason]), {stop, normal, State}; @@ -223,3 +257,39 @@ terminate(Reason, #state{}) -> code_change(_OldVsn, State, _Extra) -> {ok, State}. + +take_inflight_by_ref(Ref, Inflight) -> + maps:fold(fun(PacketId, Value = {_ReceiverPid, PacketRef, _TimerRef}, Acc) -> + case Acc of + error when PacketRef =:= Ref -> + {ok, PacketId, Value, maps:remove(PacketId, Inflight)}; + _ -> + Acc + end + end, error, Inflight). + +next_packet_id(_PacketId, Inflight) when map_size(Inflight) >= ?MAX_PACKET_ID -> + {error, inflight_full}; +next_packet_id(PacketId, Inflight) -> + next_packet_id(PacketId, Inflight, PacketId, 0). + +next_packet_id(_PacketId, _Inflight, _StartPacketId, TryCount) when TryCount > ?MAX_PACKET_ID -> + {error, inflight_full}; +next_packet_id(PacketId, Inflight, StartPacketId, TryCount) -> + case maps:is_key(PacketId, Inflight) of + false -> + {ok, PacketId, inc_packet_id(PacketId)}; + true -> + NPacketId = inc_packet_id(PacketId), + case NPacketId =:= StartPacketId of + true -> + {error, inflight_full}; + false -> + next_packet_id(NPacketId, Inflight, StartPacketId, TryCount + 1) + end + end. + +inc_packet_id(?MAX_PACKET_ID) -> + 1; +inc_packet_id(PacketId) when PacketId > 0, PacketId < ?MAX_PACKET_ID -> + PacketId + 1.