协程(Coroutine)是目前比较流行的一种并发编程模型,在主流的编程语言里都能找到协程的实现,比如libtask(C)、Boost.Coroutine(C++)、gevent(Python)等等。Lua、Go等语言还在运行时里提供了对协程的支持。Windows的纤程(Fiber),其实也是协程。
协程的“协”,是指“协作式任务管理(Cooperative Task Management)”,这也是协程和线程的主要区别。现代操作系统通常是采用抢占式的任务管理机制,即一个执行中的线程,可以被其他线程抢占。操作系统的主要职责是对硬件资源的虚拟化和抽象化,抢占式的任务管理保证了调度的公平性,让所有线程使用CPU时机会均等。在多任务场景下,抢占式调度可以保证每个任务都能得到及时的响应。比如在用播放器听音乐的时候,不会因为同时打开了一个浏览器浏览网页而产生卡顿。抢占式调度的问题是会造成频繁的上下文切换,导致CPU的计算能力不能被充分利用。对于后台服务器来讲,一台机器通常只专注很少几个任务,线程之间频繁的上下文切换会造成极大的性能浪费。所谓协作式任务管理,就是只有在当前任务主动放弃CPU(通常是为了等待I/O)的情况下,才会切换到其他任务执行。这样既不会因为某个任务阻塞而导致CPU闲置,也不会因为频繁的上下文切换而浪费资源。因此协程在注重性能的服务器开发中是优于线程的。关于协程的本质,在《Cooperative Task Management without Manual Stack Management》这篇论文里讲得很透彻,有兴趣的可以读一下。
举个协程的例子,下面是一段计算斐波那契数列的Python代码:
def fib(max):
f0, f1 = 0, 1
while f1 <= max:
yield f1
f0, f1 = f1, f0 + f1
def main():
g = fib(10)
while True:
try:
f = g.next()
print f
except:
break
if __name__ == "__main__":
main()
类似fib()这种函数在Python里面有一个专门的名称,叫做“生成器(generator)”。不过我们换个角度来看,函数fib()和main()其实就是一对协程,fib()每算出一个斐波那契数就暂停任务,把CPU让给main(),后者打印完数字后,再调用g.next()让fib()继续执行。
下面进入正题:用C++实现一个简单的协程demo。一些实现细节参考了微信开源的PhxRPC,有兴趣的可以看一下。Demo的完整代码可以在这里找到。
这个demo包含两个类:Coroutine和CoroutineRuntime。Coroutine即协程类,声明如下:
class Coroutine {
public:
Coroutine()
: status_(kTaskPending), runtime_(NULL), entry_(NULL), arg_(NULL){};
~Coroutine() { FREE_STACK(stack_); };
int Init(CoroutineRuntime *rt, size_t stack_size, CoroutineEntryType fn,
void *arg);
void Yield();
void Resume();
void Done();
bool IsDone() { return (status_ == kTaskDone); };
void Run() {
if (entry_) {
entry_(runtime_, arg_);
}
};
ucontext_t *GetMainContext();
private:
static void Runner(uint32_t low, uint32_t high);
private:
// 上下文
ucontext_t context_;
Stack stack_;
// 任务状态
int status_;
// 协程入口函数和参数
CoroutineEntryType entry_;
void *arg_;
CoroutineRuntime *runtime_;
};
每一个Coroutine类的对象代表一个协程,或者叫一个任务,因此需要保存该任务执行的上下文(context)。所谓的上下文,包括了CPU的状态(也就是各种寄存器的值)和栈空间。在当前协程放弃CPU的时候,必须保存下CPU的状态,这样下次切换回来才能继续执行。对于每一个协程,都需要分配独立的栈空间,这样才能保证它的函数调用栈不会被覆盖,在从其他协程切换回来之后,还能沿着正确的调用栈返回。上下文信息在Coroutine类中是通过context_和stack_两个成员变量来保存的。stack_是一个结构体,记录了栈空间的起始地址和当前的栈顶地址,结构定义如下:
struct Stack {
size_t size;
void *top;
void *base;
};
下面两个宏用来分配和释放栈空间:
#define ALLOC_STACK(stack, sz) \
do { \
int page_size = getpagesize(); \
stack.size = (sz + page_size - 1) / page_size * page_size; \
stack.base = aligned_alloc(page_size, stack.size); \
stack.top = stack.base; \
} while (0)
#define FREE_STACK(stack) \
do { \
if (stack.base) { \
free(stack.base); \
} \
stack.base = stack.top = NULL; \
stack.size = 0; \
} while (0)
POSIX标准里提供了getcontext()、setcontext()、makecontext()和swapcontext()四个API,目前的Linux版本都是支持这几个API的,这样我们就不需要自己编写汇编代码来处理上下文了。如果对具体的实现感兴趣,可以参考libtask里的asm.S。
成员变量context_实现上下文保存和切换的关键,其类型为ucontext_t。只要正确的对它进行初始化,就可以调用POSIX API来实现上下文的切换了。初始化的过程在Init()函数里进行,实现如下:
int Coroutine::Init(CoroutineRuntime *rt, size_t stack_size,
CoroutineEntryType fn, void *arg) {
TRY_AND_CHECK_ERRNO(ALLOC_STACK(stack_, stack_size), return -1);
getcontext(&context_);
context_.uc_stack.ss_sp = stack_.top;
context_.uc_stack.ss_size = stack_.size;
context_.uc_stack.ss_flags = 0;
context_.uc_link = GetMainContext();
uintptr_t ptr = (uintptr_t) this;
makecontext(&context_, (void (*)(void))Coroutine::Runner, 2, (uint32_t)ptr,
(uint32_t)(ptr >> 32));
runtime_ = rt;
entry_ = fn;
arg_ = arg;
return 0;
}
这里主要做了三件事:
调用getcontext()把当前上下文保存到context_。
修改context_,其中前三个字段是栈相关的,让协程使用自己的私有栈,最后一个字段用于在协程结束时返回之前的上下文。
-
最后调用makecontext()设置任务的入口。这里需要说明一下,makecontext()函数的第二个参数是入口函数,可以接受n个uint32_t类型的参数,其中n由第三个参数指定,从第四个参数开始,之后的n个参数会被传递给入口函数。在这个demo里,我们把本协程对象的指针(this)作为参数传递给入口函数,由于是64位环境,指针的长度为8字节,因此需要两个参数把高32位和低32位分别传进去。Runner()函数是所有协程的一个统一入口,它会调用协程对象的成员函数Run(),后者又会调用Init()时指定的函数fn。
void Coroutine::Runner(uint32_t low, uint32_t high) { uintptr_t ptr = (uintptr_t)low | ((uintptr_t)high << 32); Coroutine *task = (Coroutine *)ptr; assert(!task->IsDone()); task->Run(); task->Done(); }
跟上下文切换相关的两个函数是Yield()和Resume(),前者使当前任务放弃CPU,后者恢复当前任务的执行。实现如下:
void Coroutine::Yield() {
status_ = kTaskPending;
swapcontext(&context_, GetMainContext());
}
void Coroutine::Resume() {
status_ = kTaskRunning;
swapcontext(GetMainContext(), &context_);
}
这两个函数的实现都很简单,直接调用了系统提供的swapcontext(),把当前任务的上下文和主任务的上下文进行交换。GetMainContext()函数返回主任务的上下文,定义如下:
ucontext_t *Coroutine::GetMainContext() {
static __thread ucontext_t main_context;
return &main_context;
}
这里声明了一个局部静态变量main_context,并返回指向它的指针。修饰符__thread表示main_context是一个线程局部变量,即每个线程有一个单独的实例。当需要支持多线程的时候,就必须用__thread修饰符,因为我们显然不希望不同线程上的协程混在一起。
CoroutineRuntime为协程管理类,用来管理所有运行中的任务,声明如下:
class CoroutineRuntime {
public:
CoroutineRuntime(size_t stack_size, int max_tasks)
: stack_size_(stack_size), max_tasks_(max_tasks), current_(-1) {
tasks_.reserve(max_tasks);
};
~CoroutineRuntime(){};
int AddTask(CoroutineEntryType fn, void *arg);
void Done();
void Resume(int task_id);
void Yield();
void Run();
private:
size_t stack_size_;
int max_tasks_;
int current_;
std::vector<Coroutine *> tasks_;
std::deque<int> free_;
};
tasks_是任务列表,free_是已经完成的任务队列。当一个任务完成之后,其对应的Coroutine对象不会被释放,其指针仍然保存在tasks_里,但是会把它的下标加入到free_队列。当需要创建新任务时,首先会尝试从free_队列中取出一个下标,并复用之前的Coroutine对象,当free_队列为空时才会重新分配Coroutine对象。添加任务和标记任务完成的过程分别在AddTask()和Done()里实现:
int CoroutineRuntime::AddTask(CoroutineEntryType fn, void *arg) {
Coroutine *co;
int task_id = tasks_.size();
if (!free_.empty()) {
task_id = free_.front();
free_.pop_front();
co = tasks_[task_id];
} else if (task_id < max_tasks_) {
co = new Coroutine;
if (!co) {
return -1;
}
tasks_.push_back(co);
} else {
LOG(ERR, "reach max num of tasks");
return -1;
}
co->Init(this, stack_size_, fn, arg);
return task_id;
};
void CoroutineRuntime::Done() {
assert(current_ != -1);
if (current_ != -1) {
free_.push_back(current_);
current_ = -1;
}
};
成员函数Run()负责对所有任务进行调度。在这个demo里,我们使用最简单粗暴的方式:从头到尾遍历任务列表,依次唤醒每一个还没有结束的任务。在实际应用中,这个调度函数通常是事件驱动的,采用epoll/kqueue等机制,当某一个fd上有事件发生时,就切换到与之相关联的任务去执行。
void CoroutineRuntime::Run() {
while (1) {
for (int i = 0; i < tasks_.size(); i++) {
if (!tasks_[i]->IsDone()) {
LOG(INFO, "switch to task %d", i);
Resume(i);
} else {
LOG(INFO, "skip task %d", i);
}
sleep(1);
}
}
}
最后,还需要一个测试程序来验证协程能否正常工作。想当初Linus在开发Linux内核之初,只是做了一个简单的demo,让两个进程分别打印A和B。这里我们也来做一下类似的演示。下面代码创建两个协程,分别打印一段信息,每打印一次就让出CPU。
void taskfunc(CoroutineRuntime *runtime, void *arg) {
char *msg = (char *)arg;
while (1) {
LOG(INFO, "%s", msg);
runtime->Yield();
}
}
int main() {
char *msg1 = "a", *msg2 = "b";
CoroutineRuntime runtime(16 * 4096, 10);
runtime.AddTask(taskfunc, msg1);
runtime.AddTask(taskfunc, msg2);
runtime.Run();
return 0;
}
最后小结一下。协程是一个很不错的并发编程模型。一个协程的所需要的资源开销(包括内存和CPU)远远小于进程和线程,因此在强调高并发的服务器端很有用。但是,它只能解决并发问题,并不能解决并行问题。在多核处理器的环境下,还是需要结合多线程或者多进程来实现物理上的并行计算。另外,在比较复杂的应用场景下使用协程时,还要考虑到下游系统能否支撑。如果只是简单的通过协程来增加某个模块的并发度,就有可能出现下游系统被压垮的情况,这样也是没有意义的,甚至适得其反。