fix
This commit is contained in:
parent
6bce50ea65
commit
69be3c0154
@ -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 ->
|
||||
|
||||
@ -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, <<?PACKET_STUN_PROBE:8, Body/binar
|
||||
end,
|
||||
{noreply, State};
|
||||
|
||||
handle_info({udp, _, _Ip, _Port, <<?PACKET_STUN_DATA, Body/binary>>}, State = #state{socket = Sock}) ->
|
||||
handle_info({udp, _, _Ip, _Port, <<?PACKET_STUN_DATA, Body/binary>>}, 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, <<?PACKET_STUN_DATA, NData/binary>>)
|
||||
case sdlan_network:forward_by_ets(NetworkId, Sock, SrcMac, DstMac, <<?PACKET_STUN_DATA, NData/binary>>) 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]),
|
||||
@ -175,3 +191,7 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
|
||||
int_ip({Ip0, Ip1, Ip2, Ip3}) ->
|
||||
<<Ip:32>> = <<Ip0, Ip1, Ip2, Ip3>>,
|
||||
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).
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user