[erlang-questions] Ports and piping output between them

[Lukas Larsson]

Hello,

There is no back-pressure mechanism for open_port, so when you call it on
"cat /dev/random" you will get an endless stream of data into the process'
message queue which Erlang most likely cannot keep up with and that makes
the memory grow until it explodes. In order to get back-pressure you have
to put something in between "cat /dev/random" and the erlang port which you
processes can talk to and request more bytes instead of it just sending as
much as it can. A small bash script which sends X bytes for every Y bytes
it receives could do the trick.

Lukas

On Sat, Jul 16, 2016 at 7:13 PM, Nathan Fiedler <nathanfiedler@REDACTED>
wrote:

> My ultimate goal is to run two external processes, with the output of one
> feeding into the input of the other, a la shell piping. So something like
> `zfs send | zfs recv` or `tar | split`, but from within my Erlang
> application. Below is my simple example that fairly quickly balloons memory
> out of control and eventually explodes (i.e. the OS kills it).
>
> #!/usr/bin/env escript
> %%!
>
> main(_Args) ->
>     SendCmd = "cat /dev/random",
>     %
>     % not using 'binary' with open_port makes memory size grow really fast,
>     % but that's fine, I want that option anyway, just pointing it out
>     %
>     SendPort = erlang:open_port({spawn, SendCmd}, [exit_status, binary]),
>     RecvCmd = "strings",
>     RecvPort = erlang:open_port({spawn, RecvCmd}, [exit_status, binary]),
>     {ok, 0} = pipe_until_exit(SendPort, RecvPort, 0),
>     ensure_port_closed(SendPort),
>     ensure_port_closed(RecvPort),
>     ok.
>
> pipe_until_exit(SendPort, RecvPort, N) ->
>     %
>     % invoking garbage_collect/0 helps a little bit, but memory still grows
>     %
>     io:format("iteration ~w~n", [N]),
>     receive
>         {SendPort, {exit_status, Status}} ->
>             {ok, Status};
>         {RecvPort, {exit_status, Status}} ->
>             io:format("error: receive port exited ~w~n", [Status]);
>         {SendPort, {data, Data}} ->
>             true = erlang:port_command(RecvPort, Data),
>             pipe_until_exit(SendPort, RecvPort, N + 1);
>         {RecvPort, {data, _Data}} ->
>             pipe_until_exit(SendPort, RecvPort, N + 1);
>         Msg ->
>             io:format("some other message: ~w", [Msg])
>     end.
>
> ensure_port_closed(Port) ->
>     case erlang:port_info(Port) of
>         undefined -> ok;
>         _         -> erlang:port_close(Port)
>     end.
>
> My assumption is that I am doing something wrong, but I can't see it. I'll
> explore other solutions, such as simply generating a shell script and
> invoking it. In the mean time, if you can see something obvious, please let
> me know.
>
> Yes, I am aware of os:cmd/1, but that does not return the exit code, which
> I really want to have so I know that something at least appeared to work
> (zfs send|zfs recv and tar|split do not generate any success indication
> other than exit code).
>
> Thanks
>
> n
>
>
> _______________________________________________
> erlang-questions mailing list
> erlang-questions@REDACTED
> http://erlang.org/mailman/listinfo/erlang-questions
>
>
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://erlang.org/pipermail/erlang-questions/attachments/20160717/81d33f93/attachment.htm>