From 3b1e4500d7d973c3019f0535c26f04db68c905c0 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Wed, 22 Apr 2026 16:52:40 +0800 Subject: [PATCH] fix --- config/sys-dev.config | 13 +-- config/sys-prod.config | 22 ++--- src/endpoint/endpoint_buffer.erl | 157 ++++++++++++++----------------- src/endpoint/endpoint_http.erl | 26 ++--- src/endpoint/endpoint_timer.erl | 20 ++-- 5 files changed, 100 insertions(+), 138 deletions(-) diff --git a/config/sys-dev.config b/config/sys-dev.config index 5e82b3f..64f3f9f 100644 --- a/config/sys-dev.config +++ b/config/sys-dev.config @@ -28,6 +28,7 @@ {api_url, "http://127.0.0.1:18090/simulator"}, %% 支持的协议 {endpoints, [ + {root_dir, "/usr/local/code/database/"}, {support_protocols, [ http ]} @@ -49,18 +50,6 @@ {<<"test">>, <<"iot2023">>} ]} - %{pools, [ - % %% redis连接池 - % {redis_pool, - % [{size, 10}, {max_overflow, 20}, {worker_module, eredis}], - % [ - % {host, "127.0.0.1"}, - % {port, 6379}, - % {database, 1} - % ] - % } - %]} - ]}, %% 系统日志配置，使用 OTP logger diff --git a/config/sys-prod.config b/config/sys-prod.config index 37c5485..63d9332 100644 --- a/config/sys-prod.config +++ b/config/sys-prod.config @@ -21,6 +21,13 @@ {backlog, 10240} ]}, + {endpoints, [ + {root_dir, "/usr/local/code/database/"}, + {support_protocols, [ + http + ]} + ]}, + {udp_server, [ {port, 18080} ]}, @@ -37,23 +44,8 @@ {port, 8086}, {token, <<"A-ZRjqMK_7NR45lXXEiR7AEtYCd1ETzq9Z61FTMQLb5O4-1hSf8sCrjdPB84e__xsrItKHL3qjJALgbYN-H_VQ==">>} ]} - - %{pools, [ - % %% redis连接池 - % {redis_pool, - % [{size, 10}, {max_overflow, 20}, {worker_module, eredis}], - % [ - % {host, "172.30.6.175"}, - % {port, 26379}, - % {database, 1} - % ] - % } - - %]} - ]}, - %% 系统日志配置，使用 OTP logger {kernel, [ %% 设置 Logger 的 primary log level diff --git a/src/endpoint/endpoint_buffer.erl b/src/endpoint/endpoint_buffer.erl index d69ce51..9a29485 100644 --- a/src/endpoint/endpoint_buffer.erl +++ b/src/endpoint/endpoint_buffer.erl @@ -19,11 +19,7 @@ -record(buffer, { endpoint :: #endpoint{}, - next_id = 1 :: integer(), - %% 最近一次派发的数据游标 - cursor = 0 :: integer(), - %% ets存储的引用 - tid :: ets:tid(), + outbox :: endpoint_outbox:outbox(), %% 定时器 timer_pid :: pid(), %% 窗口大小，允许最大的未确认消息数 @@ -34,34 +30,38 @@ acc_num = 0 }). --record(north_data, { - id :: integer(), - tuple :: any(), - inflight = false :: boolean() -}). - -type buffer() :: #buffer{}. -spec new(Endpoint :: #endpoint{}, WindowSize :: integer()) -> Buffer :: #buffer{}. new(Endpoint = #endpoint{id = Id}, WindowSize) when is_integer(WindowSize), WindowSize > 0 -> - %% 初始化存储 - EtsName = list_to_atom("endpoint_buffer_ets:" ++ integer_to_list(Id)), - Tid = ets:new(EtsName, [ordered_set, private, {keypos, 2}]), + %% 读取配置文件 + {ok, Endpoints} = application:get_env(iot, endpoints), + RootDir = proplists:get_value(root_dir, Endpoints), + OutboxDir = filename:join(RootDir, integer_to_list(Id)), + + {ok, Outbox} = endpoint_outbox:open(OutboxDir, #{}), %% 定义重发器 {ok, TimerPid} = endpoint_timer:start_link(?RETRY_INTERVAL), + #buffer{outbox = Outbox, timer_pid = TimerPid, endpoint = Endpoint, window_size = WindowSize}. - #buffer{cursor = 0, tid = Tid, timer_pid = TimerPid, endpoint = Endpoint, window_size = WindowSize}. - --spec append(Tuple :: any(), Buffer :: #buffer{}) -> NBuffer :: #buffer{}. -append(Tuple, Buffer = #buffer{tid = Tid, next_id = NextId, window_size = WindowSize, flight_num = FlightNum}) -> - NorthData = #north_data{id = NextId, tuple = Tuple}, - true = ets:insert(Tid, NorthData), - NBuffer = Buffer#buffer{next_id = NextId + 1}, - case FlightNum < WindowSize of - true -> - trigger_next(NBuffer); - false -> - NBuffer +-spec append(Payload :: binary(), Buffer :: #buffer{}) -> NBuffer :: #buffer{}. +append(Payload, Buffer = #buffer{outbox = Outbox, window_size = WindowSize, flight_num = FlightNum}) when is_binary(Payload) -> + case endpoint_outbox:append(Payload, Outbox) of + {ok, _Seq, NOutbox} -> + NBuffer = Buffer#buffer{outbox = NOutbox}, + case FlightNum < WindowSize of + true -> + trigger_next(NBuffer); + false -> + NBuffer + end; + {dropped, capacity_reached, NOutbox} -> + logger:warning("[endpoint_buffer] outbox capacity reached, endpoint_id: ~p", [buffer_endpoint_id(Buffer)]), + Buffer#buffer{outbox = NOutbox}; + {error, Reason} -> + logger:warning("[endpoint_buffer] append outbox failed, endpoint_id: ~p, reason: ~p", + [buffer_endpoint_id(Buffer), Reason]), + Buffer end. -spec trigger_n(Buffer :: #buffer{}) -> NBuffer :: #buffer{}. @@ -71,40 +71,55 @@ trigger_n(Buffer = #buffer{window_size = WindowSize}) -> %% 触发读取下一条数据 -spec trigger_next(Buffer :: #buffer{}) -> NBuffer :: #buffer{}. -trigger_next(Buffer = #buffer{tid = Tid, cursor = Cursor, timer_pid = TimerPid, flight_num = FlightNum, window_size = WindowSize}) -> +trigger_next(Buffer = #buffer{outbox = Outbox, timer_pid = TimerPid, flight_num = FlightNum, window_size = WindowSize}) -> case FlightNum < WindowSize of false -> Buffer; true -> - case next_pending_data(Tid, Cursor) of - none -> + case endpoint_outbox:next(Outbox) of + eof -> Buffer; - #north_data{id = Id, tuple = Tuple} = NorthData -> - true = ets:insert(Tid, NorthData#north_data{inflight = true}), + {ok, Id, Payload, NOutbox} -> ReceiverPid = self(), - ReceiverPid ! {next_data, Id, Tuple}, - endpoint_timer:task(TimerPid, Id, fun() -> ReceiverPid ! {next_data, Id, Tuple} end), - Buffer#buffer{flight_num = FlightNum + 1, cursor = Id} + ReceiverPid ! {next_data, Id, Payload}, + endpoint_timer:task(TimerPid, Id, fun() -> ReceiverPid ! {next_data, Id, Payload} end), + Buffer#buffer{outbox = NOutbox, flight_num = FlightNum + 1}; + {error, Reason} -> + logger:warning("[endpoint_buffer] read next outbox failed, endpoint_id: ~p, reason: ~p", + [buffer_endpoint_id(Buffer), Reason]), + Buffer end end. -spec ack(Id :: integer(), Buffer :: #buffer{}) -> NBuffer :: #buffer{}. -ack(Id, Buffer = #buffer{timer_pid = TimerPid, tid = Tid, acc_num = AccNum, flight_num = FlightNum}) when is_integer(Id) -> - endpoint_timer:ack(TimerPid, Id), - case ets:take(Tid, Id) of - [#north_data{inflight = true}] -> - trigger_next(Buffer#buffer{acc_num = AccNum + 1, flight_num = max(FlightNum - 1, 0)}); - [#north_data{}] -> - trigger_next(Buffer#buffer{acc_num = AccNum + 1}); - [] -> +ack(Id, Buffer = #buffer{timer_pid = TimerPid, outbox = Outbox, acc_num = AccNum, flight_num = FlightNum}) when is_integer(Id) -> + case endpoint_timer:ack(TimerPid, Id) of + true -> + case endpoint_outbox:ack(Id, Outbox) of + {ok, NOutbox} -> + NBuffer = Buffer#buffer{ + outbox = NOutbox, + acc_num = AccNum + 1, + flight_num = max(FlightNum - 1, 0) + }, + trigger_next(NBuffer); + {error, Reason} -> + logger:warning("[endpoint_buffer] ack outbox failed, endpoint_id: ~p, id: ~p, reason: ~p", + [buffer_endpoint_id(Buffer), Id, Reason]), + Buffer + end; + false -> Buffer end. %% 获取当前统计信息 -spec stat(Buffer :: #buffer{}) -> map(). -stat(#buffer{acc_num = AccNum, tid = Tid, flight_num = FlightNum}) -> - QueueNum = max(ets:info(Tid, size) - FlightNum, 0), - #{ +stat(#buffer{acc_num = AccNum, outbox = Outbox, flight_num = FlightNum}) -> + OutboxStat = endpoint_outbox:stat(Outbox), + WriteSeq = maps:get(write_seq, OutboxStat, 0), + AckedSeq = maps:get(acked_seq, OutboxStat, 0), + QueueNum = max(WriteSeq - AckedSeq - FlightNum, 0), + OutboxStat#{ <<"acc_num">> => AccNum, <<"queue_num">> => QueueNum, <<"inflight_num">> => FlightNum @@ -116,10 +131,16 @@ cleanup(Buffer = #buffer{timer_pid = TimerPid}) -> Buffer. -spec recover_inflight(Buffer :: #buffer{}) -> #buffer{}. -recover_inflight(Buffer = #buffer{tid = Tid, timer_pid = TimerPid}) -> +recover_inflight(Buffer = #buffer{outbox = Outbox, timer_pid = TimerPid}) -> endpoint_timer:cleanup(TimerPid), - reset_inflight(Tid, ets:first(Tid)), - Buffer#buffer{cursor = 0, flight_num = 0}. + case endpoint_outbox:reset_reader(Outbox) of + {ok, NOutbox} -> + Buffer#buffer{outbox = NOutbox, flight_num = 0}; + {error, Reason} -> + logger:warning("[endpoint_buffer] recover inflight failed, endpoint_id: ~p, reason: ~p", + [buffer_endpoint_id(Buffer), Reason]), + Buffer#buffer{flight_num = 0} + end. -spec resize(Buffer :: #buffer{}, WindowSize :: integer()) -> #buffer{}. resize(Buffer = #buffer{}, WindowSize) when is_integer(WindowSize), WindowSize > 0 -> @@ -129,40 +150,6 @@ resize(Buffer = #buffer{}, WindowSize) when is_integer(WindowSize), WindowSize > %%% Internal functions %%%=================================================================== --spec next_pending_data(ets:tid(), integer()) -> none | #north_data{}. -next_pending_data(Tid, Cursor) -> - Key = case Cursor of - 0 -> - ets:first(Tid); - _ -> - ets:next(Tid, Cursor) - end, - next_pending_data_by_key(Tid, Key). - --spec next_pending_data_by_key(ets:tid(), '$end_of_table' | integer()) -> none | #north_data{}. -next_pending_data_by_key(_Tid, '$end_of_table') -> - none; -next_pending_data_by_key(Tid, Key) -> - case ets:lookup(Tid, Key) of - [#north_data{inflight = false} = NorthData] -> - NorthData; - [_] -> - next_pending_data_by_key(Tid, ets:next(Tid, Key)); - [] -> - next_pending_data_by_key(Tid, ets:next(Tid, Key)) - end. - --spec reset_inflight(ets:tid(), '$end_of_table' | integer()) -> ok. -reset_inflight(_Tid, '$end_of_table') -> - ok; -reset_inflight(Tid, Key) -> - NextKey = ets:next(Tid, Key), - case ets:lookup(Tid, Key) of - [#north_data{inflight = true} = NorthData] -> - true = ets:insert(Tid, NorthData#north_data{inflight = false}); - [_] -> - ok; - [] -> - ok - end, - reset_inflight(Tid, NextKey). +-spec buffer_endpoint_id(buffer()) -> integer(). +buffer_endpoint_id(#buffer{endpoint = #endpoint{id = Id}}) -> + Id. diff --git a/src/endpoint/endpoint_http.erl b/src/endpoint/endpoint_http.erl index 5bc991a..61a8049 100644 --- a/src/endpoint/endpoint_http.erl +++ b/src/endpoint/endpoint_http.erl @@ -71,14 +71,7 @@ handle_call(get_stat, _From, State = #state{buffer = Buffer}) -> {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). handle_cast({forward, Metric}, State = #state{buffer = Buffer, endpoint = #endpoint{config = #http_endpoint{token = Token}}}) -> - Tuple = case is_binary(Token) andalso Token /= <<>> of - true -> - Sign = iot_util:sha256(erlang:iolist_to_binary([Token, Metric, Token])), - {Metric, Sign}; - false -> - {Metric, <<>>} - end, - NBuffer = endpoint_buffer:append(Tuple, Buffer), + NBuffer = endpoint_buffer:append(Metric, Buffer), {noreply, State#state{buffer = NBuffer}}; handle_cast({reload, NEndpoint = #endpoint{matcher = NMatcher, config = #http_endpoint{pool_size = PoolSize}}}, State = #state{endpoint = #endpoint{matcher = Matcher}, buffer = Buffer}) -> ensure_subscription(Matcher, NMatcher), @@ -95,14 +88,8 @@ handle_cast(cleanup, State = #state{buffer = Buffer}) -> {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). -handle_info({next_data, Id, {Metric, Sign}}, State = #state{buffer = Buffer, endpoint = #endpoint{config = #http_endpoint{url = Url}}}) -> - BaseHeaders = [{<<"Content-Type">>, <<"application/json">>}], - ExtraHeaders = if - Sign =:= <<>> -> []; - true -> [{<<"X-Signature">>, Sign}] - end, - Headers = BaseHeaders ++ ExtraHeaders, - +handle_info({next_data, Id, Metric}, State = #state{buffer = Buffer, endpoint = #endpoint{config = #http_endpoint{url = Url, token = Token}}}) -> + Headers = [{<<"Content-Type">>, <<"application/json">>}] ++ patch_headers(Metric, Token), case hackney:request(post, Url, Headers, Metric) of {ok, HttpCode, _, ClientRef} when HttpCode >= 200, HttpCode < 300 -> RespBody = read_response_body(ClientRef), @@ -158,3 +145,10 @@ read_response_body(ClientRef) -> hackney:close(ClientRef), Reason end. + +-spec patch_headers(Metric :: binary(), any()) -> list(). +patch_headers(Metric, Token) when is_binary(Token), Token /= <<>> -> + Sign = iot_util:sha256(erlang:iolist_to_binary([Token, Metric, Token])), + [{<<"X-Signature">>, Sign}]; +patch_headers(_, _) -> + []. \ No newline at end of file diff --git a/src/endpoint/endpoint_timer.erl b/src/endpoint/endpoint_timer.erl index 555f247..d919be8 100644 --- a/src/endpoint/endpoint_timer.erl +++ b/src/endpoint/endpoint_timer.erl @@ -32,8 +32,9 @@ task(Pid, Id, Task) when is_pid(Pid), is_integer(Id), is_function(Task, 0) -> gen_server:cast(Pid, {task, Id, Task}). +-spec ack(pid(), integer()) -> boolean(). ack(Pid, Id) when is_pid(Pid), is_integer(Id) -> - gen_server:cast(Pid, {ack, Id}). + gen_server:call(Pid, {ack, Id}). cleanup(Pid) when is_pid(Pid) -> gen_server:cast(Pid, cleanup). @@ -67,6 +68,14 @@ init([RetryInterval]) -> {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} | {stop, Reason :: term(), NewState :: #state{}}). +handle_call({ack, Id}, _From, State = #state{timer_map = TimerMap}) -> + case maps:take(Id, TimerMap) of + error -> + {reply, false, State}; + {TimerRef, NTimerMap} -> + is_reference(TimerRef) andalso erlang:cancel_timer(TimerRef), + {reply, true, State#state{timer_map = NTimerMap}} + end; handle_call(_Request, _From, State = #state{}) -> {reply, ok, State}. @@ -81,15 +90,6 @@ handle_cast({task, Id, Task}, State = #state{retry_interval = RetryInterval, tim {noreply, State#state{timer_map = maps:put(Id, TimerRef, TimerMap)}}; %% 取消 -handle_cast({ack, Id}, State = #state{timer_map = TimerMap}) -> - case maps:take(Id, TimerMap) of - error -> - {noreply, State}; - {TimerRef, NTimerMap} -> - is_reference(TimerRef) andalso erlang:cancel_timer(TimerRef), - {noreply, State#state{timer_map = NTimerMap}} - end; - handle_cast(cleanup, State = #state{timer_map = TimerMap}) -> lists:foreach(fun({_, TimerRef}) -> catch erlang:cancel_timer(TimerRef) end, maps:to_list(TimerMap)), {noreply, State#state{timer_map = #{}}}.