源码请见: https://github.com/oneapi-src/oneTBB/blob/master/src/tbb/co_context.h
在windows系统,TBB(也就是intel 的 oneTBB库),通过windwos fiber(纤程)来实现协程(coroutine)。
创建一个协程,代码很简洁:
inline void create_coroutine(coroutine_type& c, std::size_t stack_size, void* arg) {
__TBB_ASSERT(arg, nullptr);
c = CreateFiber(stack_size, co_local_wait_for_all, arg);
__TBB_ASSERT(c, nullptr);
}
windows系统中线程(thread)与纤程(fiber)调度示意图,如下图所示
windwos中的纤程和通常说的协程类似,处在用户模式下,和内核态无关不会有切换复杂上下文的开销。一个线程一次只能执行一个纤程,实际单个线程上的上多纤程利用时间片形成并发机制。进程与线程是内核态相关的操作机制,调度过程是抢占式的。而协程调度是在用户态完成的,需要代码里显式地将CPU调度交给其他协程,这是协作式的。
在不考量多核心cpu算力扩展的情况下,只谈调度效率或者一个程序对单个cpu的利用率,协程要高效得多。而且,本质上来说,一个线程里面的协程是没有并行机制里面的数据竞争的,这意味着保证同步正确性(没有数据竞争问题)的同时具有了异步灵活性。
将 CPU 的执行从一个线程切换到另一个线程,不可避免地涉及内核调度机制,这是个昂贵的开销操作,如果两个线程经常频繁地来回切换则代价尤其大。 Windows 实现了两种机制来降低这一开销:纤程(fiber)和用户模式调度(UMS , user-mode scheduling)。
纤程使得一个应用程序可以调度它自己的“线程”的执行过程,而不必依赖于 Windows 内置的基于优先级的调度机制。纤程也常被称为“轻量”线程:从调度的角度来看,它们对于内核是不可见的,因为它们是在用户模式下在 Kemel32.dll 中实现的。为了使用纤程,首先要调用 Windows 的 ConvertThreadToFiber 函数。该函数将当前线程转变成一个正在运行的纤程。之后,在转变得到的纤程中,通过调用 CreateFiber 函数,又可以创建额外的纤程(每个纤程可以有它自己的一组纤程)。然而,与线程不同的是,纤程不会自动执行,它必须由 SwitchToFiber 函数手工选中,然后才能执行。新的纤程会一直运行,直到退出,或者调用SwitchToFiber再次选择运行另一个纤程。
感谢: https://www.cnblogs.com/5iedu/p/4830983.html
在linux系统下使用glibc中的ucontext库实现,比起基于windows纤程的协程实现,这个实现要复杂一些。
inline void create_coroutine(coroutine_type& c, std::size_t stack_size, void* arg) {
const std::size_t REG_PAGE_SIZE = governor::default_page_size();
const std::size_t page_aligned_stack_size = (stack_size + (REG_PAGE_SIZE - 1)) & ~(REG_PAGE_SIZE - 1);
const std::size_t protected_stack_size = page_aligned_stack_size + 2 * REG_PAGE_SIZE;
// Allocate the stack with protection property
std::uintptr_t stack_ptr = (std::uintptr_t)mmap(nullptr, protected_stack_size, PROT_NONE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_STACK, -1, 0);
__TBB_ASSERT((void*)stack_ptr != MAP_FAILED, nullptr);
// Allow read write on our stack (guarded pages are still protected)
int err = mprotect((void*)(stack_ptr + REG_PAGE_SIZE), page_aligned_stack_size, PROT_READ | PROT_WRITE);
__TBB_ASSERT_EX(!err, nullptr);
// Remember the stack state
c.my_stack = (void*)(stack_ptr + REG_PAGE_SIZE);
c.my_stack_size = page_aligned_stack_size;
err = getcontext(&c.my_context);
__TBB_ASSERT_EX(!err, nullptr);
c.my_context.uc_link = nullptr;
// cast to char* to disable FreeBSD clang-3.4.1 'incompatible type' error
c.my_context.uc_stack.ss_sp = (char*)c.my_stack;
c.my_context.uc_stack.ss_size = c.my_stack_size;
c.my_context.uc_stack.ss_flags = 0;
typedef void(*coroutine_func_t)();
std::uintptr_t addr = std::uintptr_t(arg);
unsigned lo = unsigned(addr);
unsigned hi = unsigned(std::uint64_t(addr) >> 32);
__TBB_ASSERT(sizeof(addr) == 8 || hi == 0, nullptr);
makecontext(&c.my_context, (coroutine_func_t)co_local_wait_for_all, 2, hi, lo);
}