From 69be3c01544492c61d69842ef055b1e151261779 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Fri, 1 May 2026 14:56:53 +0800 Subject: [PATCH] fix --- src/sdlan_network.erl | 314 +++++++++++++++++++++++++----------------- src/sdlan_stun.erl | 34 ++++- 2 files changed, 217 insertions(+), 131 deletions(-) diff --git a/src/sdlan_network.erl b/src/sdlan_network.erl index 5b0e596..a1d83b6 100644 --- a/src/sdlan_network.erl +++ b/src/sdlan_network.erl @@ -21,7 +21,7 @@ %% API -export([start_link/2]). -export([get_name/1, get_pid/1, lookup_pid/1, peer_info/3, unregister/3, debug_info/1, get_network_id/1, attach/6, arp_request/2]). --export([forward/5, update_hole/7, disable_client/2, get_channel/2]). +-export([forward_by_ets/5, update_hole/7, disable_client/2, get_channel/2]). -export([command/4, wait_command_ack/2]). -export([test_event/1]). @@ -64,11 +64,9 @@ %% 设置网络带宽 throttle_key :: atom(), + endpoint_table :: ets:tid(), %% 转发流量统计 - forward_bytes = 0, - - %% 记录已经使用了的ip, #{mac :: integer() => Host :: #endpoint{}} - endpoints = #{} + forward_bytes = 0 }). %%%=================================================================== @@ -132,9 +130,23 @@ wait_command_ack(Ref, Timeout) when is_reference(Ref), is_integer(Timeout) -> {error, timeout} end. --spec forward(pid(), Sock :: any(), SrcMac :: binary(), DstMac :: binary(), Packet :: binary()) -> no_return(). -forward(Pid, Sock, SrcMac, DstMac, Packet) when is_pid(Pid), is_binary(SrcMac), is_binary(DstMac), is_binary(Packet) -> - gen_server:cast(Pid, {forward, Sock, SrcMac, DstMac, Packet}). +-spec forward_by_ets(NetworkId :: integer(), Sock :: any(), SrcMac :: binary(), DstMac :: binary(), Packet :: binary()) -> + {ok, ForwardBytes :: integer()} | {error, Reason :: any()}. +forward_by_ets(NetworkId, Sock, SrcMac, DstMac, Packet) + when is_integer(NetworkId), is_binary(SrcMac), is_binary(DstMac), is_binary(Packet) -> + case lookup_endpoint(NetworkId, SrcMac) of + #endpoint{} -> + case sdlan_util:is_broadcast_mac(DstMac) orelse sdlan_util:is_multicast_mac(DstMac) of + true -> + forward_broadcast_by_ets(NetworkId, Sock, SrcMac, DstMac, Packet); + false -> + forward_unicast_by_ets(NetworkId, Sock, SrcMac, DstMac, Packet) + end; + undefined -> + logger:debug("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p, forward discard, src not found", + [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), + {error, src_not_found} + end. %% 更新ip地址对应的nat关系 -spec update_hole(Pid :: pid(), SessionToken :: binary(), ClientId :: binary(), Mac :: binary(), Peer :: tuple(), NatType :: integer(), V6Info :: undefined | #'SDLV6Info'{}) -> no_return(). @@ -189,9 +201,10 @@ init([Id]) when is_integer(Id) -> %% 处理加密算法 Algorithm = normalization_algorithm(Algorithm0), Key = gen_key(Algorithm), + EndpointTable = new_endpoint_table(Id), {ok, #state{network_id = Id, name = Name, domain = Domain, ipaddr = IpAddr, algorithm = Algorithm, - owner_id = OwnerId, mask_len = MaskLen, key = Key, throttle_key = ThrottleKey}}; + owner_id = OwnerId, mask_len = MaskLen, key = Key, throttle_key = ThrottleKey, endpoint_table = EndpointTable}}; {error, Reason} -> logger:warning("[sdlan_network] load network: ~p, get error: ~p", [Id, Reason]), {stop, Reason} @@ -209,14 +222,14 @@ init([Id]) when is_integer(Id) -> {stop, Reason :: term(), NewState :: #state{}}). %% 给客户端分配ip地址 handle_call({attach, ChannelPid, ClientId, Mac, Ip, Hostname}, _From, - State = #state{network_id = NetworkId, domain = Domain, endpoints = Endpoints, algorithm = Algorithm, key = Key}) -> + State = #state{network_id = NetworkId, domain = Domain, algorithm = Algorithm, key = Key, endpoint_table = EndpointTable}) -> %% 分配ip地址的时候，以mac地址为唯一基准 logger:debug("[sdlan_network] alloc_ip, network_id: ~p, client_id: ~p, mac: ~p, ip_addr: ~p", [NetworkId, ClientId, sdlan_util:format_mac(Mac), sdlan_util:int_to_ipv4(Ip)]), %% 添加域名->ip的映射关系 sdlan_hostname_regedit:insert(Hostname, Domain, Ip), - OldEndpoint = maps:get(Mac, Endpoints, undefined), + OldEndpoint = lookup_endpoint(EndpointTable, Mac), %% mac对应的Endpoint存在，并且对应的ip变了，需要通知端上清理arp maybe #endpoint{ip = OldIp} ?= OldEndpoint, @@ -232,7 +245,7 @@ handle_call({attach, ChannelPid, ClientId, Mac, Ip, Hostname}, _From, broadcast(fun(#endpoint{channel_pid = ChannelPid0}) -> sdlan_quic_channel:send_event(ChannelPid0, Event) - end, [Mac], Endpoints) + end, [Mac], EndpointTable) end, %% 重复attach需要清理之前的绑定信息；即使IP未变化，也不能保留旧channel。 cleanup_endpoint(OldEndpoint, ChannelPid, rebind), @@ -241,24 +254,26 @@ handle_call({attach, ChannelPid, ClientId, Mac, Ip, Hostname}, _From, SessionToken = gen_session_token(), Endpoint = #endpoint{channel_pid = ChannelPid, channel_ref = ChannelRef, client_id = ClientId, mac = Mac, ip = Ip, hostname = Hostname, session_token = SessionToken, last_seen = erlang:monotonic_time(second)}, + insert_endpoint(EndpointTable, Mac, Endpoint), %% 生成对应的分区id RegionId = gen_region_id(Ip), - {reply, {ok, Algorithm, Key, RegionId, SessionToken}, State#state{endpoints = maps:put(Mac, Endpoint, Endpoints)}}; + {reply, {ok, Algorithm, Key, RegionId, SessionToken}, State}; %% client设置为禁止状态，不允许重连 -handle_call({disable_client, ClientId}, _From, State = #state{endpoints = Endpoints}) -> - case search_endpoint(fun(_, #endpoint{client_id = ClientId0}) -> ClientId =:= ClientId0 end, Endpoints) of +handle_call({disable_client, ClientId}, _From, State = #state{endpoint_table = EndpointTable}) -> + case search_endpoint(fun(_, #endpoint{client_id = ClientId0}) -> ClientId =:= ClientId0 end, EndpointTable) of {ok, Mac, Endpoint} -> cleanup_endpoint(Endpoint, undefined, disabled), - {reply, ok, State#state{endpoints = maps:remove(Mac, Endpoints)}}; + delete_endpoint(EndpointTable, Mac), + {reply, ok, State}; error -> {reply, ok, State} end; -handle_call({get_channel, ClientId}, _From, State = #state{endpoints = Endpoints}) -> - case search_endpoint(fun(_, #endpoint{client_id = ClientId0}) -> ClientId =:= ClientId0 end, Endpoints) of +handle_call({get_channel, ClientId}, _From, State = #state{endpoint_table = EndpointTable}) -> + case search_endpoint(fun(_, #endpoint{client_id = ClientId0}) -> ClientId =:= ClientId0 end, EndpointTable) of {ok, _, #endpoint{channel_pid = ChannelPid}} -> {reply, {ok, ChannelPid}, State}; error -> @@ -269,12 +284,12 @@ handle_call(get_network_id, _From, State = #state{network_id = NetworkId}) -> {reply, {ok, NetworkId}, State}; %% 网络存在的nat_peer信息 -handle_call({peer_info, SrcMac, DstMac}, _From, State = #state{endpoints = Endpoints}) -> - case maps:find(DstMac, Endpoints) of - {ok, #endpoint{channel_pid = DstChannelPid, hole = #hole{peer = DstNatPeer, nat_type = DstNatType}, v6_info = DstV6Info}} -> +handle_call({peer_info, SrcMac, DstMac}, _From, State = #state{endpoint_table = EndpointTable}) -> + case lookup_endpoint(EndpointTable, DstMac) of + #endpoint{channel_pid = DstChannelPid, hole = #hole{peer = DstNatPeer, nat_type = DstNatType}, v6_info = DstV6Info} -> %% 让目标服务器发送sendRegister事件(2024-06-25 新增，提高打洞的成功率) maybe - {ok, #endpoint{hole = #hole{peer = {SrcNatIp, SrcNatPort}, nat_type = SrcNatType}, v6_info = SrcV6Info}} ?= maps:find(SrcMac, Endpoints), + #endpoint{hole = #hole{peer = {SrcNatIp, SrcNatPort}, nat_type = SrcNatType}, v6_info = SrcV6Info} ?= lookup_endpoint(EndpointTable, SrcMac), RegisterEvent = sdlan_pb:encode_msg(#'SDLEvent' { event = {send_register, #'SDLEvent.SendRegister'{ @@ -295,8 +310,8 @@ handle_call({peer_info, SrcMac, DstMac}, _From, State = #state{endpoints = Endpo end; %% arp查询 -handle_call({arp_request, TargetIp}, _From, State = #state{endpoints = Endpoints}) -> - case search_endpoint(fun(_, #endpoint{ip = Ip0}) -> Ip0 =:= TargetIp end, Endpoints) of +handle_call({arp_request, TargetIp}, _From, State = #state{endpoint_table = EndpointTable}) -> + case search_endpoint(fun(_, #endpoint{ip = Ip0}) -> Ip0 =:= TargetIp end, EndpointTable) of error -> {reply, error, State}; {ok, Mac, _} -> @@ -304,8 +319,8 @@ handle_call({arp_request, TargetIp}, _From, State = #state{endpoints = Endpoints end; %% 发送命令 -handle_call({command, ReceiverPid, ClientId, SubCommand}, _From, State = #state{endpoints = Endpoints}) -> - case search_endpoint(fun(_, #endpoint{client_id = ClientId0}) -> ClientId =:= ClientId0 end, Endpoints) of +handle_call({command, ReceiverPid, ClientId, SubCommand}, _From, State = #state{endpoint_table = EndpointTable}) -> + case search_endpoint(fun(_, #endpoint{client_id = ClientId0}) -> ClientId =:= ClientId0 end, EndpointTable) of {ok, _Mac, #endpoint{channel_pid = ChannelPid}} -> Ref = make_ref(), sdlan_quic_channel:command(ChannelPid, Ref, ReceiverPid, SubCommand), @@ -314,14 +329,14 @@ handle_call({command, ReceiverPid, ClientId, SubCommand}, _From, State = #state{ {reply, {error, <<"目标Node不在线"/utf8>>}, State} end; -handle_call(debug_info, _From, State = #state{network_id = NetworkId, ipaddr = IpAddr, mask_len = MaskLen, owner_id = OwnerId, endpoints = Endpoints}) -> +handle_call(debug_info, _From, State = #state{network_id = NetworkId, ipaddr = IpAddr, mask_len = MaskLen, owner_id = OwnerId, endpoint_table = EndpointTable}) -> Reply = #{ <<"network_id">> => NetworkId, <<"ipaddr">> => IpAddr, <<"mask_len">> => MaskLen, <<"owner_id">> => OwnerId, <<"metrics">> => network_metrics(State), - <<"used_ips">> => lists:map(fun format_endpoint/1, maps:to_list(Endpoints)) + <<"used_ips">> => lists:map(fun format_endpoint/1, list_endpoints(EndpointTable)) }, {reply, Reply, State}. @@ -331,82 +346,23 @@ handle_call(debug_info, _From, State = #state{network_id = NetworkId, ipaddr = I {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). -%% 网络数据转发, mac地址单播 -handle_cast({forward, Sock, SrcMac, DstMac, Packet}, State = #state{network_id = NetworkId, endpoints = Endpoints, throttle_key = ThrottleKey, forward_bytes = ForwardBytes}) - when is_map_key(SrcMac, Endpoints), is_map_key(DstMac, Endpoints) -> - - PacketBytes = byte_size(Packet), - case maps:find(DstMac, Endpoints) of - {ok, #endpoint{hole = #hole{peer = Peer = {NatIp, NatPort}}}} -> - case limiting_check(ThrottleKey) of - pass -> - %% client和stun之间必须有心跳机制保持nat映射可用，并且通过服务转发的udp包肯定可以到达对端的nat - logger:debug("[sdlan_network] forward data networkd_id: ~p, src_mac: ~p, dst_mac: ~p, hole: ~p", - [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac), Peer]), - - gen_udp:send(Sock, NatIp, NatPort, Packet), - {noreply, State#state{forward_bytes = ForwardBytes + PacketBytes}}; - denied -> - logger:notice("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p, rate limited, discard", - [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), - {noreply, State} - end; - {ok, _} -> - logger:debug("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p, hole not found", - [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), - {noreply, State}; - error -> - logger:debug("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p not found", - [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), - {noreply, State} - end; - -%% 网络数据转发, ip广播或组播, 不限流 -handle_cast({forward, Sock, SrcMac, DstMac, Packet}, State = #state{network_id = NetworkId, endpoints = Endpoints, forward_bytes = ForwardBytes}) - when is_map_key(SrcMac, Endpoints) -> - %% 广播地址和组播地址，需要转发到整个网络 - case sdlan_util:is_broadcast_mac(DstMac) orelse sdlan_util:is_multicast_mac(DstMac) of - true -> - PacketBytes = byte_size(Packet), - %% 消息广播 - broadcast(fun - (#endpoint{hole = #hole{peer = {NatIp, NatPort}}}) -> - gen_udp:send(Sock, NatIp, NatPort, Packet); - (#endpoint{}) -> - ok - end, [SrcMac], Endpoints), - %% client和stun之间必须有心跳机制保持nat映射可用，并且通过服务转发的udp包肯定可以到达对端的nat - logger:debug("[sdlan_network] broadcast data networkd_id: ~p, src_mac: ~p, dst_mac: ~p", - [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), - - {noreply, State#state{forward_bytes = ForwardBytes + PacketBytes}}; - false -> - logger:debug("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p, forward discard 1", - [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), - {noreply, State} - end; - -handle_cast({forward, _Sock, SrcMac, DstMac, _Packet}, State = #state{network_id = NetworkId}) -> - logger:debug("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p, forward discard 2", - [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), - {noreply, State}; - %% 删除ip的占用并关闭channel -handle_cast({unregister, _ClientId, Mac}, State = #state{network_id = NetworkId, endpoints = Endpoints}) -> +handle_cast({unregister, _ClientId, Mac}, State = #state{network_id = NetworkId, endpoint_table = EndpointTable}) -> logger:debug("[sdlan_network] networkd_id: ~p, unregister Mac: ~p", [NetworkId, sdlan_util:format_mac(Mac)]), - case maps:take(Mac, Endpoints) of - {Endpoint, NEndpoints} -> + case lookup_endpoint(EndpointTable, Mac) of + Endpoint = #endpoint{} -> cleanup_endpoint(Endpoint, Endpoint#endpoint.channel_pid, unregister), - {noreply, State#state{endpoints = NEndpoints}}; - error -> + delete_endpoint(EndpointTable, Mac), + {noreply, State}; + undefined -> {noreply, State} end; %% 需要判断，client是属于当前网络的 -handle_cast({update_hole, SessionToken, ClientId, Mac, Peer, NatType, V6Info}, State = #state{endpoints = Endpoints}) -> - case maps:find(Mac, Endpoints) of +handle_cast({update_hole, SessionToken, ClientId, Mac, Peer, NatType, V6Info}, State = #state{endpoint_table = EndpointTable}) -> + case lookup_endpoint(EndpointTable, Mac) of %% ClientId =:= ClientId0, SessionToken =:= SessionToken0 - {ok, Endpoint0 = #endpoint{ip = Ip, client_id = ClientId, hole = OldHole, session_token = SessionToken}} -> + Endpoint0 = #endpoint{ip = Ip, client_id = ClientId, hole = OldHole, session_token = SessionToken} -> NHole = #hole{peer = Peer, nat_type = NatType}, maybe true ?= not same_hole(OldHole, NHole), @@ -421,12 +377,13 @@ handle_cast({update_hole, SessionToken, ClientId, Mac, Peer, NatType, V6Info}, S logger:debug("[sdlan_network] Event: nat_changed, update_hole, client_id: ~p(~p), hole changed", [ClientId, Ip]), broadcast(fun(#endpoint{channel_pid = ChannelPid}) -> sdlan_quic_channel:send_event(ChannelPid, NatChangedEvent) - end, [Mac], Endpoints) + end, [Mac], EndpointTable) end, NEndpoint = Endpoint0#endpoint{hole = NHole, v6_info = V6Info, last_seen = erlang:monotonic_time(second)}, + insert_endpoint(EndpointTable, Mac, NEndpoint), logger:debug("[sdlan_network] mac: ~p, ip: ~p, endpoint is: ~p", [Mac, Ip, NEndpoint]), - {noreply, State#state{endpoints = maps:put(Mac, NEndpoint, Endpoints)}}; + {noreply, State}; _ -> {noreply, State} end. @@ -439,15 +396,20 @@ handle_cast({update_hole, SessionToken, ClientId, Mac, Peer, NatType, V6Info}, S {stop, Reason :: term(), NewState :: #state{}}). 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), + case ForwardBytes > 0 of + true -> + catch sdlan_api:network_forward_report(NetworkId, ForwardBytes); + false -> + ok + end, logger:debug("[sdlan_network] metrics: ~p", [network_metrics(State)]), {noreply, State#state{forward_bytes = 0}}; %% Channel进程退出, hole里面的数据也需要清理 -handle_info({'DOWN', _MRef, process, ChannelPid, Reason}, State = #state{network_id = NetworkId, endpoints = Endpoints}) -> +handle_info({'DOWN', _MRef, process, ChannelPid, Reason}, State = #state{network_id = NetworkId, endpoint_table = EndpointTable}) -> logger:notice("[sdlan_network] network_id: ~p, channel_pid: ~p, close with reason: ~p", [NetworkId, ChannelPid, Reason]), - NEndpoints = maps:filter(fun(_, #endpoint{channel_pid = ChannelPid0}) -> ChannelPid =/= ChannelPid0 end, Endpoints), - {noreply, State#state{endpoints = NEndpoints}}. + remove_channel_endpoints(ChannelPid, EndpointTable), + {noreply, State}. %% @private %% @doc This function is called by a gen_server when it is about to @@ -456,7 +418,7 @@ handle_info({'DOWN', _MRef, process, ChannelPid, Reason}, State = #state{network %% with Reason. The return value is ignored. -spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), State :: #state{}) -> none()). -terminate(Reason, #state{network_id = NetworkId, endpoints = Endpoints}) -> +terminate(Reason, #state{network_id = NetworkId, endpoint_table = EndpointTable}) -> broadcast(fun(#endpoint{channel_pid = ChannelPid}) -> case is_pid(ChannelPid) andalso is_process_alive(ChannelPid) of true -> @@ -473,7 +435,7 @@ terminate(Reason, #state{network_id = NetworkId, endpoints = Endpoints}) -> false -> ok end - end, [], Endpoints), + end, [], EndpointTable), logger:debug("[sdlan_network] network: ~p, will terminate with reason: ~p", [NetworkId, Reason]), ok. @@ -515,6 +477,105 @@ limiting_check(ThrottleKey) -> end end. +-spec endpoint_table_name(NetworkId :: integer()) -> atom(). +endpoint_table_name(NetworkId) when is_integer(NetworkId) -> + list_to_atom("sdlan_network_endpoint:" ++ integer_to_list(NetworkId)). + +-spec throttle_key(NetworkId :: integer()) -> atom(). +throttle_key(NetworkId) when is_integer(NetworkId) -> + list_to_atom("network_throttle:" ++ integer_to_list(NetworkId)). + +-spec new_endpoint_table(NetworkId :: integer()) -> ets:tid(). +new_endpoint_table(NetworkId) when is_integer(NetworkId) -> + ets:new(endpoint_table_name(NetworkId), [ + named_table, + protected, + set, + {read_concurrency, true}, + {write_concurrency, true} + ]). + +-spec insert_endpoint(Table :: ets:tid(), Mac :: binary(), Endpoint :: #endpoint{}) -> true. +insert_endpoint(Table, Mac, Endpoint = #endpoint{}) when is_binary(Mac) -> + ets:insert(Table, {Mac, Endpoint}). + +-spec delete_endpoint(Table :: ets:tid(), Mac :: binary()) -> true. +delete_endpoint(Table, Mac) when is_binary(Mac) -> + ets:delete(Table, Mac). + +-spec lookup_endpoint(NetworkIdOrTable :: integer() | ets:tid(), Mac :: binary()) -> #endpoint{} | undefined. +lookup_endpoint(NetworkId, Mac) when is_integer(NetworkId), is_binary(Mac) -> + lookup_endpoint(endpoint_table_name(NetworkId), Mac); +lookup_endpoint(Table, Mac) when is_binary(Mac) -> + case catch ets:lookup(Table, Mac) of + [{Mac, Endpoint = #endpoint{}}] -> + Endpoint; + [] -> + undefined; + {'EXIT', _} -> + undefined + end. + +-spec list_endpoints(NetworkIdOrTable :: integer() | ets:tid()) -> [{binary(), #endpoint{}}]. +list_endpoints(NetworkId) when is_integer(NetworkId) -> + list_endpoints(endpoint_table_name(NetworkId)); +list_endpoints(Table) -> + case catch ets:tab2list(Table) of + Endpoints when is_list(Endpoints) -> + Endpoints; + {'EXIT', _} -> + [] + end. + +-spec remove_channel_endpoints(ChannelPid :: pid(), Table :: ets:tid()) -> ok. +remove_channel_endpoints(ChannelPid, Table) when is_pid(ChannelPid) -> + lists:foreach(fun + ({Mac, #endpoint{channel_pid = ChannelPid0}}) when ChannelPid =:= ChannelPid0 -> + delete_endpoint(Table, Mac); + (_) -> + ok + end, list_endpoints(Table)). + +-spec forward_unicast_by_ets(NetworkId :: integer(), Sock :: any(), SrcMac :: binary(), DstMac :: binary(), Packet :: binary()) -> + {ok, integer()} | {error, any()}. +forward_unicast_by_ets(NetworkId, Sock, SrcMac, DstMac, Packet) -> + PacketBytes = byte_size(Packet), + case lookup_endpoint(NetworkId, DstMac) of + #endpoint{hole = #hole{peer = Peer = {NatIp, NatPort}}} -> + case limiting_check(throttle_key(NetworkId)) of + pass -> + logger:debug("[sdlan_network] forward data by ets networkd_id: ~p, src_mac: ~p, dst_mac: ~p, hole: ~p", + [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac), Peer]), + gen_udp:send(Sock, NatIp, NatPort, Packet), + {ok, PacketBytes}; + denied -> + logger:notice("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p, rate limited, discard", + [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), + {error, rate_limited} + end; + #endpoint{} -> + logger:debug("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p, hole not found", + [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), + {error, hole_not_found}; + undefined -> + logger:debug("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p not found", + [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), + {error, dst_not_found} + end. + +-spec forward_broadcast_by_ets(NetworkId :: integer(), Sock :: any(), SrcMac :: binary(), DstMac :: binary(), Packet :: binary()) -> + {ok, integer()}. +forward_broadcast_by_ets(NetworkId, Sock, SrcMac, DstMac, Packet) -> + lists:foreach(fun + ({Mac, #endpoint{hole = #hole{peer = {NatIp, NatPort}}}}) when Mac =/= SrcMac -> + gen_udp:send(Sock, NatIp, NatPort, Packet); + (_) -> + ok + end, list_endpoints(NetworkId)), + logger:debug("[sdlan_network] broadcast data by ets networkd_id: ~p, src_mac: ~p, dst_mac: ~p", + [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), + {ok, byte_size(Packet)}. + cleanup_endpoint(undefined, _KeepChannelPid, _Reason) -> ok; cleanup_endpoint(#endpoint{channel_ref = ChannelRef, channel_pid = ChannelPid}, KeepChannelPid, Reason) -> @@ -532,16 +593,16 @@ should_stop_channel(ChannelPid, KeepChannelPid) when is_pid(ChannelPid), Channel should_stop_channel(_, _) -> false. --spec broadcast(Fun :: fun((#endpoint{}) -> no_return()), ExcludeMacs :: [binary()], Endpoints :: map()) -> no_return(). -broadcast(Fun, ExcludeMacs, Endpoints) when is_function(Fun, 1), is_map(Endpoints), is_list(ExcludeMacs) -> - maps:foreach(fun(Mac, Endpoint) -> +-spec broadcast(Fun :: fun((#endpoint{}) -> no_return()), ExcludeMacs :: [binary()], Table :: ets:tid()) -> no_return(). +broadcast(Fun, ExcludeMacs, Table) when is_function(Fun, 1), is_list(ExcludeMacs) -> + lists:foreach(fun({Mac, Endpoint}) -> case lists:member(Mac, ExcludeMacs) of true -> ok; false -> Fun(Endpoint) end - end, Endpoints). + end, list_endpoints(Table)). -spec format_endpoint({Mac :: binary(), Host :: #endpoint{}}) -> map(). format_endpoint({Mac, #endpoint{client_id = ClientId, ip = Ip, hole = #hole{peer = {NatIp, NatPort}, nat_type = NatType}, v6_info = V6Info}}) -> @@ -579,17 +640,25 @@ format_endpoint({Mac, #endpoint{client_id = ClientId, ip = Ip, hole = undefined, v6_info => V6InfoMap }. -network_metrics(#state{network_id = NetworkId, endpoints = Endpoints, forward_bytes = ForwardBytes}) -> +network_metrics(#state{network_id = NetworkId, endpoint_table = EndpointTable, forward_bytes = ForwardBytes}) -> ProcInfo = maps:from_list(process_info(self(), [message_queue_len, memory, reductions])), ProcInfo#{ network_id => NetworkId, - endpoint_count => maps:size(Endpoints), + endpoint_count => endpoint_count(EndpointTable), forward_bytes => ForwardBytes, - channel_metrics => channel_metrics(Endpoints) + channel_metrics => channel_metrics(EndpointTable) }. -channel_metrics(Endpoints) -> - maps:fold(fun(_Mac, #endpoint{channel_pid = ChannelPid}, Acc) -> +endpoint_count(Table) -> + case catch ets:info(Table, size) of + Size when is_integer(Size) -> + Size; + _ -> + 0 + end. + +channel_metrics(Table) -> + lists:foldl(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])), @@ -597,23 +666,20 @@ channel_metrics(Endpoints) -> false -> Acc end - end, [], Endpoints). + end, [], list_endpoints(Table)). --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)). -search_endpoint0(F, Iter) when is_function(F, 2) -> - case maps:next(Iter) of - {Key, Value, NextIter} -> +-spec search_endpoint(F :: fun((term(), term()) -> boolean()), Table :: ets:tid()) -> error | {ok, Key :: any(), Val :: any()}. +search_endpoint(F, Table) when is_function(F, 2) -> + search_endpoint0(F, list_endpoints(Table)). +search_endpoint0(F, [{Key, Value}|Rest]) when is_function(F, 2) -> case F(Key, Value) of true -> {ok, Key, Value}; false -> - search_endpoint0(F, NextIter) + search_endpoint0(F, Rest) end; - 'none' -> - error - end. +search_endpoint0(_F, []) -> + error. -spec same_hole(Hole :: #hole{}, Hole :: #hole{}) -> boolean(). same_hole(#hole{peer = OldPeer, nat_type = OldNatType}, #hole{peer = Peer, nat_type = NatType}) when OldPeer =:= Peer, OldNatType =:= NatType -> diff --git a/src/sdlan_stun.erl b/src/sdlan_stun.erl index 6e94129..021d424 100644 --- a/src/sdlan_stun.erl +++ b/src/sdlan_stun.erl @@ -21,9 +21,11 @@ -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -define(SERVER, ?MODULE). +-define(FLOW_REPORT_INTERVAL, 60 * 1000). -record(state, { - socket + socket, + flow_counters = #{} }). %%%=================================================================== @@ -62,6 +64,7 @@ init([Port]) -> ], {ok, Socket} = gen_udp:open(Port, Opts), inet_udp:controlling_process(Socket, self()), + erlang:start_timer(?FLOW_REPORT_INTERVAL, self(), flow_report_ticker), logger:debug("[sdlan_stun] start at port: ~p", [Port]), {ok, #state{socket = Socket}}. @@ -133,17 +136,30 @@ handle_info({udp, Sock, ClientIp, ClientPort, <>}, State = #state{socket = Sock}) -> +handle_info({udp, _, _Ip, _Port, <>}, State = #state{socket = Sock, flow_counters = FlowCounters}) -> Data = catch sdlan_pb:decode_msg(Body, 'SDLData'), - maybe + NFlowCounters = maybe #'SDLData'{network_id = NetworkId, src_mac = SrcMac, dst_mac = DstMac, ttl = TTL} ?= Data, logger:debug("[sdlan_stun] forward data, network_id: ~p", [NetworkId]), - {ok, NetworkPid} ?= sdlan_network:lookup_pid(NetworkId), %% 重新打包数据ttl需要减1 NData = sdlan_pb:encode_msg(Data#'SDLData'{ttl = TTL - 1, is_p2p = false}), - sdlan_network:forward(NetworkPid, Sock, SrcMac, DstMac, <>) + case sdlan_network:forward_by_ets(NetworkId, Sock, SrcMac, DstMac, <>) of + {ok, ForwardBytes} -> + incr_flow_counter(NetworkId, ForwardBytes, FlowCounters); + {error, _Reason} -> + FlowCounters + end + else _ -> + FlowCounters end, - {noreply, State}; + {noreply, State#state{flow_counters = NFlowCounters}}; + +handle_info({timeout, _, flow_report_ticker}, State = #state{flow_counters = FlowCounters}) -> + erlang:start_timer(?FLOW_REPORT_INTERVAL, self(), flow_report_ticker), + maps:foreach(fun(NetworkId, ForwardBytes) -> + catch sdlan_api:network_forward_report(NetworkId, ForwardBytes) + end, FlowCounters), + {noreply, State#state{flow_counters = #{}}}; handle_info(Info, State) -> logger:error("[sdlan_stun] get a unknown message: ~p, channel will closed", [Info]), @@ -174,4 +190,8 @@ code_change(_OldVsn, State = #state{}, _Extra) -> -spec int_ip(tuple()) -> integer(). int_ip({Ip0, Ip1, Ip2, Ip3}) -> <> = <>, - Ip. \ No newline at end of file + Ip. + +-spec incr_flow_counter(NetworkId :: integer(), ForwardBytes :: integer(), Counters :: map()) -> map(). +incr_flow_counter(NetworkId, ForwardBytes, Counters) when is_integer(NetworkId), is_integer(ForwardBytes), is_map(Counters) -> + maps:update_with(NetworkId, fun(V) -> V + ForwardBytes end, ForwardBytes, Counters).