Erlang / Golang 端口示例中的缓冲区大小

Question

我有一个粗略的 Erlang 到 Golang 端口示例，将数据从 Erlang 传递到 Golang 并回显响应。问题是我可以传输的数据量似乎仅限于 2^8 字节（见下文）。我认为问题可能出在 Golang 方面（没有创建足够大的缓冲区）但是用 bufio.NewReaderSize 替换 bufio.NewReader 没有用。所以我现在认为问题可能出在 Erlang 方面。我需要做什么来增加缓冲区大小/能够回显大于 2^8 字节的消息？射线justin@justin-ThinkPad-X240:~/work/erlang_golang_port$ erl -pa ebinErlang/OTP 17 [erts-6.4.1] [source] [64-bit] [smp:4:4] [async-threads:10] [kernel-poll:false]Eshell V6.4.1  (abort with ^G)1> port:start("./echo").<0.35.0>2> port:ping(65000).650003> port:ping(66000).** exception error: bad argument     in function  port:call_port/1 (port.erl, line 20)4> port:start("./echo").<0.40.0>5> port:ping(66000).    65536走package mainimport (    "bufio"    "os")const Delimiter = '\n'func main() {    // reader := bufio:NewReader(os.Stdin)    reader := bufio.NewReaderSize(os.Stdin, 1677216) // 2**24;    bytes, _ := reader.ReadBytes(Delimiter)    os.Stdout.Write(bytes[:len(bytes)-1])}Erlang-module(port).-export([start/1, stop/0, init/1]).-export([ping/1]).-define(DELIMITER, [10]).start(ExtPrg) ->    spawn(?MODULE, init, [ExtPrg]).stop() ->    myname ! stop.ping(N) ->    Msg=[round(65+26*random:uniform()) || _ <- lists:seq(1, N)],    call_port(Msg).call_port(Msg) ->    myname ! {call, self(), Msg},    receive    {myname, Result} ->        length(Result)    end.init(ExtPrg) ->    register(myname, self()),    process_flag(trap_exit, true),    Port = open_port({spawn, ExtPrg}, []),    loop(Port).loop(Port) ->    receive    {call, Caller, Msg} ->        Port ! {self(), {command, Msg++?DELIMITER}},        receive        {Port, {data, Data}} ->            Caller ! {myname, Data}        end,        loop(Port);    stop ->        Port ! {self(), close},        receive        {Port, closed} ->            exit(normal)        end;    {'EXIT', Port, _Reason} ->        exit(port_terminated)    end.

Answer 1

1. 2^8 是 256，而不是 65536，后者是 2^16（或 2 个字节）。

2. 对于排除 golang 程序，您可以简单地将您的替换echo为 GNUcat

3. 端口通信的默认消息最大大小为 64k，因此当您的端口接收消息时，第一个是字符串的前导 64k。您可以再次读取端口 以获取剩余数据，但只需将它们放入代码中即可。

4. 如果你真的想上线为基础的协议进行通信，你应该配置端口相应：

{line, L}

消息是按行传送的。每行（由依赖于操作系统的换行序列分隔）在一条消息中传递。消息数据格式为 {Flag, Line}，其中 Flag 是 eol 或 noeol，Line 是实际传递的数据（没有换行序列）。

L指定最大行长度（以字节为单位）。比这更长的行将在不止一条消息中传递，除了最后一条消息之外，所有的 Flag 都设置为 noeol。如果在换行序列之后的任何其他地方遇到文件结尾，最后一行也将被设置为 noeol 的标志。在所有其他情况下，行交付时 Flag 设置为 eol。

在{packet, N}和{line, L}设置相互排斥。

所以你的代码将是

Port = open_port({spawn, ExtPrg}, [{line, ?PACKET_SIZE]),

%%...

{call, Caller, Msg} ->

    Port ! {self(), {command, Msg++?DELIMITER}},

    D = read_data(Port, []),

    Caller ! {myname, D},

    loop(Port);

%%...

read_data(Port, Prefix) ->

receive

    {Port, {data, {noeol, Data}}} ->

        read_data(Port, Prefix ++ Data);

    {Port, {data, {eol, Data}}} ->

        Prefix ++ Data

end.

Answer 2

我一直在努力解决类似的问题。这里是管道模块的完整代码。

它允许将文本数据发送到端口并读取所有回复。

-module(apr_pipe).

-export([open_pipe/2,send/2,close/1]).

-export([loop/1,status/1,init/1]).

-include_lib("kernel/include/logger.hrl").

-define(MAX_LINE_LEN,4096).

open_pipe(Path,Cmd) ->

   State = #{path => Path, cmd => Cmd},

    Pid = spawn(?MODULE,init,[State]),

    Pid.

init(State) ->

    #{path := Path,cmd := Cmd} = State,

    FullFn = filename:join(Path,Cmd),

    Settings = [{line,?MAX_LINE_LEN},use_stdio,stderr_to_stdout,hide,binary,exit_status],

    Port = erlang:open_port({spawn_executable,FullFn},Settings),

    State2 = State#{port => Port, data => #{}},

    loop(State2).

send(Pid,Data)  -> Pid!{self(),send,Data}.

close(Pid)      -> Pid!{self(),send,close}.

status(Pid)     -> Pid!{self(),status}.

get_eol() -> <<"\n">>.

loop(State) ->

    receive

        {_Pid,send,close} -> 

                    ?LOG(notice,"got cmd: Close",[]),

                    Port = maps:get(port,State),

                    port_close(Port),

                    exit(normal);

        {Pid,send,Data} ->

                    ?LOG(notice,"Send Data ...",[]),

                    Port = maps:get(port,State),

                    port_command(Port,Data),

                    port_command(Port,get_eol()),

                    State2 = State#{status => data_sent, client => Pid},

                    loop(State2);

         {Pid,status} -> 

                    Port = maps:get(port,State),

                    ?LOG(notice,"Status: Port: ~p State: ~p",[Port,State]),

                    Pid!{status,Port,State},

                    loop(State);

        % port messages.

        {Port, {data,{noeol,Data}}} ->

                ?LOG(notice,"Port: ~p Data: ~p",[Port,Data]),

                CurData = maps:get(cur_data,State,[]),

                State2 = State#{cur_data => [Data | CurData]},

                loop(State2);

        {Port, {data, {eol,Data}}} ->

                ?LOG(notice,"Port: ~p Data: ~p",[Port,Data]),

                CurData = [Data | maps:get(cur_data,State,[])],

                CurData2 = lists:reverse(CurData),

                Reply    = list_to_binary(CurData2),

                Client = maps:get(client,State,undefined),

                State2 = State#{cur_data => [], client => undefined},

                case Client of

                    undefined -> ?LOG(error,"can not sent reply. Client: ~p Reply: ~p", [Client,Reply]),

                                 loop(State2);

                    _ -> Client!{reply,Reply},

                         loop(State2)

                 end;

        {_Port, closed} ->

                ?LOG(warning, "Port: ~p closed",[]),

                exit(normal);

        {'EXIT',  Port, Reason} ->

                 ?LOG(notice,"Port: ~p exit. Reason: ~p",[Port,Reason]),

                 exit(Reason);

        _Other -> ?LOG(error,"unexpected message: ~p",[_Other]),

                  exit({error,{unexpected_message,_Other}})

    end.