erlang_mysql_driver源码分析(三)

erlang

#mysql_recv:start_link

1
2
3
4
5
init(Host, Port, User, Password, Database, LogFun, Encoding, PoolId, Parent) ->
    case mysql_recv:start_link(Host, Port, LogFun, self()) of
	{ok, RecvPid, Sock} ->
	    case mysql_init(Sock, RecvPid, User, Password, LogFun) of
		{ok, Version} ->

上次在mysql_conn:init里面调用了mysql_recv:start_link,我们知道返回一个RecvPid接收进程的Pid和Sock套接字。mysql_recv:start_link过程如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
start_link(Host, Port, LogFun, Parent) when is_list(Host), is_integer(Port) ->
    RecvPid =
	spawn_link(fun () ->
			   init(Host, Port, LogFun, Parent)
		   end),
    %% wait for the socket from the spawned pid
    receive
	{mysql_recv, RecvPid, init, {error, E}} ->
	    {error, E};
	{mysql_recv, RecvPid, init, {ok, Socket}} ->
	    {ok, RecvPid, Socket}
    after ?CONNECT_TIMEOUT ->
	    catch exit(RecvPid, kill),
	    {error, "timeout"}
    end.

与mysql_conn大体相同，同样spawn一个进程，将mysql_conn的Pid传入当做Parent父进程，然后mysql_conn等待receive消息，如果spawn的mysql_recv进程返回{mysql_recv, RecvPid, init, {ok, Socket}},就代表创建成功，然后将RecvPid, Socket这些有用的变量返回，以便于mysql_init进行,
如果超过?CONNECT_TIMEOUT的时间，则会调用exit(RecvPid, kill)关闭mysql_recv进程，并返回{error, "timeout"}.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
init(Host, Port, LogFun, Parent) ->
    case gen_tcp:connect(Host, Port, [binary, {packet, 0}]) of
	{ok, Sock} ->
	    Parent ! {mysql_recv, self(), init, {ok, Sock}},
	    State = #state{socket  = Sock,
			   parent  = Parent,
			   log_fun = LogFun,
			   data    = <<>>
			  },
	    loop(State);
	E ->
	    LogFun(?MODULE, ?LINE, error,
		   fun() ->
			   {"mysql_recv: Failed connecting to ~p:~p : ~p",
			    [Host, Port, E]}
		   end),
	    Msg = lists:flatten(io_lib:format("connect failed : ~p", [E])),
	    Parent ! {mysql_recv, self(), init, {error, Msg}}
    end.

init里面通过gen_tcp:connect连接了端口，返回套接字，将这个套接字发送给父进程也就是在上面receive的地方，之后进入loop。如果连接失败，打印错误，并通知父进程。

state{socket, %套接字
parent, %父进程pid
log_fun, %打印日志函数
data, %端口发来的数据
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
loop(State) ->
    Sock = State#state.socket,
    receive
	{tcp, Sock, InData} ->
	    NewData = list_to_binary([State#state.data, InData]),
	    %% send data to parent if we have enough data
	    Rest = sendpacket(State#state.parent, NewData),
	    loop(State#state{data = Rest});
	{tcp_error, Sock, Reason} ->
	    LogFun = State#state.log_fun,
	    LogFun(?MODULE, ?LINE, error,
		   fun() ->
			   {"mysql_recv: Socket ~p closed : ~p",
			    [Sock, Reason]}
		   end),
	    State#state.parent ! {mysql_recv, self(), closed, {error, Reason}},
	    error;
	{tcp_closed, Sock} ->
	    LogFun = State#state.log_fun,
	    LogFun(?MODULE, ?LINE, debug,
		   fun() ->
			   {"mysql_recv: Socket ~p closed", [Sock]}
		   end),
	    State#state.parent ! {mysql_recv, self(), closed, normal},
	    error
    end.

gen_tcp:connect没有指定{active, false},这样的话，端口可以主动发给recv进程消息。实际上在在gen_tcp:connect之后我们得到Socket的同时，mysql_recv进程的消息队列里已经有Socket的发来的消息了。我们在shell里面可以试一下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
1> gen_tcp:connect("localhost", 3306, [binary, {packet, 0}]).
{ok,#Port<0.595>}
2> rp(process_info(self())).
[{current_function,{erl_eval,do_apply,6}},
 {initial_call,{erlang,apply,2}},
 {status,running},
 {message_queue_len,2},
 {messages,[{tcp,#Port<0.595>,
                 <<91,0,0,0,10,53,46,53,46,51,55,45,48,117,98,117,110,116,
                   117,48,46,49,52,46,48,52,46,49,0,187,1,0,0,57,98,97,67,
                   104,98,52,123,0,255,247,8,2,0,15,128,21,0,0,0,0,0,0,0,
                   0,0,0,50,36,92,38,67,96,59,67,125,112,59,115,0,109,121,
                   115,113,108,95,110,97,116,105,118,101,95,112,97,115,
                   115,119,111,114,100,0>>},
            {tcp_closed,#Port<0.595>}]},
 {links,[<0.26.0>]},
 {dictionary,[]},
 {trap_exit,false},
 {error_handler,error_handler},
 {priority,normal},
 {group_leader,<0.25.0>},
 {total_heap_size,3571},
 {heap_size,987},
 {stack_size,37},
 {reductions,4070},
 {garbage_collection,[{min_bin_vheap_size,46368},
                      {min_heap_size,233},
                      {fullsweep_after,65535},
                      {minor_gcs,5}]},
 {suspending,[]}]
ok

我们所做的操作以mysql_recv:init里面是一样的。打开了3306端口。这是shell的messagebox里面已经有了两天Socket发来的消息。loop函数将这段消息交给了sendpacket处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
sendpacket(Parent, Data) ->
    case Data of
	<<Length:24/little, Num:8, D/binary>> ->
	    if
		Length =< size(D) ->
		    {Packet, Rest} = split_binary(D, Length),
		    Parent ! {mysql_recv, self(), data, Packet, Num},
		    sendpacket(Parent, Rest);
		true ->
		    Data
	    end;
	_ ->
	    Data
    end.

packet头3个字节表示数据长度，第四个字节是帧序号，之后所有的是数据。那么这段数据头三个<<91,0,0>>代表数据长度是91字节（小端存储），之后的0代表这个帧序号是0，<<10....>>就是真正的数据。sendpacket将这段数据发给了parent。在如下的地方匹配

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
init(Host, Port, User, Password, Database, LogFun, Encoding, PoolId, Parent) ->
    case mysql_recv:start_link(Host, Port, LogFun, self()) of
	{ok, RecvPid, Sock} ->
	    case mysql_init(Sock, RecvPid, User, Password, LogFun) of
		{ok, Version} ->
------
mysql_init(Sock, RecvPid, User, Password, LogFun) ->
    case do_recv(LogFun, RecvPid, undefined) of
	{ok, Packet, InitSeqNum} ->
		    {Version, Salt1, Salt2, Caps} = greeting(Packet, LogFun),
-----
do_recv(LogFun, RecvPid, SeqNum)  when is_function(LogFun);
				       LogFun == undefined,
				       SeqNum == undefined ->
    receive
        {mysql_recv, RecvPid, data, Packet, Num} ->
	    {ok, Packet, Num};
	{mysql_recv, RecvPid, closed, _E} ->
	    {error, "mysql_recv: socket was closed"}
    end;

在mysql_conn的do_recv完成对消息的匹配。
上面的数据其实就是

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
=INFO REPORT==== 10-Jun-2015::11:59:21 ===
I(<0.116.0>:mysql_conn:620) : greeting packet <<10,53,46,53,46,51,55,45,48,117,
                                                98,117,110,116,117,48,46,49,52,
                                                46,48,52,46,49,0,173,1,0,0,100,
                                                44,77,124,75,63,60,78,0,255,
                                                247,8,2,0,15,128,21,0,0,0,0,0,
                                                0,0,0,0,0,84,60,110,54,95,93,
                                                55,57,88,46,122,66,0,109,121,
                                                115,113,108,95,110,97,116,105,
                                                118,101,95,112,97,115,115,119,
                                                111,114,100,0>> version "5.5.37-0ubuntu0.14.04.1" (protocol 10) salt "d,M|K?<N" caps 63487 serverchar <<8,2,0,
                                                                                                                                                        15,128,
                                                                                                                                                        21,0,0,
                                                                                                                                                        0,0,0,0,
                                                                                                                                                        0,0,0,0>>salt2 "T<n6_]79X.zB"

greeting函数用来转化为版本号，salt等参数的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
greeting(Packet, LogFun) ->
    <<Protocol:8, Rest/binary>> = Packet,
    {Version, Rest2} = asciz(Rest),
    <<_TreadID:32/little, Rest3/binary>> = Rest2,
    {Salt, Rest4} = asciz(Rest3),
    <<Caps:16/little, Rest5/binary>> = Rest4,
    <<ServerChar:16/binary-unit:8, Rest6/binary>> = Rest5,
    {Salt2, _Rest7} = asciz(Rest6),
    ?Log2(LogFun, debug,
	  "greeting packet ~p version ~p (protocol ~p) salt ~p caps ~p serverchar ~p"
	  "salt2 ~p",
	  [Packet, Version, Protocol, Salt, Caps, ServerChar, Salt2]),
    {normalize_version(Version, LogFun), Salt, Salt2, Caps}.
------------------------------------------
asciz(Data) when is_binary(Data) ->
    mysql:asciz_binary(Data, []).
-----------------------------------------
asciz_binary(<<>>, Acc) ->
    {lists:reverse(Acc), <<>>};
asciz_binary(<<0:8, Rest/binary>>, Acc) ->
    {lists:reverse(Acc), Rest};
asciz_binary(<<C:8, Rest/binary>>, Acc) ->
    asciz_binary(Rest, [C | Acc]).

Packet第一个字节是protocol,之后asciz函数将剩下的二进制以0位分界点分开。那么
Packet被分成了

Protocol : <<10>>
Version:<<53,46,53,46,51,55,45,48,117,98,117,110,116,117,48,46,49,52,46,48,52,46,49>>,
Salt:<<100,44,77,124,75,63,60,78>>
Caps:<<255,247>>
ServerChar:<<15,128,21,0,0,0,0,0,0,0,0,0,0,84,60,110>>(ServerChar是16个字节不是16个bit)
Salt2:<<97,116,105,118,101,95,112,97,115,115,119,111,114,100>>

在shell中检测下

1
2
3
4
5
6
7
8
4> <<53,46,53,46,51,55,45,48,117,98,117,110,116,117,48,46,49,52,46,48,52,46,49>>.
<<"5.5.37-0ubuntu0.14.04.1">>
5> <<100,44,77,124,75,63,60,78>>.
<<"d,M|K?<N">>
6> <<Caps:16/little>> = <<255,247>>.
<<"ÿ÷">>
7> Caps.
63487

和上面打印的log是一样的。