自学内容网 自学内容网
自学内容网 > 正文

【C++20】从0开始自制协程库

🕗 发布于 2024-07-25 10:25 c++20  

文章目录

很多人对协程的理解就是在用户态线程把CPU对线程的调度复制了一遍，减少了线程的数量，也就是说在一个线程内完成对协程的调度，不需要线程切换导致上下文切换的开销。但是线程切换是CPU行为，就算你的程序只有一个线程，多个应用一样会切换上下文，所以并不能提升效率。
所以协程必须引入线程池以及epoll的多路复用才能在大部分情况下提升每个线程的运行效率，防止线程空转。

协程本质上是一种用户态的轻量级线程，它允许在一个线程内切换执行多个任务，而不需要进入内核态。尽管协程的设计可以显著减少上下文切换的开销和线程调度的复杂性，但它们并不是万能的。要在大多数情况下提升每个线程的运行效率，并防止线程空转，确实需要引入一些机制，如线程池和 epoll 的多路复用。

为什么需要线程池和 epoll
线程池：

目的：线程池通过预先创建和管理一组线程，减少了线程创建和销毁的开销。线程池中的线程可以重复使用，从而提高系统的吞吐量。
协程与线程池的结合：协程在每个线程上运行，可以充分利用线程池中的每个线程，避免每次任务执行都创建和销毁线程的高成本操作。
防止空转：当没有任务时，线程可以阻塞等待新的任务，从而避免空转。
epoll 多路复用：

目的：epoll 提供了高效的 I/O 事件通知机制，可以监视多个文件描述符，只有当文件描述符准备好进行 I/O 操作时才会通知应用程序，从而避免了轮询所有文件描述符的低效操作。
协程与 epoll 的结合：协程可以与 epoll 结合使用，通过 epoll 进行 I/O 多路复用，协程可以在等待 I/O 操作时挂起，直到 I/O 操作完成后再恢复执行，从而实现非阻塞 I/O 和高效的并发处理。
示例：使用协程、线程池和 epoll
以下是一个简单的示例，展示如何结合使用协程、线程池和 epoll 来实现高效的并发处理。

cpp
复制代码
#include
#include
#include
#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>
#include

// 简单的协程任务类
class CoroutineTask {
public:
struct promise_type {
CoroutineTask get_return_object() { return {}; }
std::suspend_never initial_suspend() { return {}; }
std::suspend_never final_suspend() noexcept { return {}; }
void return_void() {}
void unhandled_exception() { std::terminate(); }
};

CoroutineTask() = default;

};

// 线程池类
class ThreadPool {
public:
ThreadPool(size_t num_threads) {
for (size_t i = 0; i < num_threads; ++i) {
workers.emplace_back([this] {
while (true) {
std::function<void()> task;
{
std::unique_lockstd::mutex lock(this->queue_mutex);
this->condition.wait(lock, [this] { return this->stop || !this->tasks.empty(); });
if (this->stop && this->tasks.empty()) return;
task = std::move(this->tasks.front());
this->tasks.pop();
}
task();
}
});
}
}

template <class F>
void enqueue(F&& f) {
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        tasks.emplace(std::forward<F>(f));
    }
    condition.notify_one();
}

~ThreadPool() {
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
    }
    condition.notify_all();
    for (std::thread& worker : workers) {
        worker.join();
    }
}

private:
std::vectorstd::thread workers;
std::queue<std::function<void()>> tasks;
std::mutex queue_mutex;
std::condition_variable condition;
bool stop = false;
};

// 设置文件描述符为非阻塞模式
void set_non_blocking(int fd) {
int flags = fcntl(fd, F_GETFL, 0);
fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}

// 协程任务：处理 I/O 事件
CoroutineTask handle_io(int epoll_fd, int fd) {
epoll_event ev;
ev.events = EPOLLIN | EPOLLET;
ev.data.fd = fd;
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &ev);

while (true) {
    co_await std::suspend_always{}; // 挂起协程，等待 I/O 事件

    char buf[1024];
    ssize_t n = read(fd, buf, sizeof(buf));
    if (n > 0) {
        std::cout << "Read " << n << " bytes from fd " << fd << std::endl;
    } else if (n == 0) {
        std::cout << "EOF on fd " << fd << std::endl;
        close(fd);
        break;
    } else if (errno != EAGAIN && errno != EWOULDBLOCK) {
        std::cerr << "Error reading from fd " << fd << std::endl;
        close(fd);
        break;
    }
}

}

int main() {
int epoll_fd = epoll_create1(0);
if (epoll_fd == -1) {
perror(“epoll_create1”);
return 1;
}

ThreadPool pool(4);

int listen_fd = /* ... 初始化监听 socket ... */;
set_non_blocking(listen_fd);

epoll_event ev;
ev.events = EPOLLIN | EPOLLET;
ev.data.fd = listen_fd;
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ev);

std::vector<std::thread> threads;
for (int i = 0; i < 4; ++i) {
    threads.emplace_back([&] {
        while (true) {
            epoll_event events[10];
            int nfds = epoll_wait(epoll_fd, events, 10, -1);
            if (nfds == -1) {
                perror("epoll_wait");
                return;
            }
            for (int n = 0; n < nfds; ++n) {
                if (events[n].data.fd == listen_fd) {
                    int conn_fd = /* ... 接受新的连接 ... */;
                    set_non_blocking(conn_fd);
                    pool.enqueue([epoll_fd, conn_fd] {
                        handle_io(epoll_fd, conn_fd);
                    });
                } else {
                    pool.enqueue([fd = events[n].data.fd] {
                        handle_io(epoll_fd, fd);
                    });
                }
            }
        }
    });
}

for (auto& t : threads) {
    t.join();
}

close(epoll_fd);
return 0;

}
代码解释
协程任务类：定义了一个简单的协程任务类 CoroutineTask。
线程池类：定义了一个线程池 ThreadPool，用来管理工作线程。
非阻塞设置函数：set_non_blocking 函数用于将文件描述符设置为非阻塞模式。
协程任务 handle_io：处理 I/O 事件的协程任务，使用 co_await std::suspend_always{} 挂起协程，等待 I/O 事件。
主函数：
创建 epoll 文件描述符。
初始化线程池。
设置监听 socket 为非阻塞模式，并将其添加到 epoll 实例中。
创建多个线程，每个线程循环等待 epoll 事件，并将事件处理任务加入线程池中。
总结
通过结合使用协程、线程池和 epoll，可以显著提高单个线程的运行效率，并防止线程空转。协程在用户态实现轻量级任务切换，线程池管理工作线程，epoll 提供高效的 I/O 事件通知机制，三者结合可以构建高效的并发系统。

参考


原文地址：https://blog.csdn.net/u011436427/article/details/140653836

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

相关文章

自学内容网
Copyright © 2015-2024