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

news2024/11/27 0:22:47

首发原文链接：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 errCode
    set_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
#endif
        if (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 SSL
    if (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_OPENSSL
    want_event = SW_EVENT_NULL;
#endif
    swoole_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_OPENSSL
    if (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_callback
            if ((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_callback
            if ((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_callback
            if ((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 中的运行逻辑，打开我们的思路，便于我们更好的体会到协程所带来的高性能价值。

在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1565187.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【软件工程】详细设计（二）

【软件工程】详细设计（二）

这里是详细设计文档的第二部分。前一部分点这里 4. 学生端模块详细设计 学生端模块主要由几个组件构成&#xff1a;学生登录界面&#xff0c;成绩查询界面等界面。因为学生端的功能相对来说比较单一&#xff0c;因此这里只给出两个最重要的功能。 图4.1 学生端模块流程图 4.…
阅读更多...
数据库加载驱动问题（java.lang.ClassNotFoundException: com.mysql.cj.jdbc.Driver）

数据库加载驱动问题（java.lang.ClassNotFoundException: com.mysql.cj.jdbc.Driver）

java.lang.ClassNotFoundException: com.mysql.cj.jdbc.Driver 遇到此问题&#xff0c;首先检查IDEA外部库中是否有mysql数据库驱动。如下所示&#xff1a; 如果发现外部库中存有mysql数据库驱动&#xff0c;需要在数据库配置文件中查看是否设置有时区mysql8.0以上版本需要设…
阅读更多...
多态--下

多态--下

文章目录 概念多态如何实现的指向谁调谁&#xff1f;例子分析 含有虚函数类的大小是多少&#xff1f;虚函数地址虚表地址多继承的子类的大小怎么计算&#xff1f;练习题虚函数和虚继承 概念 优先使用组合、而不是继承; 继承会破坏父类的封装、因为子类也可以调用到父类的函数;…
阅读更多...
Springboot整合Milvus向量库

Springboot整合Milvus向量库

1. Milvus的Maven依赖&#xff0c; 配置如下 <dependency><groupId>io.milvus</groupId><artifactId>milvus-sdk-java</artifactId><version>2.3.4</version><exclusions><exclusion><artifactId>log4j-slf4j-imp…
阅读更多...
【蓝桥备赛】异或和——树状数组、DFS

【蓝桥备赛】异或和——树状数组、DFS

题目链接 异或和 思路分析 树上每个点都有一个点权&#xff0c;对树上的更新操作是修改指定点的点权&#xff0c;查询操作是查询指定点为根结点的子树点权异或和。 这里的这些操作都和树状数组的单点修改和区间查询非常相似&#xff0c;即我们在修改一个点时&#xff0c;同时…
阅读更多...
浅谈物联网高速公路智慧配电室系统构建方案

浅谈物联网高速公路智慧配电室系统构建方案

关键词&#xff1a;高速公路&#xff1b;智慧供配电&#xff1b;电力监控&#xff1b;配电室智能运维托管&#xff1b;安全隐患 0、引言 随着高速公路事业的不断发展和路网的不断延伸&#xff0c;传统的管理方式已难以满足日益增长的需求&#xff0c;动态管理和安全隐患预警成…
阅读更多...
当面试官问你插入排序算法，你敢说自己会吗？

当面试官问你插入排序算法，你敢说自己会吗？

算法学习的重要性 在程序员的世界里&#xff0c;算法就如同一座桥梁&#xff0c;连接着问题与解决方案&#xff0c;是实现优秀程序的关键。 掌握算法&#xff0c;就能够在面对各种问题时&#xff0c;找到最合适的解决方法&#xff0c;以最少的时间和空间&#xff0c;实现最优的…
阅读更多...
【airtest】自动化入门教程（四）Poco元素定位

【airtest】自动化入门教程（四）Poco元素定位

目录 一、基础操作 1、通过属性名等方式 2、通过属性组合 3、子节点方式 4、子节点加属性组合方式 5、孙节点offspring 6、兄弟节点sibling 7、父节点parent 8、正则表达式 9、直到某个元素出现 10、直到某个元素消失 二、通过局部坐标定位 1、使用局部坐标系的cli…
阅读更多...
计算多个元素的累乘结果累乘器start默认初始为1 math.prod()

计算多个元素的累乘结果累乘器start默认初始为1 math.prod()

【小白从小学Python、C、Java】 【计算机等考500强证书考研】 【Python-数据分析】 计算多个元素的累乘结果 累乘器start默认初始为1 math.prod() [太阳]选择题 请问题目中的代码最后输出什么? import math list1 [1, 2, 3] print("【显示】list1 ",list1) pri…
阅读更多...
留学生在美国大学利用AI工具到底算不算作弊呢？

留学生在美国大学利用AI工具到底算不算作弊呢？

自2022年以来&#xff0c;美国大学就开启了一场AI作弊与反作弊大战 战场小至测验&#xff0c;大至申请 这场战争并没有一方胜利&#xff0c;作弊者心思费尽 校方反作弊弄得教授们苦不堪言 那么作为中国留学生该如何避免这场战役呢&#xff1f; 毕竟还是学业要紧呢…… 故事…
阅读更多...
[Leetcode笔记] 动态规划相关

[Leetcode笔记] 动态规划相关

前言 写题目写到了一些和动态规划相关的内容&#xff0c;所以在这里记录一下 LCR 089 题解思路 总的来说&#xff0c;就是用一个数组去存储当前的最优解&#xff0c;然后从0开始一路向上顺推过去&#xff0c;求得最后一位的最优解。 class Solution { public:int rob(vect…
阅读更多...
【HTML】标签学习（下.2）

【HTML】标签学习（下.2）

&#xff08;大家好哇&#xff0c;今天我们将继续来学习HTML&#xff08;下.2&#xff09;的相关知识&#xff0c;大家可以在评论区进行互动答疑哦~加油&#xff01;&#x1f495;&#xff09; 目录 二.列表标签 2.1 无序列表(重点) 2.2有序列表(理解) 2.3 自定义列表(重点…
阅读更多...
数据结构算法系列----广度优先搜索（bfs）

数据结构算法系列----广度优先搜索（bfs）

目录 一、什么是bfs 二、bfs和dfs的差异 搜索顺序&#xff1a; 数据结构&#xff1a; 搜索方式&#xff1a; 三、bfs解决的主要问题 四、例题 一、什么是bfs BFS&#xff08;广度优先搜索&#xff09;是一种图搜索算法&#xff0c;用于在图或树数据结构中进行遍历。BFS从…
阅读更多...
管理科学笔记

管理科学笔记

1.线性规划 画出区域&#xff0c;代入点计算最大最小值 2.最小生成树 a.断线法&#xff0c;从大的开始断 b.选择法&#xff0c;从小的开始选 3.匈牙利法 维度数量直线覆盖所有的0 4.一直选最当前路线最短路径 5.线性规划 6.决策论
阅读更多...
反射——获取Class对象的三种方法，构造器、成员变量、方法

反射——获取Class对象的三种方法，构造器、成员变量、方法

作用 学习如何获取类的信息&#xff0c;操作它们 获取class对象的三种方法 package com.zz.reflection;import com.zz.Interface.studentMannger.ClassManager;//目标&#xff1a;获取class对象 public class Test1Class {public static void main(String args[]) throws Class…
阅读更多...
类方法介绍、使用细节

类方法介绍、使用细节

阅读更多...
【Qt】使用Qt实现Web服务器（九）：EventSource+JSON实现工业界面数据刷新

【Qt】使用Qt实现Web服务器（九）：EventSource+JSON实现工业界面数据刷新

1、效果 效果如下，实时刷新温度、湿度 2、源码 2.1 index.html <html><body> <!-- 页面布局，本人对HTML标签不熟悉，凑合看吧 --> <div><label for
阅读更多...
微信怎么恢复好友？7个方法助你轻松寻回失联好友

微信怎么恢复好友？7个方法助你轻松寻回失联好友

在数字化社交日益盛行的今天&#xff0c;微信作为我们日常生活中不可或缺的沟通工具&#xff0c;承载着与亲朋好友、同事伙伴之间的深厚情谊。然而&#xff0c;有时由于误操作或其他原因&#xff0c;我们可能会不小心删除了某些重要的微信好友&#xff0c;这时&#xff0c;如何…
阅读更多...
【二叉树】Leetcode 105. 从前序与中序遍历序列构造二叉树【中等】

【二叉树】Leetcode 105. 从前序与中序遍历序列构造二叉树【中等】

从前序与中序遍历序列构造二叉树 给定两个整数数组 preorder 和 inorder &#xff0c;其中 preorder 是二叉树的先序遍历&#xff0c; inorder 是同一棵树的中序遍历&#xff0c;请构造二叉树并返回其根节点。 示例1&#xff1a; 输入: preorder [3,9,20,15,7], inorder …
阅读更多...
刷题之动态规划-子数组

刷题之动态规划-子数组

前言 大家好&#xff0c;我是jiantaoyab&#xff0c;开始刷动态规划的子数组类型相关的题目 动态规划5个步骤 状态表示 &#xff1a;dp数组中每一个下标对应值的含义是什么>dp[i]表示什么状态转移方程&#xff1a; dp[i] 等于什么1 和 2 是动态规划的核心步骤&#xff0c;…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023