系统技术非业余研究

这是未公开的一个特性, 很方便查看Erlang内部的port使用状态.

演示下:

root@ubuntu:~# erl
Erlang R13B04 (erts-5.7.5) [source] [smp:2:2] [rq:2] [async-threads:0] [hipe] [kernel-poll:false]

Eshell V5.7.5  (abort with ^G)
1> 
BREAK: (a)bort (c)ontinue (p)roc info (i)nfo (l)oaded
       (v)ersion (k)ill (D)b-tables (d)istribution
o
=port:#Port<0.1>
Slot: 1
Connected: <0.3.0>
Links: <0.3.0>
Port controls linked-in driver: efile
=port:#Port<0.49>
Slot: 49
Connected: <0.18.0>
Links: <0.18.0>
Port controls linked-in driver: efile
=port:#Port<0.306>
Slot: 306
Connected: <0.21.0>
Links: <0.21.0>
Port is UNIX fd not opened by emulator: 2/2
=port:#Port<0.315>
Slot: 315
Connected: <0.23.0>
Links: <0.23.0>
Port controls linked-in driver: tty_sl -c -e

在做网络程序的时候我们会经常用到多进程模式. 主进程执行bind调用得到句柄后, 同时fork N个子进程, 把句柄传递给子进程, 多进程同时accept来处理.
这个模型在erlang下很有现实意义的. 在之前的测试中,我们演示了erlang的单处理器模式的威力,最多的时候在单cpu上可以发起40,000个tcp广播包.
但是erlang如何利用这个能力呢? 其实Erlang的port也是靠fork来实现的, 是支持这个能力的, 只不过官方的版本会在fork的时候, 把继承过来的句柄全部关闭.

让我们crack下代码来绕过这个问题.
erts/emulator/sys/unix/sys.c

1513            if(0) /*/*fork路径*/*/
1514              for (i = opts->use_stdio ? 3 : 5; i < max_files; i++)
1515                (void) close(i);
...
1581        fprintf(stderr, "cracked\n");    /*vfork路径*/
1582        sprintf(fd_close_range, "%d:%d", opts->use_stdio ? 3 : 5, opts->use_stdio ? 3 : 5);

记得重新make && make install

我们下面的代码演示在主进程把一个tcp句柄传递过去, 然后在子进程中恢复成gen_tcp.

root@nd-desktop:~/otp/test# ls *.erl
child.erl  test.erl
root@ubuntu:~/otp/test# erlc *.erl

root@nd-desktop:~/otp/test# cat test.erl

-module(test).
-export([start/0]).
start()->
    process_flag(trap_exit, true),
    {ok, Sock} = gen_tcp:listen(0, []),
    {ok, Handle} = inet:getfd(Sock),
    Command ="erl -noshell -s child -handle " ++ integer_to_list(Handle),
    io:format("child command line: ~p~n", [Command]),
    Child = case (catch open_port({spawn, Command}, [in, {line, 256}])) of
                {'EXIT', Reason}->
                    io:format("open child error, reason: ~p~n", [Reason]),
                    halt(1);
                Port-> Port
            end,
    register(?MODULE, self()),
    io:format("STOP ME: test!stop. ~n",[]),
    loop(Child),
    io:format("bye~n",[]).

loop(Child)->
    receive
        {Child, {data, Result}} ->
            io:format("child say: ~p~n", [Result]),
            loop(Child);
        stop->
            halt(0);
        Other->            
            io:format("got msg: ~p~n", [Other]),
            loop(Child)
    end.

root@nd-desktop:~/otp/test# cat child.erl

-module(child).
-export([start/0]).

start()->
    {ok, [[HandleArg|_]|_]} = init:get_argument(handle),
    Handle = list_to_integer(HandleArg),
    io:format("handle: ~w~n", [Handle]),
    case gen_tcp:fdopen(Handle, []) of
        {ok, Socket} ->
            io:format("got socket ok: ~p~n", [Socket]);
        _ ->
            io:format("got socket fail~n", [])
    end,

    halt(0).

root@ubuntu:~/otp/test# erl -noshell -s test
child command line: "erl -noshell -s child -handle 8"
cracked
STOP ME: test!stop. 
child say: {eol,"handle: 8"}
child say: {eol,"got socket ok: #Port<0.354>"}
got msg: {'EXIT',#Port<0.360>,normal}
^C
BREAK: (a)bort (c)ontinue (p)roc info (i)nfo (l)oaded
       (v)ersion (k)ill (D)b-tables (d)istribution

Bingo! 成功实现目的!

这里给大家一个思路就是说明fork是可行的, 如果你有这个需求把上面的fork patch做好点就行.

从lib/megaco/src/tcp/megaco_tcp_connection.erl摘抄的代码, 挺详细的关于系统的信息:

    SchedId      = erlang:system_info(scheduler_id),
    SchedNum     = erlang:system_info(schedulers),
    ProcCount    = erlang:system_info(process_count),
    ProcLimit    = erlang:system_info(process_limit),
    ProcMemUsed  = erlang:memory(processes_used),
    ProcMemAlloc = erlang:memory(processes),
    MemTot       = erlang:memory(total),
    io:format("abormal termination: "
              "~n   Scheduler id:                         ~p"
              "~n   Num scheduler:                        ~p"
              "~n   Process count:                        ~p"
              "~n   Process limit:                        ~p"
              "~n   Memory used by erlang processes:      ~p"
              "~n   Memory allocated by erlang processes: ~p"
              "~n   The total amount of memory allocated: ~p"
              "~n~p",
              [SchedId, SchedNum, ProcCount, ProcLimit,
               ProcMemUsed, ProcMemAlloc, MemTot, Reason]),
    ok.

看图不说话:

有兴趣的同学google之!

看到erlang-china.org上有个帖子在问这个问题

一直不太明白iolist,跟list到底有什么区别？请各位大侠指教。。

我翻看了半天erlang的文档也没写的太明白,所以就看看源码:
erts/emulator/beam/utils.c

3015int io_list_len(Eterm obj)
3016{
3017    Eterm* objp;
3018    Sint len = 0;
3019    DECLARE_ESTACK(s);
3020    goto L_again;
3021
3022    while (!ESTACK_ISEMPTY(s)) {
3023        obj = ESTACK_POP(s);
3024    L_again:
3025        if (is_list(obj)) {
3026        L_iter_list:
3027            objp = list_val(obj);
3028            /* Head */
3029            obj = CAR(objp);
3030            if (is_byte(obj)) {
3031                len++;
3032            } else if (is_binary(obj) && binary_bitsize(obj) == 0) {
3033                len += binary_size(obj);
3034            } else if (is_list(obj)) {
3035                ESTACK_PUSH(s, CDR(objp));
3036                goto L_iter_list; /* on head */
3037            } else if (is_not_nil(obj)) {
3038                goto L_type_error;
3039            }
3040            /* Tail */
3041            obj = CDR(objp);
3042            if (is_list(obj))
3043                goto L_iter_list; /* on tail */
3044            else if (is_binary(obj) && binary_bitsize(obj) == 0) {
3045                len += binary_size(obj);
3046            } else if (is_not_nil(obj)) {
3047                goto L_type_error;
3048            }
3049        } else if (is_binary(obj) && binary_bitsize(obj) == 0) { /* Tail was binary */
3050            len += binary_size(obj);
3051        } else if (is_not_nil(obj)) {
3052            goto L_type_error;
3053        }
3054    }
3055
3056    DESTROY_ESTACK(s);
3057    return len;
3058
3059 L_type_error:
3060    DESTROY_ESTACK(s);
3061    return -1;
3062}

从源码可以看出来iolist是这样的定义的:
1. []
2. binary
3. 列表, 每个元素是int(0-255)或者binary或者iolist.
其中binary是指 bitsize % 8 == 0 .
int 是0-255

root@ubuntu:/usr/src/otp# erl
Erlang R13B04 (erts-5.7.5) [source] [smp:2:2] [rq:2] [async-threads:0] [hipe] [kernel-poll:false]

Eshell V5.7.5  (abort with ^G)
2> iolist_size(<<>>).
0
3> iolist_size(<<1:1>>).
** exception error: bad argument
 in function  iolist_size/1
 called as iolist_size(<<1:1>>)
4> iolist_size(<<1:8>>).
1
5> iolist_size([]).
0
6> iolist_size(<<1,2>>).
2
7> iolist_size([1,2]).
2
8> iolist_size([1,2, <<1,2>>]).
4
9> iolist_size([1,2, <<1,2>>, [2]]).
5
10> iolist_size([1,2, <<1,2>>, [2]]).
5
11> iolist_size([<<1:1>>]).
** exception error: bad argument
 in function  iolist_size/1
 called as iolist_size([<<1:1>>])
12> iolist_size([257]).
** exception error: bad argument
     in function  iolist_size/1
        called as iolist_size([257])

Iolist的作用是用于往port送数据的时候.由于底层的系统调用如writev支持向量写, 就避免了无谓的iolist_to_binary这样的扁平话操作, 避免了内存拷贝,极大的提高了效率.
建议多用.

前段时间arksea同学提出这个问题, 因为文档里面写的很不明白.

erlang:send_after(Time, Dest, Msg) -> TimerRef
Time = int()
0 <= Time <= 4294967295
Dest = pid() | RegName
LocalPid = pid() (of a process, alive or dead, on the local node)
Msg = term()
TimerRef = ref()
Starts a timer which will send the message Msg to Dest after Time milliseconds.

If Dest is an atom, it is supposed to be the name of a registered process. The process referred to by the name is looked up at the time of delivery. No error is given if the name does not refer to a process.

If Dest is a pid, the timer will be automatically canceled if the process referred to by the pid is not alive, or when the process exits. This feature was introduced in erts version 5.4.11. Note that timers will not be automatically canceled when Dest is an atom.

See also erlang:start_timer/3, erlang:cancel_timer/1, and erlang:read_timer/1.

Failure: badarg if the arguments does not satisfy the requirements specified above.

erlang:start_timer(Time, Dest, Msg) -> TimerRef
Time = int()
0 <= Time <= 4294967295
Dest = LocalPid | RegName
LocalPid = pid() (of a process, alive or dead, on the local node)
RegName = atom()
Msg = term()
TimerRef = ref()
Starts a timer which will send the message {timeout, TimerRef, Msg} to Dest after Time milliseconds.

If Dest is an atom, it is supposed to be the name of a registered process. The process referred to by the name is looked up at the time of delivery. No error is given if the name does not refer to a process.

If Dest is a pid, the timer will be automatically canceled if the process referred to by the pid is not alive, or when the process exits. This feature was introduced in erts version 5.4.11. Note that timers will not be automatically canceled when Dest is an atom.

See also erlang:send_after/3, erlang:cancel_timer/1, and erlang:read_timer/1.

Failure: badarg if the arguments does not satisfy the requirements specified above.

表面上看这2个API没有什么大的差别,使用上也一样, 那为什么要搞二个呢? 好奇怪!

好, 让我们来好好研究下典型应用.

这2个API都返回 TimerRef. 用户可以用这个TimerRef来取消定时器. 唯一的差别是在超时的时候发送的消息不同: send_after是Msg, start_timer是{timeout, TimerRef, Msg}.
问题就出在取消timer的时候. 如果这个timer还没有超时的时候, 那么取消就没问题. 如果超时了麻烦就来了, 这个消息已经有可能已经被放到目标进程的消息队列里,等待派遣处理了.

这时候send_after里面存放的是Msg, 那用户如何知道Msg是对于那个TimerRef的呢? 读者可能说, 那我可以在消息里面加入TimerRef. 这个主意不错, 但是问题是在send_after调用返回之前, 你是无法得到TimerRef, 当然也就无从构造这个消息, 那就无法处理这个可能的超时信息, 就会破坏逻辑.
所以erts version 5.4.11 引入了, start_timer来解决这个问题. 它是自动的在超时后, 要发送消息前, 在消息里面添加了{timeout, TimerRef, Msg}, 达到识别的目的.

结论: 文档里面一眼带过的东西, 其实是有很多设计方面的考虑, 要认真考虑它的存在的意义.

在大家的印象中, gen_tcp:send是个很朴素的函数, 一调用数据就喀嚓喀嚓到了对端. 这是个很大的误解, Erlang的otp文档写的很不清楚. 而且这个功能对于大部分的网络程序是至关重要的, 它的使用对否极大了影响了应用的性能. 我听到很多同学在抱怨erlang的性能低或者出了很奇怪的问题, 很多是由于对系统的不了解, 误用的. 我下面就来解刨下, 文章很长, 而且需要读者熟悉erlang和底层的知识, 跟我来吧.

这篇文章是基于Erlang R13B04这个版本写的.

以下是从gen_tcp文档中摘抄的:

gen_tcp:send(Socket, Packet) -> ok | {error, Reason}
* Socket = socket()
* Packet =

[char()] | binary()
* Reason = posix()
* Sends a packet on a socket.

There is no send call with timeout option, you use the send_timeout socket option if timeouts are desired. See the examples section.

典型的使用如下:

client(PortNo,Message) ->
{ok,Sock} = gen_tcp:connect("localhost",PortNo,[{active,false},
{packet,2}]),
gen_tcp:send(Sock,Message),
A = gen_tcp:recv(Sock,0),
gen_tcp:close(Sock),
A.

很简单是把? 乍一看确实很简单, 但是这是迷惑人的开始.

我们上源代码:

lib/kernel/src/gen_tcp.erl

124send(S, Packet) when is_port(S) ->    %这里可以看出 S是个port
125    case inet_db:lookup_socket(S) of
126        {ok, Mod} ->                  %Mod可能是inet_tcp.erl 或者  inet6_tcp.erl
127            Mod:send(S, Packet);
128        Error ->
129            Error
130    end.

lib/kernel/src/inet_tcp.erl

 49send(Socket, Packet, Opts) -> prim_inet:send(Socket, Packet, Opts). %转给prim_inet模块
 50send(Socket, Packet) -> prim_inet:send(Socket, Packet, []).

erts/preloaded/src/prim_inet.erl

 360send(S, Data, OptList) when is_port(S), is_list(OptList) ->
 361    ?DBG_FORMAT("prim_inet:send(~p, ~p)~n", [S,Data]),
 362    try erlang:port_command(S, Data, OptList) of     <strong>%推给底层的port模块来处理</strong>
 363        false -> % Port busy and nosuspend option passed
 364            ?DBG_FORMAT("prim_inet:send() -> {error,busy}~n", []),
 365            {error,busy};
 366        true -> <strong>% Port模块接受数据</strong>
 367            receive
 368                {inet_reply,S,Status} ->  <strong>%阻塞, 等待回应</strong>
 369                    ?DBG_FORMAT("prim_inet:send() -> ~p~n", [Status]),
 370                    Status
 371            end
 372    catch
 373        error:_Error ->
 374            ?DBG_FORMAT("prim_inet:send() -> {error,einval}~n", []),
 375             {error,einval}
 376    end.
 377
 378send(S, Data) ->
 379    send(S, Data, []).

从上面这几段代码我们可以看出,当我们调用gen_tcp:send的时候, kernel模块会根据gen_tcp socket的类型决定调用相应的模块. 这个模块要么是inet_tcp, 要么是inet6_tcp. 这个模块会把发送请求委托给
prim_inet模块. prim_inet模块首先检查Socket是否合法, 如果合法然后调用erlang:port_command把系统推到ERTS运行期.
这个推的结果有2个: 1. 成功, 进程挂起等待运行期的反馈. 2. 失败,立即返回.
什么情况下会失败呢?
1. 驱动不支持soft_busy, 但是我们用了force标志
2. 驱动已经busy了, 但是我们不允许进程挂起.
Read more…