当前位置：首页 > article >正文

从零开始实现一个C++高性能服务器框架----IO协程调度模块

article 2024/12/28 3:58:56

此项目是根据sylar框架实现，是从零开始重写sylar，也是对sylar丰富与完善
项目地址：https://gitee.com/lzhiqiang1999/server-framework

简介

项目介绍：实现了一个基于协程的服务器框架，支持多线程、多协程协同调度；支持以异步处理的方式提高服务器性能；封装了网络相关的模块，包括socket、http、servlet等，支持快速搭建HTTP服务器或WebSokcet服务器。
详细内容：日志模块，使用宏实现流式输出，支持同步日志与异步日志、自定义日志格式、日志级别、多日志分离等功能。线程模块，封装pthread相关方法，封装常用的锁包括（信号量，读写锁，自旋锁等）。IO协程调度模块，基于ucontext_t实现非对称协程模型，以线程池的方式实现多线程，多协程协同调度，同时依赖epoll实现了事件监听机制。定时器模块，使用最小堆管理定时器，配合IO协程调度模块可以完成基于协程的定时任务调度。hook模块，将同步的系统调用封装成异步操作（accept, recv, send等），配合IO协程调度能够极大的提升服务器性能。Http模块，封装了sokcet常用方法，支持http协议解析，客户端实现连接池发送请求，服务器端实现servlet模式处理客户端请求，支持单Reator多线程，多Reator多线程模式的服务器。

IO协程调度模块

IO协程调度模块继承自协程调度模块Scheduler和定时器模块TimeManager，能够处理IO事件与定时任务。
对于IO事件，可以在对应的fd上注册事件和事件回调函数，当事件触发时，会由调度器执行事件回调函数。
对于定时器，可以设置定时时间及定时任务，当时间一到，又调度器执行定时任务。
IO协程调度器使用一对管道fd来tickle()调度协程，当调度器空闲时，idle协程通过epoll_wait阻塞在管道的读描述符上，等管道的可读事件。添加新任务时，tickle()方法写管道，idle协程检测到管道可读后退出，调度器执行调度。

1. 主要功能

继承自Scheduler，重写了tickle和idle方法，解决了线程阻塞，和idle空转的问题
使用epoll系列方法，在协程上实现了IO操作的调度
支持添加，删除IO事件，并完成事件响应
分离了对与fd的监听与IO操作

2. 功能演示

这里模拟了一个客户端请求连接，在cfd上注册可写事件，当连接成功，会触发该事件

   IOManager iom;
   iom.schedule(&test_fiber1);

   int cfd = socket(AF_INET, SOCK_STREAM, 0);
   fcntl(cfd, F_SETFL, O_NONBLOCK);

   sockaddr_in addr;
   memset(&addr, 0, sizeof(sockaddr_in));
   addr.sin_family = AF_INET;
   addr.sin_port = htons(80);
   inet_pton(AF_INET, "180.101.50.188", &addr.sin_addr.s_addr);
	
	//注册可写事件
   iom.addEvent(cfd, johnsonli::IOManager::WRITE, [](){
       LOG_INFO(g_logger) << "connected";
   });
	
	//发起连接
   connect(cfd, (const sockaddr*)&addr, sizeof(addr));

3. 模块介绍

3.1 IOManager

IO协程调度器，主要用于完成对IO操作和定时任务的调度。关于定时器会在下篇文章讲解

class IOManager : public Scheduler {
public:
    typedef std::shared_ptr<IOManager> ptr;
    typedef RWMutex RWMutexType;
...
}

读写时间。直接继承epoll的枚举值

/**
 * @brief IO事件，继承自epoll对事件的定义
 * @details 这里只关心socket fd的读和写事件，其他epoll事件会归类到这两类事件中
 */
enum Event {
    NONE = 0x0,		/// 无事件
    READ = 0x1,	   	/// 读事件(EPOLLIN)
    WRITE = 0x4,	/// 写事件(EPOLLOUT)
};

对于IO协程调度来说，每次调度都包含一个三元组信息，分别是描述符fd–事件类型（可读或可写）–回调函数，调度器记录全部需要调度的三元组信息，其中描述符fd和事件类型用于epoll_wait，回调函数用于协程调度。这里使用struct FdContext表示

//socket fd的上下文类
struct FdContext
{
    typedef Mutex MutexType;

    // 事件上下文类
    struct EventContext
    {
        Scheduler* scheduler = nullptr;     //事件执行的调度器
        Fiber::ptr fiber;                   //事件协程
        std::function<void()> cb;           //事件的回调函数
    };

    //获取指定事件的上下文
    EventContext& getEventContext(Event event);
    //重置事件的上下文
    void resetEventContext(EventContext& ctx);
    //触发指定的事件(添加到任务协程)
    void triggerEvent(Event event);

    EventContext read;      //读事件上下文
    EventContext write;     //写事件上下文

    int fd = 0;             //事件关联的句柄
    Event events = NONE;    //当前注册的事件
    MutexType mutex;        //事件的mutex
};

IOManager的成员变量

int m_epfd;                                     //epoll 文件句柄
int m_tickleFds[2];                             //pipe 文件句柄，用于通知任务
std::atomic<size_t> m_pendingEventCount = {0};  //当前等待执行的事件数量
std::vector<FdContext*> m_fdContexts;           //socket事件上下文的容器
MutexType m_mutex;

IOManager的构造函数。具体需要完成以下4个操作

初始化Scheduler
创建m_epfd
监听m_tickleFds[0]读事件
初始化m_fdContexts
运行调度器
这里需要说明的是为什么需要监听管道读端m_tickleFds[0]的读事件。因为当所有的任务协程都执行完毕，此时会陷入到idle协程的epoll_wait，如果此时有新任务加入，而epoll上注册的事件又还没有触发，此时会一直阻塞在epoll_wait。因此，当有新的协程任务到来时，应当向管道写端m_tickleFds[1]写入数据，此时epoll上m_tickleFds[0]的读事件将被触发，不会陷入epoll_wait，继续调度新的任务协程。

IOManager::IOManager(size_t threads, bool use_caller, const std::string& name)
 :Scheduler(threads, use_caller, name) {
 m_epfd = epoll_create(5000);
 DO_ASSERT(m_epfd > 0);
 
 //给m_tickleFds[0]注册读事件，当加入任务时，可以往m_tickleFds[1]写，保证程序不会被阻塞，从而监听到新加入的任务
 int rt = pipe(m_tickleFds);
 DO_ASSERT(!rt);

 epoll_event event;
 memset(&event, 0, sizeof(epoll_event));
 event.events = EPOLLIN | EPOLLET; //监听读事件，边沿触发（一次触发后，之后不再触发，一般设置为非阻塞轮询）
 event.data.fd = m_tickleFds[0];

 //非阻塞轮询
 rt = fcntl(m_tickleFds[0], F_SETFL, O_NONBLOCK);
 DO_ASSERT(!rt);

 rt = epoll_ctl(m_epfd, EPOLL_CTL_ADD, m_tickleFds[0], &event);
 DO_ASSERT(!rt);

 //初始化m_fdContexts
 contextResize(32);
 
 start();    //初始化好后就开始Scheduler的start
 
}

tickle()函数。有新任务协程时，向管道写端m_tickleFds[1]写入数据，唤醒epoll_wait上m_tickleFds[0]的读事件，保证，继续调度新的任务协程。

void IOManager::tickle() {
    //没有空闲的线程，就不用唤醒
    if(!hasIdleThreads()) return;
    int rt = write(m_tickleFds[1], "T", 1);
    DO_ASSERT(rt == 1);
}

ilde()函数。调度器无调度任务时会阻塞idle协程上，对IO调度器而言，idle状态应该关注两件事，一是有没有新的调度任务，对应Schduler::schedule()，如果有新的调度任务，那应该立即退出idle状态，并执行对应的任务；二是关注当前注册的所有IO事件有没有触发，如果有触发，那么应该执行。

void IOManager::idle() {
 
    // 一次epoll_wait最多检测256个就绪事件，如果就绪事件超过了这个数，那么会在下轮epoll_wati继续处理
    const uint64_t MAX_EVNETS = 256;
    epoll_event *events       = new epoll_event[MAX_EVNETS]();
    std::shared_ptr<epoll_event> shared_events(events, [](epoll_event *ptr) {
        delete[] ptr;
    });
 
    while (true) {
        if(stopping()) {
            LOG_DEBUG(g_logger) << "name=" << getName() << "idle stopping exit";
            break;
        }
        // 阻塞在epoll_wait上，等待事件发生
        static const int MAX_TIMEOUT = 5000;
        int rt = epoll_wait(m_epfd, events, MAX_EVNETS, MAX_TIMEOUT);
        if(rt < 0) {
            if(errno == EINTR) {
                continue;
            }
            LOG_ERROR(g_logger) << "epoll_wait(" << m_epfd << ") (rt="
                                      << rt << ") (errno=" << errno << ") (errstr:" << strerror(errno) << ")";
            break;
        }
 
        // 遍历所有发生的事件，根据epoll_event的私有指针找到对应的FdContext，进行事件处理
        for (int i = 0; i < rt; ++i) {
            epoll_event &event = events[i];
            if (event.data.fd == m_tickleFds[0]) {
                // ticklefd[0]用于通知协程调度，这时只需要把管道里的内容读完即可，本轮idle结束Scheduler::run会重新执行协程调度
                uint8_t dummy[256];
                while (read(m_tickleFds[0], dummy, sizeof(dummy)) > 0;
                continue;
            }
             
            // 通过epoll_event的私有指针获取FdContext
            FdContext *fd_ctx = (FdContext *)event.data.ptr;
            FdContext::MutexType::Lock lock(fd_ctx->mutex);
            /**
             * EPOLLERR: 出错，比如写读端已经关闭的pipe
             * EPOLLHUP: 套接字对端关闭
             * 出现这两种事件，应该同时触发fd的读和写事件，否则有可能出现注册的事件永远执行不到的情况
             */
            if (event.events & (EPOLLERR | EPOLLHUP)) {
                event.events |= (EPOLLIN | EPOLLOUT) & fd_ctx->events;
            }
            int real_events = NONE;
            if (event.events & EPOLLIN) {
                real_events |= READ;
            }
            if (event.events & EPOLLOUT) {
                real_events |= WRITE;
            }
 
            if ((fd_ctx->events & real_events) == NONE) {
                continue;
            }
 
            // 剔除已经发生的事件，将剩下的事件重新加入epoll_wait，
            // 如果剩下的事件为0，表示这个fd已经不需要关注了，直接从epoll中删除
            int left_events = (fd_ctx->events & ~real_events);
            int op          = left_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
            event.events    = EPOLLET | left_events;
 
            int rt2 = epoll_ctl(m_epfd, op, fd_ctx->fd, &event);
            if (rt2) {
                LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
                                          << (EpollCtlOp)op << ", " << fd_ctx->fd << ", " << (EPOLL_EVENTS)event.events << "):"
                                          << rt2 << " (" << errno << ") (" << strerror(errno) << ")";
                continue;
            }
 
            // 处理已经发生的事件，也就是让调度器调度指定的函数或协程
            if (real_events & READ) {
                fd_ctx->triggerEvent(READ);
                --m_pendingEventCount;
            }
            if (real_events & WRITE) {
                fd_ctx->triggerEvent(WRITE);
                --m_pendingEventCount;
            }
        } // end for
 
        /**
         * 一旦处理完所有的事件，idle协程yield，这样可以让调度协程(Scheduler::run)重新检查是否有新任务要调度
         * 上面triggerEvent实际也只是把对应的fiber重新加入调度，要执行的话还要等idle协程退出
         */
        Fiber::ptr cur = Fiber::GetThis();
        auto raw_ptr   = cur.get();
        cur.reset();
 
        raw_ptr->back();
    } // end while(true)
}

注册事件。主要有以下几个步骤

从m_fdContexts中拿到对应的fd: fd_ctx
修改fd_ctx
添加到m_epfd

int IOManager::addEvent(int fd, Event event, std::function<void()> cb) {
    FdContext* fd_ctx = nullptr;
    MutexType::ReadLock lock(m_mutex);
    
    //从m_fdContexts中拿到对应的fd
    if((int)m_fdContexts.size() > fd) {
        fd_ctx = m_fdContexts[fd];  //下标志就是对应的fd
        lock.unlock();
    } else {
        lock.unlock();
        MutexType::WriteLock lock2(m_mutex);
        contextResize(fd * 1.5);
        fd_ctx = m_fdContexts[fd];
    }

    //修改fd
    FdContext::MutexType::Lock lock2(fd_ctx->mutex);
    fd_ctx->fd = fd;
    if(fd_ctx->events & event)  //如果fd_ctx上已经有这个事件了，出错
    {
        LOG_ERROR(g_logger) << "addEvent assert fd=" << fd
        << " event=" << (EPOLL_EVENTS)event
        << " fd_ctx.event=" << (EPOLL_EVENTS)fd_ctx->events;
        DO_ASSERT(!(fd_ctx->events & event));
    }
    Event old_event = fd_ctx->events;
    fd_ctx->events = (Event)(fd_ctx->events | event);
    
    FdContext::EventContext& event_ctx = fd_ctx->getEventContext(event);
    DO_ASSERT(!event_ctx.scheduler
        && !event_ctx.fiber
        && !event_ctx.cb);

    event_ctx.scheduler = Scheduler::GetThis();
    if(cb) {
        event_ctx.cb.swap(cb);
    }else {
        event_ctx.fiber = Fiber::GetThis();
        DO_ASSERT2(event_ctx.fiber->getState() == Fiber::EXEC
                    ,"state=" << event_ctx.fiber->getState());
    }

    //添加到m_epfd
    int op = old_event ? EPOLL_CTL_MOD : EPOLL_CTL_ADD;    //fd_ctx之前有，添加；否则修改
    epoll_event epevent;
    epevent.events = EPOLLET | fd_ctx->events;
    epevent.data.ptr = fd_ctx;

    int rt = epoll_ctl(m_epfd, op, fd, &epevent);
    if(rt) {
        LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
            << op << ", " 
            << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
            << rt << " (" << errno << ") (" << strerror(errno) << ") fd_ctx->events="
            << (EPOLL_EVENTS)fd_ctx->events;
        return -1;
    }

    //添加了一个事件
    ++m_pendingEventCount;

    return 0;
}

删除事件。主要有以下几个步骤

从m_fdContexts中拿到对应的fd: fd_ctx
修改fd_ctx
从m_epfd删除

bool IOManager::delEvent(int fd, Event event) {
    MutexType::ReadLock lock(m_mutex);

    //1. 从m_fdContexts中拿到对应的fd: fd_ctx
    if((int)m_fdContexts.size() <= fd) {
        return false;
    }
    FdContext* fd_ctx = m_fdContexts[fd];
    lock.unlock();

    //2. 修改fd_ctx
    if(!(fd_ctx->events & event)) {     //fd_ctx中没有该事件
        return false;
    }
 
    fd_ctx->events = (Event)(fd_ctx->events & ~event);
    FdContext::EventContext& event_ctx = fd_ctx->getEventContext(event);
    fd_ctx->resetEventContext(event_ctx);

    //3. 从m_epfd删除
    int op = fd_ctx->events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL; //减完event有，修改；没有，删除
    epoll_event epevent;
    epevent.events = EPOLLET | fd_ctx->events;
    epevent.data.ptr = fd_ctx;

    int rt = epoll_ctl(m_epfd, op, fd, &epevent);
    if(rt) {
        LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
            << op << ", " 
            << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
            << rt << " (" << errno << ") (" << strerror(errno) << ")";
        return false;
    }

    //删除了一个事件
    --m_pendingEventCount;
    return true;
}

取消事件并触发。主要有以下几个步骤

从m_fdContexts中拿到对应的fd: fd_ctx
从m_epfd上删除
触发事件

bool IOManager::cancelEvent(int fd, Event event) 
{
    MutexType::ReadLock lock(m_mutex);

    //1. 从m_fdContexts中拿到对应的fd: fd_ctx
    if((int)m_fdContexts.size() <= fd) {
        return false;
    }
    FdContext* fd_ctx = m_fdContexts[fd];
    lock.unlock();

    //2. 从m_epfd上删除
    FdContext::MutexType::Lock lock2(fd_ctx->mutex);
    if(!(fd_ctx->events & event)) {
        return false;
    }
    Event new_events = (Event)(fd_ctx->events & ~event);

    int op = new_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    epoll_event epevent;
    epevent.events = EPOLLET | new_events;
    epevent.data.ptr = fd_ctx;

    int rt = epoll_ctl(m_epfd, op, fd, &epevent);
    if(rt) {
        LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
            << op << ", " << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
            << rt << " (" << errno << ") (" << strerror(errno) << ")";
        return false;
    }

    //3. 触发事件
    fd_ctx->triggerEvent(event);
    --m_pendingEventCount;
    return true;
    
}

取消所有事件，并触发。主要有以下几个步骤

从m_fdContexts中拿到对应的fd: fd_ctx
从m_epfd上删除所有事件
触发所有事件

 bool IOManager::cancelAll(int fd) 
{
    //1. 从m_fdContexts中拿到对应的fd: fd_ctx
    MutexType::ReadLock lock(m_mutex);
    if((int)m_fdContexts.size() <= fd) {
        return false;
    }
    FdContext* fd_ctx = m_fdContexts[fd];
    lock.unlock();

    //2. 从m_epfd上删除所有事件
    FdContext::MutexType::Lock lock2(fd_ctx->mutex);
    if(!fd_ctx->events) {   //没有任何事件
        return false;
    }

    int op = EPOLL_CTL_DEL;
    epoll_event epevent;
    epevent.events = 0;
    epevent.data.ptr = fd_ctx;

    int rt = epoll_ctl(m_epfd, op, fd, &epevent);
    if(rt) {
        LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
            << op << ", " << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
            << rt << " (" << errno << ") (" << strerror(errno) << ")";
        return false;
    }

    //3. 触发事件
    if(fd_ctx->events & READ) {
        fd_ctx->triggerEvent(READ);
        --m_pendingEventCount;
    }
    if(fd_ctx->events & WRITE) {
        fd_ctx->triggerEvent(WRITE);
        --m_pendingEventCount;
    }
    
    //所有事件已经删除完毕
    DO_ASSERT(fd_ctx->events == 0);
    return true;
}

触发事件。实际上，触发事件只是将事件处理加入到当前IO协程调度器的任务协程队列。由调度器统一分配。

void IOManager::FdContext::triggerEvent(Event event) {
    DO_ASSERT(events & event);
    events = (Event)(events & ~event);

    EventContext& ctx = getEventContext(event);
    if(ctx.cb) {
        ctx.scheduler->schedule(&ctx.cb);
    } else {
        ctx.scheduler->schedule(&ctx.fiber);
    }
    ctx.scheduler = nullptr;

    return;
}

IOManage的析构。首先要等Scheduler调度完所有的任务，然后再关闭epoll句柄和pipe句柄，然后释放所有的FdContext。

IOManager::~IOManager() {
    stop();
    close(m_epfd);
    close(m_tickleFds[0]);
    close(m_tickleFds[1]);
 
    for (size_t i = 0; i < m_fdContexts.size(); ++i) {
        if (m_fdContexts[i]) {
            delete m_fdContexts[i];
        }
    }
}

stopping函数。IOManager在判断是否可退出时，还要加上所有IO事件都完成调度的条件。

bool IOManager::stopping() {
    // 对于IOManager而言，必须等所有待调度的IO事件都执行完了才可以退出
    return m_pendingEventCount == 0 && Scheduler::stopping();
}

查看全文

http://www.kler.cn/a/6290.html

Nmap基础入门及常用命令汇总

Java旅程（五）Spring 框架与微服务架构了解 JVM 内部原理和调优

前端Python应用指南（二）深入Flask：理解Flask的应用结构与模块化设计

QT从入门到精通（三）——实现文件列表遍历的图像浏览器

linux命令中cp命令-rf与-a的差别

企业数字化转型中的“烟囱效应”：从小烟囱到大烟囱的折中之道

word打latex公式显示不成功，出现【打不出左大括号

Springboot高级（一）缓存

（七）Tomcat源码阅读：Host组件分析

熬夜肝完~ 阿里P8的Java进阶知识典藏版，我从18K飙到30K

【js】超详细js函数基础

React createContext 与 useContext 的基本使用

【Mysql】事务原理

MySQL安装部署02-VirtualBox虚拟机上Centos6.8安装MySQL5.1.73

sql注入靶场练习

（五）MyBatis源码阅读： MyBatis基础模块-类型转换模块

Flink (六) --------- Flink 中的时间和窗口函数

系统上线前，SQL脚本的9大坑

【Java项目】SpringBoot实现一个请求同时上传多个文件和类并附上代码实例

SIP终端常见的功能及协议支持

RocketMQ通信协议

IDEA在console中编写sql语句报红

操作系统作业1

MongoDB - 索引知识

众人围剿，GPT-5招惹了谁

深入学习Redis：持久化

简介

IO协程调度模块

1. 主要功能

2. 功能演示

3. 模块介绍

3.1 IOManager

相关文章：