c++协程库libfiber之5：协程池

素颜马尾好姑娘i 2022-12-31 02:25 194阅读 0赞

## 目的 ##

我想在ibfiber的基础上实现一个简单的协程池类。我希望它是单例模式的，并且是线程安全的。  
执行的任务由std::function加入到任务队列里。

## 协程池类的实现 ##

可以看到，下面的协程池类主要是三个接口：getInstance()、AddTask()和Clear()

//协成池，负责管理和调度协程
    class FiberPool {
        //使用单例模式来实现
        private:
        static acl::fiber_event FiberLock;
        static FiberPool *local_instance;
        FiberPool(){};
        FiberPool(const FiberPool&);
        FiberPool& operator=(const FiberPool&);
    
        public:
    
        //单例模式，获取指针
        static FiberPool *getInstance()
        {
            FiberLock.wait();
            if (local_instance == nullptr)
            {
                local_instance = new FiberPool();
                //启动一个线程，创建协程池开始调度
                std::thread t(FiberThread);
                t.detach();
            }
            FiberLock.notify();
            return local_instance;
        }
    
        //返回：-1-失败，大于等于零, 成功
        int AddTask(const TaskFuncType &Func) {
            TaskFuncType *ptr = new TaskFuncType(Func);
            m_TaskTBox.push(ptr);
            return 0;
        }
    
        //清空任务队列
        int Clear() {
            m_TaskTBox.clear(true);
            return 0;
        }
    
        private:
        //创建调度线程
        static void FiberThread() {
            std::cout << "FiberThread, started!" << std::endl;
            local_instance->CreateFibers(3);//协程池里有三个协程
            std::cout << "after create fibers!" << std::endl;
            acl::fiber::schedule_with(acl::FIBER_EVENT_T_KERNEL); //调度协程
            std::cout << "FiberThread, exit!" << std::endl;
        }
    
        //用于存放入库的任务函数
        TaskBoxType m_TaskTBox;
    
        //用于存放协程
        std::vector<std::shared_ptr<TaskFiber>> FiberVector;
        void CreateFibers(unsigned int Num) {
            std::cout << "create [" << Num << "]" << " fibers for pool>>" << std::endl;
            for (unsigned int i = 0; i < Num; i++) {
                auto task = std::make_shared<TaskFiber>(&m_TaskTBox);
                task->start();
                FiberVector.push_back(task);
            }
            return;
        }
    
    };

其中getInstance()用于获取单例的指针。为了能够线程安全，使用静态的acl::fiber\_event进行互斥。在第一次调用getInstance()时，除了申请内存，还会创建单独的协程池调度线程。协程池线程的函数FiberThread()也是静态的，创建的协程数可以改变。

协程池的任务使用std::function封装，这样接口的调用方完全可以用std::bind来把参数和函数绑定好传进来。任务队列用acl::fiber\_tbox<>,这个类型类似一个动态数组，但是可以确保线程数据的安全，不需要额外的同步方式。当然也可以用alc::channel<>来作为任务队列，只是alc::channel<>默认容量是100，不能动态递增。

//任务类型
    using TaskFuncType = std::function<int (void)>;
    //任务队列类型
    using TaskBoxType = acl::fiber_tbox<TaskFuncType>;

AddTask()和Clear()分别用于提交任务和清空任务。

注意，上述所有的接口都是线程安全的，也是“协程-线程”安全的，这都得益于libfiber精巧的互斥变量设计。

## 协程类的实现 ##

协程类的实现很简单，继承acl::fiber，实现run()函数即可。在run()函数里，只要不停的从任务队列读任务即可。

//协程类，作用就是从tbox获取任务然后执行
    class TaskFiber : public acl::fiber {
        public:
        //实例化的时候必须传入TaskBoxType的指针
        TaskFiber(TaskBoxType *PtrTaskBox):m_PtrTaskBox(PtrTaskBox){};
    
        private:
        //保存指向任务队列的指针
        TaskBoxType *m_PtrTaskBox = NULL;
        
        protected:
    	// @override
    	void run(void) {
            
            std::cout << "fiber-id=[" << get_id() << "], started!" << std::endl;
            //循环检测是否有任务
            while(!self_killed()) {
                //获取任务，超时时间1000ms
                TaskFuncType* PtrFunc =  m_PtrTaskBox->pop(1000);
                if (NULL != PtrFunc) {
                    std::cout << "fiber-id=[" << get_id() << "]" << ", get a task>>" << std::endl;
                    (*PtrFunc)();
                    delete PtrFunc;
                    std::cout << "fiber-id=[" << get_id() << "]" << ", task done<<" << std::endl;
                    yield(); //把本协程调度出去
                }
            }
    
            std::cout << "fiber-id=[" << get_id() << "]" << ", fiber EXIT>>" << std::endl;
        }
    
    };

因为我是用acl::fiber\_tbox<>做任务队列，这个结构只能存指针，所以加入任务时申请内存，执行完要释放内存，这里可以加一个缓存池，避免频繁的申请释放内存。或者使用固定容量的alc::channel<>来作为任务队列，是alc::channel<>保存的是拷贝而不是指针，但是默认容量是100，不能动态递增。

## 调用协程池 ##

为一个函数绑定两个不同的参数作为任务提交给协程池，等待协程执行完毕。

#include <iostream>
    #include <functional>
    #include <thread>
    #include <chrono>
    
    #include "fiberPool.h"
    
    int do_nothing(int num)
    {
        std::cout<< "I'm worker [" << num << "]" << std::endl;
        return 0;
    }
    
    
    int main()
    {
        FiberPool* ptrPool = FiberPool::getInstance();
    
        //为函数绑定两个不同的参数进行测试
        TaskFuncType func1 = std::bind(do_nothing, 1);
        TaskFuncType func2 = std::bind(do_nothing, 2);
    
        std::cout << "add 2 task now!" << std::endl;
        ptrPool->AddTask(func1);
        ptrPool->AddTask(func2);
    
        using namespace std::chrono_literals;
        std::this_thread::sleep_for(5s);
    }

输出内容是：

$ ./fiberPool
    FiberThread, started!add 2 task now!
    create [3] fibers for pool>>
    
    after create fibers!
    fiber-id=[1], started!
    fiber-id=[1], get a task>>
    I'm worker [1]
    fiber-id=[1], task done<<
    fiber-id=[2], started!
    fiber-id=[2], get a task>>
    I'm worker [2]
    fiber-id=[2], task done<<
    fiber-id=[3], started!

## 完整的代码 ##

总共一个头文件，一个CPP文件就可以编译成一个协程池的动态库

#### 协程池头文件 fiberPool.h ####

#include <functional>
    #include <memory>
    #include <vector>
    #include <iostream>
    #include <thread>
    #include <mutex>
    
    #include "fiber/libfiber.hpp"
    
    //任务类型
    using TaskFuncType = std::function<int (void)>;
    //任务队列类型
    using TaskBoxType = acl::fiber_tbox<TaskFuncType>;
    
    
    //协程类，作用就是从tbox获取任务然后执行
    class TaskFiber : public acl::fiber {
        public:
        //实例化的时候必须传入TaskBoxType的指针
        TaskFiber(TaskBoxType *PtrTaskBox):m_PtrTaskBox(PtrTaskBox){};
    
        private:
        //保存指向任务队列的指针
        TaskBoxType *m_PtrTaskBox = NULL;
        
        protected:
    	// @override
    	void run(void) {
            
            std::cout << "fiber-id=[" << get_id() << "], started!" << std::endl;
            //循环检测是否有任务
            while(!self_killed()) {
                //获取任务，超时时间1000ms
                TaskFuncType* PtrFunc =  m_PtrTaskBox->pop(1000);
                if (NULL != PtrFunc) {
                    std::cout << "fiber-id=[" << get_id() << "]" << ", get a task>>" << std::endl;
                    (*PtrFunc)();
                    delete PtrFunc;
                    std::cout << "fiber-id=[" << get_id() << "]" << ", task done<<" << std::endl;
                    yield(); //把本协程调度出去
                }
            }
    
            std::cout << "fiber-id=[" << get_id() << "]" << ", fiber EXIT>>" << std::endl;
        }
    
    };
    
    
    //协成池，负责管理和调度协程
    class FiberPool {
        //使用单例模式来实现
        private:
        static acl::fiber_event FiberLock;
        static FiberPool *local_instance;
        FiberPool(){};
        FiberPool(const FiberPool&);
        FiberPool& operator=(const FiberPool&);
    
        public:
    
        static FiberPool *getInstance()
        {
            FiberLock.wait();
            if (local_instance == nullptr)
            {
                local_instance = new FiberPool();
                //启动一个线程，创建协程池开始调度
                std::thread t(FiberThread);
                t.detach();
            }
            FiberLock.notify();
            return local_instance;
        }
    
        //返回：-1-失败，大于等于零, 成功
        int AddTask(const TaskFuncType &Func) {
            TaskFuncType *ptr = new TaskFuncType(Func);
            m_TaskTBox.push(ptr);
            return 0;
        }
    
        //清空任务队列
        int Clear() {
            m_TaskTBox.clear(true);
            return 0;
        }
    
        //目前协会池设计不会终止
    
        private:
        //创建调度线程
        static void FiberThread() {
            std::cout << "FiberThread, started!" << std::endl;
            local_instance->CreateFibers(3);//协程池里有三个协程
            std::cout << "after create fibers!" << std::endl;
            acl::fiber::schedule_with(acl::FIBER_EVENT_T_KERNEL); //调度协程
            std::cout << "FiberThread, exit!" << std::endl;
        }
    
        //用于存放入库的任务函数
        TaskBoxType m_TaskTBox;
    
        //用于存放协程
        std::vector<std::shared_ptr<TaskFiber>> FiberVector;
        void CreateFibers(unsigned int Num) {
            std::cout << "create [" << Num << "]" << " fibers for pool>>" << std::endl;
            for (unsigned int i = 0; i < Num; i++) {
                auto task = std::make_shared<TaskFiber>(&m_TaskTBox);
                task->start();
                FiberVector.push_back(task);
            }
            return;
        }
    
    };

#### 协程池源文件 fiberPool.cpp ####

大部分逻辑都写在头文件里了，源文件只要定义静态变量即可。

#include "fiberPool.h"
    
    acl::fiber_event FiberPool::FiberLock;
    FiberPool *FiberPool::local_instance = nullptr;

#### 测试源码 main.cpp ####

## 参考资料 ##

[c++协程库libfiber之4：定时和延时][c_libfiber_4]

[c++协程库libfiber之3：调度协程的三种方式][c_libfiber_3]

[c++协程库libfiber之2：编译及例子][c_libfiber_2]

[c++协程库libfiber之1：简单介绍][c_libfiber_1]

[libfiber的源码工程主页][libfiber]

libfiber的贡献者爱奇艺的技术分享：[爱奇艺网络协程编写高并发应用实践][Link 1]

ACL作者介绍libfiber的博客：[acl开发–协程篇][acl]

[c_libfiber_4]: https://blog.csdn.net/yuanlulu/article/details/109107775
[c_libfiber_3]: https://blog.csdn.net/yuanlulu/article/details/109063477
[c_libfiber_2]: https://blog.csdn.net/yuanlulu/article/details/109019489
[c_libfiber_1]: https://blog.csdn.net/yuanlulu/article/details/108970474
[libfiber]: https://github.com/iqiyi/libfiber
[Link 1]: https://www.infoq.cn/article/BOu7GVc9kP7CWkYP58jo
[acl]: https://blog.csdn.net/zsxxsz/category_8736935.html