06 | Swoole 源码分析之 Coroutine 协程模块

2024-04-03 11:52
文章标签 分析 模块 源码 swoole 06 协程 coroutine

本文主要是介绍06 | Swoole 源码分析之 Coroutine 协程模块，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

首发原文链接：Swoole 源码分析之 Coroutine 协程模块
大家好，我是码农先森。

引言

协程又称轻量级线程，但与线程不同的是；协程是用户级线程，不需要操作系统参与。由用户显式控制，可以在需要的时候挂起、或恢复执行。

通过协程程序可以在执行的过程中保存当前的状态，并在恢复后从该状态处继续执行，整体上来说创建、销毁、切换的成本低。

但在 Swoole 中的协程是无法利用多核 CPU 的，如果想利用多核 CPU 则需要依赖 Swoole 的多进程模型。

协程的出现为 Swoole 程序提升并发效率、及系统的处理能力，注入了强劲的动力；可以说是 Swoole 作为高性能通信框架的的核心模块。

源码拆解

这次我们以下面这段代码，来作为本次拆解源码的切入点。

// 协程容器
Swoole\Coroutine\run(function () {// Socket 协程客户端$socket = new Swoole\Coroutine\Socket(AF_INET, SOCK_STREAM, 0);// 建立连接，在建立连接的过程中会发生协程切换$retval = $socket->connect('127.0.0.1', 9601);if ($retval) {// 发送数据，在发送数据的过程中会发生协程切换$n = $socket->send('hello');var_dump($n);// 解释数据，在接收数据的过程中会发生协程切换$data = $socket->recv();var_dump($data);// 关闭连接$socket->close();}
});

这段代码主要是使用 Socket 的协程客户端与本地的 9601 端口建立连接，并且发送、接收数据。在分析源码之前，我对这次的源码做了一个图解梳理，把整个调用链路上的函数串联了起来。我们可以先对整体有个大致的了解，便于后面分析源代码。

Socket 协程客户端

Socket 协程客户端是专门用于 Swoole 在协程环境中使用的，可以实现在 IO 调用时切换协程，让出 CPU 的使用权。例如：在连接建立、发送数据、接收数据 等阶段会进行协程的切换。

这个函数主要是发起 Socket 连接的建立，并且在 wait_event 这个函数内部实现了协程的切换。

// swoole-src/src/coroutine/socket.cc:595
bool Socket::connect(const struct sockaddr *addr, socklen_t addrlen) {if (sw_unlikely(!is_available(SW_EVENT_RDWR))) {return false;}int retval;do {// 发起连接建立retval = ::connect(sock_fd, addr, addrlen);} while (retval < 0 && errno == EINTR);if (retval < 0) {if (errno != EINPROGRESS) {set_err(errno);return false;} else {TimerController timer(&write_timer, connect_timeout, this, timer_callback);// wait_event 这个函数内部实现了协程的切换if (!timer.start() || !wait_event(SW_EVENT_WRITE)) {if (is_closed()) {set_err(ECONNABORTED);}return false;} else {if (socket->get_option(SOL_SOCKET, SO_ERROR, &errCode) < 0 || errCode != 0) {set_err(errCode);return false;}}}}connected = true;set_err(0);return true;
}

再看看 wait_event 函数的内部实现，先是获取到当前的协程，然后根据事件的类型调用函数 add_event 将事件添加到事件管理的结构体中，最后将当前的协程切换出去，让出其 CPU 的控制权。

// swoole-src/src/coroutine/socket.cc:147
bool Socket::wait_event(const EventType event, const void **__buf, size_t __n) {EventType added_event = event;// 获取到当前的协程Coroutine *co = Coroutine::get_current_safe();if (!co) {return false;}if (sw_unlikely(socket->close_wait)) {set_err(SW_ERROR_CO_SOCKET_CLOSE_WAIT);return false;}// clear the last errCodeset_err(0);
#ifdef SW_USE_OPENSSL// 根据事件的类型调用函数 add_event 将事件添加到事件管理的结构体中if (sw_unlikely(socket->ssl && ((event == SW_EVENT_READ && socket->ssl_want_write) ||(event == SW_EVENT_WRITE && socket->ssl_want_read)))) {if (sw_likely(socket->ssl_want_write && add_event(SW_EVENT_WRITE))) {want_event = SW_EVENT_WRITE;} else if (socket->ssl_want_read && add_event(SW_EVENT_READ)) {want_event = SW_EVENT_READ;} else {return false;}added_event = want_event;} else
#endifif (sw_unlikely(!add_event(event))) {return false;}swoole_trace_log(SW_TRACE_SOCKET,"socket#%d blongs to cid#%ld is waiting for %s event",sock_fd,co->get_cid(),get_wait_event_name(this, event));Coroutine::CancelFunc cancel_fn = [this, event](Coroutine *co) { return cancel(event); };// 将当前的协程切换出去，让出其 CPU 的控制权if (sw_likely(event == SW_EVENT_READ)) {read_co = co;read_co->yield(&cancel_fn);read_co = nullptr;} else if (event == SW_EVENT_WRITE) {if (sw_unlikely(!zero_copy && __n > 0 && *__buf != get_write_buffer()->str)) {write_buffer->clear();if (write_buffer->append((const char *) *__buf, __n) != SW_OK) {set_err(ENOMEM);goto _failed;}*__buf = write_buffer->str;}write_co = co;write_co->yield(&cancel_fn);write_co = nullptr;} else {assert(0);return false;}
_failed:
#ifdef SW_USE_OPENSSL// maybe read_co and write_co are all waiting for the same event when we use SSLif (sw_likely(want_event == SW_EVENT_NULL || !has_bound()))
#endif{Reactor *reactor = SwooleTG.reactor;if (sw_likely(added_event == SW_EVENT_READ)) {reactor->remove_read_event(socket);} else {reactor->remove_write_event(socket);}}
#ifdef SW_USE_OPENSSLwant_event = SW_EVENT_NULL;
#endifswoole_trace_log(SW_TRACE_SOCKET,"socket#%d blongs to cid#%ld trigger %s event",sock_fd,co->get_cid(),get_trigger_event_name(this, added_event));return !is_closed() && !errCode;
}

同理 send() 和 recv() 函数，也和 connect() 函数是一样的实现方式。

// swoole-src/src/coroutine/socket.cc:847
ssize_t Socket::send(const void *__buf, size_t __n) {if (sw_unlikely(!is_available(SW_EVENT_WRITE))) {return -1;}ssize_t retval;TimerController timer(&write_timer, write_timeout, this, timer_callback);do {// 发送数据retval = socket->send(__buf, __n, 0);} while (retval < 0 && socket->catch_write_error(errno) == SW_WAIT && timer.start() &&wait_event(SW_EVENT_WRITE, &__buf, __n));check_return_value(retval);return retval;
}// swoole-src/src/coroutine/socket.cc:874
ssize_t Socket::recv(void *__buf, size_t __n) {if (sw_unlikely(!is_available(SW_EVENT_READ))) {return -1;}ssize_t retval;TimerController timer(&read_timer, read_timeout, this, timer_callback);do {// 接收数据retval = socket->recv(__buf, __n, 0);} while (retval < 0 && socket->catch_read_error(errno) == SW_WAIT && timer.start() && wait_event(SW_EVENT_READ));check_return_value(retval);return retval;
}

也是调用 wait_event() 函数来实现当前的协程切换，唯一的区别就是事件的类型不同，一个是读事件，一个是写事件。

Run 协程容器

在 Swoole 中要想使用协程，那么必须要在协程的环境中使用协程的客户端，或者支持 Hook 的原生 PHP 函数。才能享受到 Swoole 中协程带来的高性能，不然和普通的 PHP 执行程序没有什么区别，变成了同步阻塞。

在源码中协程容器主要是实现了事件循环的初始化、协程上下文的创建管理、事件循环的 IO 事件监听，接下来我们会主要分析关于事件管理的部分内容。

// swoole-src/src/coroutine/base.cc:210
namespace coroutine {bool run(const CoroutineFunc &fn, void *arg) {// 事件循环的初始化if (swoole_event_init(SW_EVENTLOOP_WAIT_EXIT) < 0) {return false;}// 协程上下文的创建管理Coroutine::activate();long cid = Coroutine::create(fn, arg);// 事件循环的 IO 事件监听swoole_event_wait();Coroutine::deactivate();return cid > 0;}
}

Event 事件初始化

Event 事件初始化主要是定义一些事件的回调函数，便于在事件被触发时恢复对应的协程进行后续的逻辑处理，例如：读事件回调函数 readable_event_callback、写事件回调函数 writable_event_callback 等。

// swoole-src/src/wrapper/event.cc:37
int swoole_event_init(int flags) {if (!SwooleG.init) {std::unique_lock<std::mutex> lock(init_lock);swoole_init();}// 创建一个 Reactor 实例对象Reactor *reactor = new Reactor(SW_REACTOR_MAXEVENTS);if (!reactor->ready()) {return SW_ERR;}if (flags & SW_EVENTLOOP_WAIT_EXIT) {reactor->wait_exit = 1;}// Socket 事件初始化coroutine::Socket::init_reactor(reactor);coroutine::System::init_reactor(reactor);network::Client::init_reactor(reactor);SwooleTG.reactor = reactor;return SW_OK;
}

// swoole-src/include/swoole_coroutine_sokcet.h:157
static inline void init_reactor(Reactor *reactor) {// 定义对应事件的回调函数reactor->set_handler(SW_FD_CO_SOCKET | SW_EVENT_READ, readable_event_callback);reactor->set_handler(SW_FD_CO_SOCKET | SW_EVENT_WRITE, writable_event_callback);reactor->set_handler(SW_FD_CO_SOCKET | SW_EVENT_ERROR, error_event_callback);
}

// swoole-src/src/coroutine/socket.c:48
int Socket::readable_event_callback(Reactor *reactor, Event *event) {Socket *socket = (Socket *) event->socket->object;socket->set_err(0);
#ifdef SW_USE_OPENSSLif (sw_unlikely(socket->want_event != SW_EVENT_NULL)) {if (socket->want_event == SW_EVENT_READ) {// 恢复对应的协程socket->write_co->resume();}} else
#endif{if (socket->recv_barrier && (*socket->recv_barrier)() && !event->socket->event_hup) {return SW_OK;}// 恢复对应的协程socket->read_co->resume();}return SW_OK;
}

Event 事件监听

Event 事件监听主要是针对被加入到事件循环中的 Socket 进行 IO 事件的监听，如果有读或写 IO 事件的触发，则回调到对应的处理函数上进行执行。

// swoole-src/src/warpper/event.cc:84
int swoole_event_wait() {Reactor *reactor = SwooleTG.reactor;int retval = 0;if (!reactor->wait_exit or !reactor->if_exit()) {// 事件循环等待调用retval = reactor->wait(nullptr);}swoole_event_free();return retval;
}

// swoole-src/src/reactor/epoll.cc:153
int ReactorEpoll::wait(struct timeval *timeo) {Event event;ReactorHandler handler;int i, n, ret;int reactor_id = reactor_->id;int max_event_num = reactor_->max_event_num;if (reactor_->timeout_msec == 0) {if (timeo == nullptr) {reactor_->timeout_msec = -1;} else {reactor_->timeout_msec = timeo->tv_sec * 1000 + timeo->tv_usec / 1000;}}reactor_->before_wait();while (reactor_->running) {if (reactor_->onBegin != nullptr) {reactor_->onBegin(reactor_);}// 监听 IO 事件n = epoll_wait(epfd_, events_, max_event_num, reactor_->get_timeout_msec());if (n < 0) {if (!reactor_->catch_error()) {swoole_sys_warning("[Reactor#%d] epoll_wait failed", reactor_id);return SW_ERR;} else {goto _continue;}} else if (n == 0) {reactor_->execute_end_callbacks(true);SW_REACTOR_CONTINUE;}for (i = 0; i < n; i++) {event.reactor_id = reactor_id;event.socket = (Socket *) events_[i].data.ptr;event.type = event.socket->fd_type;event.fd = event.socket->fd;if (events_[i].events & (EPOLLRDHUP | EPOLLERR | EPOLLHUP)) {event.socket->event_hup = 1;}// read 读事件，这里的 handler 对应 readable_event_callbackif ((events_[i].events & EPOLLIN) && !event.socket->removed) {handler = reactor_->get_handler(SW_EVENT_READ, event.type);ret = handler(reactor_, &event);if (ret < 0) {swoole_sys_warning("EPOLLIN handle failed. fd=%d", event.fd);}}// write 写事件，这里的 handler 对应 writable_event_callbackif ((events_[i].events & EPOLLOUT) && !event.socket->removed) {handler = reactor_->get_handler(SW_EVENT_WRITE, event.type);ret = handler(reactor_, &event);if (ret < 0) {swoole_sys_warning("EPOLLOUT handle failed. fd=%d", event.fd);}}// error 错误处理，这里的 handler 对应 error_event_callbackif ((events_[i].events & (EPOLLRDHUP | EPOLLERR | EPOLLHUP)) && !event.socket->removed) {// ignore ERR and HUP, because event is already processed at IN and OUT handler.if ((events_[i].events & EPOLLIN) || (events_[i].events & EPOLLOUT)) {continue;}handler = reactor_->get_error_handler(event.type);ret = handler(reactor_, &event);if (ret < 0) {swoole_sys_warning("EPOLLERR handle failed. fd=%d", event.fd);}}if (!event.socket->removed && (event.socket->events & SW_EVENT_ONCE)) {reactor_->_del(event.socket);}}_continue:reactor_->execute_end_callbacks(false);SW_REACTOR_CONTINUE;}return 0;
}

总结

  • 协程又称轻量级线程，协程是用户级线程；不需要操作系统参与，创建切换成本低。
  • Swoole 中的协程是无法利用多核 CPU 的，如果想利用多核 CPU 则需要依赖 Swoole 的多进程模型。
  • Swoole 中协程的是利用的 Event 事件循环进行调度的，将遇到 IO 操作的 Socket 统一加入到事件循环中。
  • 本次的源码分析旨在了解整个协程在 Swoole 中的运行逻辑，打开我们的思路，便于我们更好的体会到协程所带来的高性能价值。

在这里插入图片描述

这篇关于06 | Swoole 源码分析之 Coroutine 协程模块的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/872851。

相关文章

Springboot中分析SQL性能的两种方式详解

Springboot中分析SQL性能的两种方式详解

《Springboot中分析SQL性能的两种方式详解》文章介绍了SQL性能分析的两种方式：MyBatis-Plus性能分析插件和p6spy框架,MyBatis-Plus插件配置简单,适用于开发和测试环... 目录SQL性能分析的两种方式:功能介绍实现方式:实现步骤:SQL性能分析的两种方式:功能介绍记录
阅读更多...
最长公共子序列问题的深度分析与Java实现方式

最长公共子序列问题的深度分析与Java实现方式

《最长公共子序列问题的深度分析与Java实现方式》本文详细介绍了最长公共子序列（LCS）问题,包括其概念、暴力解法、动态规划解法,并提供了Java代码实现,暴力解法虽然简单,但在大数据处理中效率较低,... 目录最长公共子序列问题概述问题理解与示例分析暴力解法思路与示例代码动态规划解法DP 表的构建与意义动
阅读更多...
Python利用自带模块实现屏幕像素高效操作

Python利用自带模块实现屏幕像素高效操作

《Python利用自带模块实现屏幕像素高效操作》这篇文章主要为大家详细介绍了Python如何利用自带模块实现屏幕像素高效操作,文中的示例代码讲解详,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1、获取屏幕放缩比例2、获取屏幕指定坐标处像素颜色3、一个简单的使用案例4、总结1、获取屏幕放缩比例from
阅读更多...
nginx-rtmp-module模块实现视频点播的示例代码

nginx-rtmp-module模块实现视频点播的示例代码

《nginx-rtmp-module模块实现视频点播的示例代码》本文主要介绍了nginx-rtmp-module模块实现视频点播,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习... 目录预置条件Nginx点播基本配置点播远程文件指定多个播放位置参考预置条件配置点播服务器 192.
阅读更多...
C#使用DeepSeek API实现自然语言处理,文本分类和情感分析

C#使用DeepSeek API实现自然语言处理,文本分类和情感分析

《C#使用DeepSeekAPI实现自然语言处理,文本分类和情感分析》在C#中使用DeepSeekAPI可以实现多种功能,例如自然语言处理、文本分类、情感分析等,本文主要为大家介绍了具体实现步骤,... 目录准备工作文本生成文本分类问答系统代码生成翻译功能文本摘要文本校对图像描述生成总结在C#中使用Deep
阅读更多...
Go中sync.Once源码的深度讲解

Go中sync.Once源码的深度讲解

《Go中sync.Once源码的深度讲解》sync.Once是Go语言标准库中的一个同步原语,用于确保某个操作只执行一次,本文将从源码出发为大家详细介绍一下sync.Once的具体使用,x希望对大家有... 目录概念简单示例源码解读总结概念sync.Once是Go语言标准库中的一个同步原语，用于确保某个操
阅读更多...
Redis主从/哨兵机制原理分析

Redis主从/哨兵机制原理分析

《Redis主从/哨兵机制原理分析》本文介绍了Redis的主从复制和哨兵机制,主从复制实现了数据的热备份和负载均衡,而哨兵机制可以监控Redis集群,实现自动故障转移,哨兵机制通过监控、下线、选举和故... 目录一、主从复制1.1 什么是主从复制1.2 主从复制的作用1.3 主从复制原理1.3.1 全量复制
阅读更多...
Redis主从复制的原理分析

Redis主从复制的原理分析

《Redis主从复制的原理分析》Redis主从复制通过将数据镜像到多个从节点,实现高可用性和扩展性,主从复制包括初次全量同步和增量同步两个阶段,为优化复制性能,可以采用AOF持久化、调整复制超时时间、... 目录Redis主从复制的原理主从复制概述配置主从复制数据同步过程复制一致性与延迟故障转移机制监控与维
阅读更多...
多模块的springboot项目发布指定模块的脚本方式

多模块的springboot项目发布指定模块的脚本方式

《多模块的springboot项目发布指定模块的脚本方式》该文章主要介绍了如何在多模块的SpringBoot项目中发布指定模块的脚本,作者原先的脚本会清理并编译所有模块,导致发布时间过长,通过简化脚本... 目录多模块的springboot项目发布指定模块的脚本1、不计成本地全部发布2、指定模块发布总结多模
阅读更多...
Redis连接失败：客户端IP不在白名单中的问题分析与解决方案

Redis连接失败：客户端IP不在白名单中的问题分析与解决方案

《Redis连接失败：客户端IP不在白名单中的问题分析与解决方案》在现代分布式系统中,Redis作为一种高性能的内存数据库,被广泛应用于缓存、消息队列、会话存储等场景,然而,在实际使用过程中,我们可能... 目录一、问题背景二、错误分析1. 错误信息解读2. 根本原因三、解决方案1. 将客户端IP添加到Re
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2