mysqlpool_ErlangpoolmanagementEmysqlpool

从这篇开始,这一系列主要分析在开源社区中,Erlang 相关pool 的管理和使用.

在开源社区,Emysql 是Erlang 较为受欢迎的一个MySQL 驱动. Emysql 对pool 的管理和使用是非常典型的,pool 的管理角色中,主要有available(记录当前pool 中可供使用的成员),locked(记录当前pool 中正在被使用的成员),waiting(记录当前正在处理等待该pool 的用户).用户进程在使用pool 过程中, pool 中的成员在这三个角色中来回迁移.

pool 数据结构

Emysql pool 的数据结构如下:

1 -record(pool, {pool_id :: atom(),2 size :: number(),3 user :: string(),4 password :: string(),5 host :: string(),6 port :: number(),7 database :: string(),8 encoding :: utf8 | latin1 | {utf8, utf8_unicode_ci} |{utf8, utf8_general_ci},9 available&＃61;queue:new() :: queue(),10 locked&＃61;gb_trees:empty() :: gb_tree(),11 waiting&＃61;queue:new() :: queue(),12 start_cmds&＃61;[] :: string(),13 conn_test_period&＃61;0:: number(),14 connect_timeout&＃61;infinity :: number() |infinity,15 warnings&＃61;false :: boolean()}).

L1 处的pool_id 为pool 的标识

L2 定义了pool 中成员的数量

L3 L4 为用以连接MySQL 数据库的用户名和密码

L5 L6 L7 L8 为连接MySQL 数据库的IP, 端口, 默认数据库, 编码

L9 用以记录当前pool 中可用的成员

L10 用以记录当前pool 中正在被使用的成员

L11 用以记录当前等待pool 中成员的用户

L12 为在与数据库建立连接后的初始化命令

L14 是用于gen_tcp:connect 时的超时参数

pool 添加操作

在Emysql 项目中,emysql module 定义了所有外部操作的API, 其中添加操作的API有:

1, add_pool/2

2, add_pool/8

3, add_pool/9

以下代码片段为add_pool 的实质性处理逻辑:

1 add_pool(#pool{pool_id&＃61;PoolId,size&＃61;Size,user&＃61;User,password&＃61;Password,host&＃61;Host,port&＃61;Port,2 database&＃61;Database,encoding&＃61;Encoding,start_cmds&＃61;StartCmds,3 connect_timeout&＃61;ConnectTimeout,warnings&＃61;Warnings}&＃61;PoolSettings)->

4 config_ok(PoolSettings),5 case emysql_conn_mgr:has_pool(PoolId) of

6 true ->

7 {error,pool_already_exists};8 false ->

9 Pool &＃61;#pool{10 pool_id &＃61;PoolId,11 size &＃61;Size,12 user &＃61;User,13 password &＃61;Password,14 host &＃61;Host,15 port &＃61;Port,16 database &＃61;Database,17 encoding &＃61;Encoding,18 start_cmds &＃61;StartCmds,19 connect_timeout &＃61;ConnectTimeout,20 warnings &＃61;Warnings21 },22 Pool2 &＃61; case emysql_conn:open_connections(Pool) of

23 {ok, Pool1} ->Pool1;24 {error, Reason} ->throw(Reason)25 end,26 emysql_conn_mgr:add_pool(Pool2)27 end.

处理逻辑主要有:

1, 确认参数的数据类型

2, 检查当前是否已经有相同ID的 pool

3, 与MySQL server 建立connection

4, 在emysql_conn_mgr 中添加 该pool

在当前的Emysql 项目中,emysql_conn_mgr 是用来管理所有pool 的gen_server 进程. 对于所有的pool 而言,其内部成员的状态管理, 都是由emysql_conn_mgr 调度的, 包括某个pool 中connection 成员的使用,归还等.

确认参数的数据类型

确认参数的数据类型主要使用erlang:is_{{type}} bif func, 在Erlang VM 内存,对于每种数据类型都是以后缀来识别的.

1 config_ok(#pool{pool_id&＃61;PoolId,size&＃61;Size,user&＃61;User,password&＃61;Password,host&＃61;Host,port&＃61;Port,2 database&＃61;Database,encoding&＃61;Encoding,start_cmds&＃61;StartCmds,3 connect_timeout&＃61;ConnectTimeout,warnings&＃61;Warnings})4 whenis_atom(PoolId),5 is_integer(Size),6 is_list(User),7 is_list(Password),8 is_list(Host),9 is_integer(Port),10 is_list(Database) orelse Database &＃61;&＃61;undefined,11 is_list(StartCmds),12 is_integer(ConnectTimeout) orelse ConnectTimeout &＃61;&＃61;infinity,13 is_boolean(Warnings) ->

14 encoding_ok(Encoding);15 config_ok(_BadOptions) ->

16 erlang:error(badarg).17

18 encoding_ok(Enc) when is_atom(Enc) ->ok;19 encoding_ok({Enc, Coll}) when is_atom(Enc), is_atom(Coll) ->ok;20 encoding_ok(_) -> erlang:error(badarg).

因此, 这部分的操作足够高效.以 is_list/1为例:

1 #define TAG_PRIMARY_LIST 0x1

2 #define is_list(x) (((x) & _TAG_PRIMARY_MASK) &＃61;&＃61; TAG_PRIMARY_LIST)

3 #define is_not_list(x) (!is_list((x)))

与MySQL server 建立链接

此处与MySQL server 建立链接是调用emysql_conn module 中的open_connections 函数.

1 %%&＃64;doc Opens connections for the necessary pool.

2 %%3 %%If connection opening fails, removes all connections from the pool

4 %%Does not remove pool from emysql_conn_mgr due to a possible deadlock.

5 %%Caller must do it by itself.

6 open_connections(Pool) ->

7 %-% io:format("open connections loop: .. "),8 case (queue:len(Pool#pool.available) &＃43; gb_trees:size(Pool#pool.locked))

9 true ->

10 case catch open_connection(Pool) of

11 #emysql_connection{} &＃61; Conn ->

12 open_connections(Pool#pool{available &＃61;queue:in(Conn, Pool#pool.available)});13 {&＃39;EXIT&＃39;, Reason} ->

14 AllConns &＃61;lists:append(15 queue:to_list(Pool#pool.available),16 gb_trees:values(Pool#pool.locked)17 ),18 lists:foreach(fun emysql_conn:close_connection/1, AllConns),19 {error, Reason}20 end;21 false ->

22 {ok, Pool}23 end.

链接的总数为pool 结构中的size 字段(L8),成功建立链接后,将Conn 放入 available queue 中(L12).

在emysql_conn_mgr 中添加 pool

当pool 与MySQL server 建立链接完成后,需要将pool 添加到emysql_conn_mgr 中, 以便emysql_conn_mgr gen_server 进程对pool 进行管理.

添加pool add_pool/1 的操作:

1 add_pool(Pool) ->

2 do_gen_call({add_pool, Pool}).3

4 ...5

6 handle_call({add_pool, Pool}, _From, State) ->

7 case find_pool(Pool#pool.pool_id, State#state.pools) of

8 {_, _} ->

9 {reply, {error, pool_already_exists}, State};10 undefined ->

11 {reply, ok, State#state{pools &＃61; [Pool|State#state.pools]}}12 end;

如果当前emysql_conn_mgr gen_server 进程中,并未记录(L8)该pool 的话,就将该pool 添加到emysql_conn_mgr gen_server 进程的state 数据中(L11).

has_pool/1 的操作:

1 has_pool(Pool) ->

2 do_gen_call({has_pool, Pool}).3

4 ....5

6 handle_call({has_pool, PoolID}, _From, State) ->

7 case find_pool(PoolID, State#state.pools) of

8 {_, _} ->

9 {reply, true, State};10 undefined ->

11 {reply, false, State}12 end;

以上,即为add_pool 操作的整个流程. emysql_conn_mgr gen_server 进程是管理pool 的非常重要的进程.

pool 使用管理

取出connection

在execute 执行一条SQL语句时, 用户进程需要先请求emysql_conn_mgr gen_server 进程从给定pool_id 的pool 中取出一个成员.

1 execute(PoolId, Query, Args, Timeout) when (is_list(Query) orelse is_binary(Query)) andalso is_list(Args) andalso (is_integer(Timeout) orelse Timeout &＃61;&＃61; infinity) ->

2 Connection &＃61;emysql_conn_mgr:wait_for_connection(PoolId),3 monitor_work(Connection, Timeout, [Connection, Query, Args]);

1 wait_for_connection(PoolId ,Timeout)->

2 %%try to lock a connection. if no connections are available then

3 %%wait to be notified of the next available connection

4 %-% io:format("~p waits for connection to pool ~p~n", [self(), PoolId]),5 case do_gen_call({lock_connection, PoolId, true, self()}) of

6 unavailable ->

7 %-% io:format("~p is queued~n", [self()]),8 receive

9 {connection, Connection} ->Connection10 after Timeout ->

11 do_gen_call({abort_wait, PoolId}),12 receive

13 {connection, Connection} ->Connection14 after

15 0 ->exit(connection_lock_timeout)16 end

17 end;18 Connection ->

19 %-% io:format("~p gets connection~n", [self()]),20 Connection21 end.

然后对于wait_for_connection/2 函数的操作, 首先会调用emysql_conn_mgr gen_server handle_call 操作 lock_connection

而lock_next_connection 函数的主要功能是从pool 的available queue 中out 一个元素, 并monitor 调用进程(以防调用进程异常退出而没有归还conn).

1 lock_next_connection(Available ,Locked, Who) ->

2 case queue:out(Available) of

3 {{value, Conn}, OtherAvailable} ->

4 MonitorRef &＃61;erlang:monitor(process, Who),5 NewConn &＃61;connection_locked_at(Conn, MonitorRef),6 MonitorTuple &＃61;{MonitorRef,7 {NewConn#emysql_connection.pool_id, NewConn#emysql_connection.id}},8 NewLocked &＃61;gb_trees:enter(NewConn#emysql_connection.id, NewConn, Locked),9 {ok, NewConn, OtherAvailable, NewLocked, MonitorTuple};10 {empty, _} ->

11 unavailable12 end.

L2 处会尝试从available queue 中取出元素conn, 如果queue 此时不为空, emysql_conn_mgr 进程就会monitor (L4)用户进程,然后将该conn gb_tress enter 到locked tree 中(L8).

在"无可用的conn"的情况下, emysql_conn_mgr gen_server 进程会将用户进程写入到pool 的waiting queue中, 并且返回&＃39;unavailable&＃39;, 用户进程就会等待conn 的其他使用者归还conn .

归还connection

在用户使用完conn 之后,应该及时归还给pool, 以防链接资源泄露.

在 emysql_conn_mgr module 中定义了pass_connection/1 函数以及在 handl_call callback 中实现了 handle_call({{replace_connection, Kind}, OldConn, NewConn}, _From, State).

在handle_call callback 函数中, 首先会从locked tree 中delete 掉该conn.然后从waiting queue 中取出之前等待的用户进程ID, 将conn 发送给alive 的等待进程, 并更新locked tree waiting queue. 如果waiting queue 中无alive 等待进程, 就将conn 还回给available queue, 并更新相关的管理角色.

1 serve_waiting_pids(Waiting, Available, Locked, MonitorRefs) ->

2 case queue:is_empty(Waiting) of

3 false ->

4 Who &＃61;queue:get(Waiting),5 case lock_next_connection(Available, Locked, Who) of

6 {ok, Connection, OtherAvailable, NewLocked, NewRef} ->

7 {{value, Pid}, OtherWaiting} &＃61;queue:out(Waiting),8 case erlang:is_process_alive(Pid) of

9 true ->

10 erlang:send(Pid, {connection, Connection}),11 serve_waiting_pids(OtherWaiting, OtherAvailable, NewLocked, [NewRef |MonitorRefs]);12 _ ->

13 serve_waiting_pids(OtherWaiting, Available, Locked, MonitorRefs)14 end;15 unavailable ->

16 {Waiting, Available, Locked, MonitorRefs}17 end;18 true ->

19 {Waiting, Available, Locked, MonitorRefs}20 end.

conn 使用进程退出

如果某用户进程在从pool 中取出conn 使用, 但是在使用过程中, 用户进程异常退出, 无法调用pass_connection/1 函数归还conn , 就会出现资源泄露的问题.

在emysql_conn_mgr module 中, 是使用monitor 用户进程的方式处理的. 在用户进程获得一个conn 之后, emysql_conn_mgr 会使用BIF erlang:monitor/2 函数 monitor 用户进程.当用户进程异常退出后,emysql_conn_mgr 进程就会收到&＃39;DOWN&＃39; 的message.然后在emysql_conn_mgr module 中的handle_info callback 函数中处理:

1 handle_info({&＃39;DOWN&＃39;, MonitorRef, _, _, _}, State) ->

2 case dict:find(MonitorRef, State#state.lockers) of

3 {ok, {PoolId, ConnId}} ->

4 case find_pool(PoolId, State#state.pools) of

5 {Pool, _} ->

6 case gb_trees:lookup(ConnId, Pool#pool.locked) of

7 {value, Conn} ->async_reset_conn(State#state.pools, Conn);8 _ ->ok9 end;10 _ ->

11 ok12 end;13 _ ->

14 ok15 end,16 {noreply, State};

可以看出,在emysql_conn_mgr 进程接收到&＃39;DOWN&＃39; 的message 之后, 会在进程dict 的locker 中 查找poolID和 connID, 继而重置conn .

等待进程 abort_wait

某进程在获取pool 中 conn 时, 在Timeout 之后, 用户进程会调用abort_wait, emysql_conn_mgr 进程就会从waiting queue 中, 将不再等待的用户进程remove

1 handle_call({abort_wait, PoolId}, {From, _Mref}, State) ->

2 case find_pool(PoolId, State#state.pools) of

3 {Pool, OtherPools} ->

4 %%Remove From from the wait queue

5 QueueNow &＃61;queue:filter(6 fun(Pid) -> Pid &＃61;/&＃61; From end,7 Pool#pool.waiting),8 PoolNow &＃61; Pool#pool{ waiting &＃61;QueueNow },9 %%See if the length changed to know if From was removed.

10 OldLen &＃61;queue:len(Pool#pool.waiting),11 NewLen &＃61;queue:len(QueueNow),12 if

13 OldLen &＃61;:&＃61; NewLen ->

14 Reply &＃61;not_waiting;15 true ->

16 Reply &＃61;ok17 end,18 {reply, Reply, State#state{pools&＃61;[PoolNow|OtherPools]}};19 undefined ->

20 {reply, {error, pool_not_found}, State}21 end;

这样的设计同样是为了尽可能的保证避免资源的泄露,试想如果emysql_conn_mgr 进程将conn 发送给已经不再等待(不再需要)的进程,那该conn 就不可能再被归还.

在大多数情况下,这种设计是可以保证conn 不会被发送给不再等待的用户进程,但是在达成"gen_server 进程的处理是顺序性"这样共识的前提下考虑以下情况:

也就是在处理pass_connection , emysql_conn_mgr 进程从waiting queue 取出了ProcessA进程的同时, ProcessA 进程因为Timeout 调用了abort_wait 且exit 退出. emysql_conn_mgr 的顺序性处理,使得只有在处理完pass_connection 之后, 才能处理abort_wait 操作. 最终导致的结果就是将conn 发送给已经exit (但是还没有从waiting queue remove) 的用户进程ProcessA, 是conn 不能再被归还.

当这种情况出现的时候,就应该在发送conn 给 ProcessA 进程之前, 判断ProcessA 是否alive(这个pr 已经被merged了).

总结

在Emysql 的pool 管理中,主要使用了:

1, available queue 记录所有可用的pool 成员

2, locked tree 记录所有正在被使用的pool 成员

3, waiting queue 记录所有正在等待的用户进程

4, monitor 所有正在使用pool 成员的用户进程, 处理异常退出的case

5, 处理等待进程的abort_wait 请求, 更新waiting queue

6, 在发送pool 成员之前, 应该判断用户进程是否alive, 防止资源泄露

遗漏

现在的emysql_conn_mgr gen_server 进程属于单点,也就是所有的pool 的管理调度都是由一个进程来完成.

------------------------------

觉得写的还可以&＃xff0c;就扫个码&＃xff0c;打个赏呗。