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 ebin

Erlang/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).

65000

3> 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 main

import (

    "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

2^8 是 256，而不是 65536，后者是 2^16（或 2 个字节）。对于排除 golang 程序，您可以简单地将您的替换echo为 GNUcat端口通信的默认消息最大大小为 64k，因此当您的端口接收消息时，第一个是字符串的前导 64k。您可以再次读取端口&nbsp;以获取剩余数据，但只需将它们放入代码中即可。如果你真的想上线为基础的协议进行通信，你应该配置端口相应：{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} ->&nbsp; &nbsp; Port ! {self(), {command, Msg++?DELIMITER}},&nbsp; &nbsp; D = read_data(Port, []),&nbsp; &nbsp; Caller ! {myname, D},&nbsp; &nbsp; loop(Port);%%...read_data(Port, Prefix) ->receive&nbsp; &nbsp; {Port, {data, {noeol, Data}}} ->&nbsp; &nbsp; &nbsp; &nbsp; read_data(Port, Prefix ++ Data);&nbsp; &nbsp; {Port, {data, {eol, Data}}} ->&nbsp; &nbsp; &nbsp; &nbsp; Prefix ++ Dataend.

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) ->&nbsp; &nbsp;State = #{path => Path, cmd => Cmd},&nbsp; &nbsp; Pid = spawn(?MODULE,init,[State]),&nbsp; &nbsp; Pid.init(State) ->&nbsp; &nbsp; #{path := Path,cmd := Cmd} = State,&nbsp; &nbsp; FullFn = filename:join(Path,Cmd),&nbsp; &nbsp; Settings = [{line,?MAX_LINE_LEN},use_stdio,stderr_to_stdout,hide,binary,exit_status],&nbsp; &nbsp; Port = erlang:open_port({spawn_executable,FullFn},Settings),&nbsp; &nbsp; State2 = State#{port => Port, data => #{}},&nbsp; &nbsp; loop(State2).send(Pid,Data)&nbsp; -> Pid!{self(),send,Data}.close(Pid)&nbsp; &nbsp; &nbsp; -> Pid!{self(),send,close}.status(Pid)&nbsp; &nbsp; &nbsp;-> Pid!{self(),status}.get_eol() -> <<"\n">>.loop(State) ->&nbsp; &nbsp; receive&nbsp; &nbsp; &nbsp; &nbsp; {_Pid,send,close} ->&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ?LOG(notice,"got cmd: Close",[]),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Port = maps:get(port,State),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; port_close(Port),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; exit(normal);&nbsp; &nbsp; &nbsp; &nbsp; {Pid,send,Data} ->&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ?LOG(notice,"Send Data ...",[]),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Port = maps:get(port,State),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; port_command(Port,Data),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; port_command(Port,get_eol()),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; State2 = State#{status => data_sent, client => Pid},&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; loop(State2);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;{Pid,status} ->&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Port = maps:get(port,State),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ?LOG(notice,"Status: Port: ~p State: ~p",[Port,State]),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Pid!{status,Port,State},&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; loop(State);&nbsp; &nbsp; &nbsp; &nbsp; % port messages.&nbsp; &nbsp; &nbsp; &nbsp; {Port, {data,{noeol,Data}}} ->&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ?LOG(notice,"Port: ~p Data: ~p",[Port,Data]),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; CurData = maps:get(cur_data,State,[]),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; State2 = State#{cur_data => [Data | CurData]},&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; loop(State2);&nbsp; &nbsp; &nbsp; &nbsp; {Port, {data, {eol,Data}}} ->&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ?LOG(notice,"Port: ~p Data: ~p",[Port,Data]),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; CurData = [Data | maps:get(cur_data,State,[])],&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; CurData2 = lists:reverse(CurData),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Reply&nbsp; &nbsp; = list_to_binary(CurData2),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Client = maps:get(client,State,undefined),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; State2 = State#{cur_data => [], client => undefined},&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; case Client of&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; undefined -> ?LOG(error,"can not sent reply. Client: ~p Reply: ~p", [Client,Reply]),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;loop(State2);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; _ -> Client!{reply,Reply},&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;loop(State2)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;end;&nbsp; &nbsp; &nbsp; &nbsp; {_Port, closed} ->&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ?LOG(warning, "Port: ~p closed",[]),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; exit(normal);&nbsp; &nbsp; &nbsp; &nbsp; {'EXIT',&nbsp; Port, Reason} ->&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;?LOG(notice,"Port: ~p exit. Reason: ~p",[Port,Reason]),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;exit(Reason);&nbsp; &nbsp; &nbsp; &nbsp; _Other -> ?LOG(error,"unexpected message: ~p",[_Other]),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; exit({error,{unexpected_message,_Other}})&nbsp; &nbsp; end.