对Socket的{active, true}参数进行一些测试

以前不是很了解Erlang网络编程中流量控制，现在做一下笔记。

Erlang中Socket设置了{active, true}之后，接收到的网络消息会通过{tcp, Socket, Data}的格式主动发送给进程，这样做有个弊处就是没有做流量控制。要是有个客户端疯狂发包过来，服务器不做处理就等着堆溢出。
在《Programming Erlang》的14.2里面，介绍可以使用 {active, once} 来做控制。

{ok, Listen} = gen_tcp:listen(Port, [..,{active, once}...]),
{ok, Socket} = gen_tcp:accept(Listen),
loop(Socket).
  loop(Socket) ->
receive
  {tcp, Socket, Data} ->
    ... do something with the data ...
    %% when you're ready enable the next message
    inet:setopts(Sock, [{active, once}]),
    loop(Socket);
  {tcp_closed, Socket} ->
    ...
end.

不过锋爷认为这种做法会影响性能，因为每次接收信息都要调用一次 inet:setopts(Sock, [{active, once}]) 。

既然这样，那就先使用{active, true}，如果判断到接收到的消息包太多，再改成 {active, once}。我写了个简单的代码：

listen_socket(Socket, Mode) ->
receive
	{tcp, Socket, Bin} ->
		process_req(Bin),
			{message_queue_len, MsgQueueSize} = erlang:process_info(self(), message_queue_len),
			if 
				(MsgQueueSize > 500) and (Mode =:= active_true) ->
					io:format("Queue size is ~p~n", [MsgQueueSize]),
					inet:setopts(Socket, [{active, once}]),
					listen_socket(Socket, active_once);
				(MsgQueueSize < 10) and (Mode =:= active_once) ->
					io:format("Queue size is ~p~n", [MsgQueueSize]),
					inet:setopts(Socket, [{active, true}]),
					listen_socket(Socket, active_true);
				true ->
					io:format("Queue size is ~p~n", [MsgQueueSize]),
					listen_socket(Socket, Mode)
            end;

判断当前进程消息队列的数量，大于500时被动接收，小于10时再次改为主动接收。
打开服务器后，客户端使用阻塞方式（{active, false}）连续来发送2000个消息包给服务器，很快服务器端的消息累计达到500以上，此时{active, once}被设置，数据包开始在系统底层堆积，很快就看到客户端的gen_tcp:send调用被阻塞，直到服务器设置{active, true}后，堆积的消息被加载到内存消息队列中，gen_tcp:send恢复正常发送。
可见当{active, once}被设置后，若消息队列中还有数据包，则系统底层的数据包会被堆积，起到了流量控制的作用。

评论

4 楼 norain2050 2012-06-15  

AvinDev 写道

@ccppasm2
触发一次后，队列长度依然>500，则会再次调用
inet:setopts(Socket, [{active, once}]),  否则会调用
inet:setopts(Socket, [{active, true}]), 
因此都可以逐渐减少到<10

哈哈，版主解释的有点不对。当发现接收（控制）进程的信箱里的消息队列的长度大于500后，就改为半主动模式了（连接Socket,你就别再发给我消息了——Socket再转告操作系统——操作系统回复说“马上好，马上好，再接几个就让客户端阻塞”），此时接收进程里还有500多消息，因此，无须使用recv来从Socket要消息——因为接收进程自己的信箱里已经存有500多了，只需receive就成了。
  反复receive400到500多次后（每次无须调用inet:setopts），当信箱里的消息小于10了，才调用inet:setopts,告诉Socket可以继续转发消息了，Socket同时也告诉操作系统可以继续接收客户端的消息了（其实，此刻Socket里还保存了一些上次操作系统多接收的一点消息）。

3 楼 AvinDev 2008-10-24  

@ccppasm2
触发一次后，队列长度依然>500，则会再次调用
inet:setopts(Socket, [{active, once}]), 
否则会调用
inet:setopts(Socket, [{active, true}]), 
因此都可以逐渐减少到<10

2 楼 ccppasm2 2008-10-20  

>500后recieve只触发一次,如何再由500减少到<10?

1 楼 mryufeng 2008-05-22  

不会溢出。 inet:setopts {read_packets, N} 这个参数可以控制最大的未处理的包个数。

{read_packets, Integer}(UDP sockets)
    Sets the max number of UDP packets to read without intervention from the socket when data is available. When this many packets have been read and delivered to the destination process, new packets are not read until a new notification of available data has arrived. The default is 5, and if this parameter is set too high the system can become unresponsive due to UDP packet flooding.