A Better Erlang TCP listening pattern: addressing the fast packet loss problem
December 31, 2008 Erlang, Programming, Tools and Libraries, Tutorials 11 Comments[digg-reddit-me]I’ve had mixed reactions to this when I’ve discussed it with people on IRC. This may be well known to oldbear Erlang programmers. I suppose it’s also possible that I’m wrong, though I’ve talked to several people I respect, and more than one of them have suggested that they were already aware of this problem. If I’m wrong, please let me know; I’m open to the possibility that there’s a better answer that I just don’t know about. I’ve never seen it discussed on the web, at least. Update: Serge Aleynikov points out that this TrapExit tutorial documents this approach.
I think this is probably real.
I believe there is a significant defect in the idiomatic listener pattern as discussed by most Erlang websites and as provided by most Erlang example code, and which is found in many Erlang applications. This defect is relatively easily repaired once noticed without significant impact on surrounding code, and I have added functionality to my ScUtil Library to handle this automatically under the name standard_listener.
The fundamental problem is essentially a form of race condition. The idiomatic listener is, roughly:
do_listen(Port, Opts, Handler) ->
case gen_tcp:listen(Port, Opts) of
{ ok, ListeningSocket } ->
listen_loop(ListeningSocket, Handler);
{ error, E } ->
{ error, E }
end.
listen_loop(ListeningSocket, Handler) ->
case gen_tcp:accept(ListeningSocket) of
{ ok, ConnectedSocket } ->
spawn(node(), Handler, [ConnectedSocket]),
listen_loop(ListeningSocket);
{ error, E } ->
{ error, E }
end.
Now consider the case of a server under very heavy load. Further, consider that the listening socket is opened either {active,true} or {active,once}, which is true in the vast bulk of Erlang network applications, meaning that packets are delivered automatically to the socket owner. The general pattern is that the listening socket accepts a connection, spawns a handling process, passes the connected socket to that handling process, and the handling process takes ownership of the socket.
The problem is that it takes time for that all to happen, and Erlang doesn’t specify or allow you to control its timeslicing behavior (as well it should not). As active sockets are managed by a standalone process, this means that if the connecting client is fast and the network is fast, the first packet (even the first several under extreme circumstances) could be delivered before the socket has been taken over by the handling PID, meaning that its contents would be dispatched to the wrong process, with no indication of where they were meant to go. This invalidates connections and fills a non-discarding mailbox, which is a potentially serious memory leak (especially given that erlang’s response to out of memory conditions is to abort an entire VM.)
Obviously, this is intolerable. There are better answers, though, than to switch to {active,false}. One suggestion I heard was to pre-spawn handlers in order to reduce the gap time, but that doesn’t solve the problem, it just makes it less likely.
The approach that I took is to lie. standard_listener takes the following steps to resolve the problem:
- Add the default {active,true} to the inet options list, if it isn’t already present.
- Strip out the {active,Foo} from the inet options list, and store it as ActiveStatus.
- Add {active,false} to the inet options list, and use that options list to open the listener.
- When spawning handler processes, pass a shunt as the starting function, taking the real handling function and the real ActiveStatus as arguments
- The shunt sets the real ActiveStatus from inside the handler process, at which point the socket begins delivering messages
This neatly closes the problem. A free, MIT license implementation can be found in ScUtil beginning in version 96. A simplified, corrected example follows for immediate reference; the thing in ScUtil is more feature complete.
do_listen(Port, Opts) -> ActiveStatus = case proplists:get_value(active, SocketOptions) of undefined -> true; Other -> Other end, FixedOpts = proplists:delete(active, SocketOptions) ++ [{active, false}], case gen_tcp:listen(Port, FixedOpts) of { ok, ListeningSocket } -> listen_loop(ActiveStatus, ListeningSocket); { error, E } -> { error, E } end. listen_loop(ActiveStatus, ListeningSocket, Handler) -> case gen_tcp:accept(ListeningSocket) of { ok, ConnectedSocket } -> spawn(?MODULE, shunt, [ActiveStatus,ConnectedSocket,Handler]), listen_loop(ActiveStatus, ListeningSocket, Handler); { error, E } -> { error, E } end. shunt(ActiveStatus, ConnectedSocket, Handler) -> controlling_process(ConnectedSocket, self()), inet:setopts(ConnectedSocket, [{active, ActiveStatus}]), Handler(ConnectedSocket).