[erlang-questions] Wondering on active TCP socket

[Kresten Krab Thorup]

Hi,

In "active" mode, a socket acts like an infinite loop that simply forwards incoming data to the designated target process.  Incoming data is repackaged according to the socket's current packet mode.

There is no flow control, unlike normal message sends where the sender is punished with reductions proportional to the size of the message queue.  The port driver does not have a concept of reductions that will slow it down, and most of it's operations run in the asynch threads that are independent of the normal scheduling of Erlang processes; so driver's are at liberty to run infinitely.

So perhaps you can insert something like this in your consume loop to do your own pseudo flow control - this will  switch active on/off according to some high/low watermark for the consuming process queue length.  But since the port runs independently, there is no guarantee that the queue does not overrun.

Kresten


init() ->
   Socket = inets:accept(...);
   ControlState = set_controlled_active(Socket, 100, 200),
   main_loop(ControlState).

main_loop(CS) ->
     CS2 = control_active(true, CS),
     receive 
        {data, DataPkt} -> ...
            main_loop(CS2)
        ...


%%%%%%%%%%%%%%%%%%%%%%%
%% pseudo flow control for active sockets %%

-record(flow_control,{socket,is_active,low,high}).


set_controlled_active(Socket, LowWaterMark, HighWaterMark) ->
    [{active, IsActive}] = inet:getopts(Socket, [active]),
    control_active(true,  
		   #flow_control { socket=Socket,
				   is_active=IsActive, 
				   low=LowWaterMark, 
				   high=HighWaterMark}).



%% you want it to be active, and it is currently set to active; test
%% if our queue size is above the high watermark
control_active(true, #flow_control{is_active=true, high=HW}=State) -> 
    {message_queue_len, QLen} = erlang:process_info(message_queue_len),
    case QLen > HW of 
	true -> 
	    ok = inet:setopts(State#flow_control.socket, [{active, false}]),
	    State#flow_control{is_active=false};
	false ->
	    State
    end;

%% you want it to be active, but it is currently set to non-active; test
%% if our queue size is below the LowWatermark
control_active(true, #flow_control{is_active=false, low=LW}=State) -> 
    {message_queue_len, QLen} = erlang:process_info(message_queue_len),
    case QLen < LW of
	true ->
	    ok = inet:setopts(State#flow_control.socket, [{active, true}]),
	    State#flow_control{is_active=true};
	false ->
	    State
    end;

%% setting active to false
control_active(false, #flow_control{socket=Socket}=State ) -> 
    ok = inet:setopts(Socket,[{active, false}]),
    State#flow_control{is_active=false}.



On Aug 12, 2010, at 15:51 , Kaiduan Xie wrote:

> Hi, all,
> 
> From Joe's book,
> 
> "Once an active socket has been created, the controlling process will
> be sent {tcp, Socket, Data} messages as data is received. There
> is no way the controlling process can control the flow of these
> messages. A rogue client could send thousands of messages to
> the system, and these would all be sent to the controlling process.
> The controlling process cannot stop this flow of messages."
> 
> "This process cannot control the flow of messages to the server loop.
> If the client produces data faster than the server can consume this
> data,
> then the system can be flooded with messages—the message buffers will
> fill up, and the system might crash or behave strangely."
> 
> For this case, I think that the flow control of underlying OS TCP
> stack will kick in, and the sender can not send more packets. How the
> system can be flooded with messages in erlang?
> 
> Can someone elaborate more on this point?
> 
> Thanks,
> 
> Kaiduan
> 
> ________________________________________________________________
> erlang-questions (at) erlang.org mailing list.
> See http://www.erlang.org/faq.html
> To unsubscribe; mailto:erlang-questions-unsubscribe@REDACTED
> 

Kresten Krab Thorup, CTO, Trifork