转载

Erlang pool management -- RabbitMQ worker_pool 2

上一篇已经分析了rpool 的三个module , 以及简单的物理关系. 这次主要分析用户进程和 worker_pool 进程还有worker_pool_worker 进程之间的调用关系. 在开始之前, 必须先明确一点, 就是一个worker_pool_worker 进程只有在处理完一个用户进程的任务之后才能开始处理另一用户进程的任务.

worker_pool 状态管理

在上一篇已经说明, worker_pool 管理了rpool 工作进程的ready idle busy 状态,从worker_pool 的代码可以看出, 对于idle 状态和busy 状态的处理逻辑是相同的.

1 handle_call({next_free, CPid}, _From, State = #state { available = 2                                                            [WPid | Avail1] }) -> 3     worker_pool_worker:next_job_from(WPid, CPid), 4     {reply, WPid, State #state { available = Avail1 }, hibernate};

以上是submit 请求在 available queue 不为空时的处理逻辑.

1 handle_cast({run_async, Fun}, State = #state { available = [WPid | Avail1] }) -> 2     worker_pool_worker:submit_async(WPid, Fun), 3     {noreply, State #state { available = Avail1 }, hibernate};

而以上是submit_async 请求在available queue 不为空时的处理逻辑.

 1 handle_cast({idle, WPid}, State = #state { available = Avail,  2                                            pending   = Pending }) ->  3     {noreply,  4      case queue:out(Pending) of  5          {empty, _Pending} ->  6              State #state { available = ordsets:add_element(WPid, Avail) };  7          {{value, {next_free, From, CPid}}, Pending1} ->  8              worker_pool_worker:next_job_from(WPid, CPid),  9              gen_server2:reply(From, WPid), 10              State #state { pending = Pending1 }; 11          {{value, {run_async, Fun}}, Pending1} -> 12              worker_pool_worker:submit_async(WPid, Fun), 13              State #state { pending = Pending1 } 14      end, hibernate};

当某工作进程完成对用户进程任务的处理之后, cast 给worker_pool 进程 idle 消息, worker_pool 会:

1, 判断pending 是否为空, 若空, 则将工作进程add 到available中;

2, 若pending 为 next_free, 则执行submit 请求时的处理流程;

3, 若pending 为run_async, 则执行submit_async 请求时的处理流程.

submit

Erlang pool management -- RabbitMQ worker_pool 2

用户进程在调用submit 请求时的基本流程如上图所示:

1, 用户进程call 请求 worker_pool 进程 next_free

2, worker_pool 进程获取available 中的工作进程(PidA)并返回给用户进程

3, 用户进程call 请求PidA 工作进程submit

工作进程PidA 在ready 或 idle 时,工作进程的state 信息就会被置为undefined .

也就是在first message 到达工作进程时, 工作进程的state 信息为undefined, 那么工作进程处理上图first message 的逻辑为:

1 handle_cast({next_job_from, CPid}, undefined) -> 2     MRef = erlang:monitor(process, CPid), 3     {noreply, {from, CPid, MRef}, hibernate};

也就是 monitor 用户进程UPA, 并将自身的state 重置为 {from, CPid, MRef}, 然后等待UPA 进程的submit 请求, 也就是second message .

如果在这中间, UPA 异常退出, 工作进程就会收到{'DOWN' ... } message:

1 handle_info({'DOWN', MRef, process, CPid, _Reason}, {from, CPid, MRef}) -> 2     ok = worker_pool:idle(self()), 3     {noreply, undefined, hibernate};

然后将state 重置为undefined .

submit_async

Erlang pool management -- RabbitMQ worker_pool 2

用户进程在调用submit_async 请求时的基本流程如上图所示:

1, 用户进程cast 请求worker_pool 进程 run_async 参数是需要execute Fun

2, worker_pool 进程获取available 中的工作进程(PidA)并 cast submit_async 请求给PidA 参数为PidA 和 Fun

3, 然后进程在handle_cast callback 中进行处理.

1 handle_cast({submit_async, Fun}, undefined) -> 2     run(Fun), 3     ok = worker_pool:idle(self()), 4     {noreply, undefined, hibernate};

和submit 操作相比, 工作进程在处理submit_async 请求时,不需要monitor 用户进程(UPA), 不需要将Fun execute 执行的结果返回用户进程.

总结

在Mac 下omnigraffle 真的挺好用的,就是太贵了. :(

正文到此结束

所属分类：编程技术

本文标签： 管理 UI 进程总结参数 value 代码
版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
本文海报： 生成海报一生成海报二

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

Erlang pool management -- RabbitMQ worker_pool 2

worker_pool 状态管理

submit

submit_async

总结

热门推荐

相关文章

说给你听

本文目录

随机标签

书籍教程

近期评论

网站信息

其他链接

关于本站

问题交流