结合Coroutine和Callback

Coroutine切换成本相对还是比较高的，把一个并发程序改成Coroutine实现性能上可能有比较大的损失。Coroutine切换主要是大量寄存器压栈和弹栈，栈切换也会影响到cache。目前C/C++大量使用的是回调方式，比如win32窗口编程、libevent等，优点是高效，函数调用成本是很低的，也没有栈切换，缺点是不如Coroutine这样连贯。有没有方法让它具有两者的优点呢？有Coroutine连贯的过程，也有Callback的高效？还是考虑这个简单的例子：

（本文代码使用D语法，但并没有编译测试）

void client_loop(int fd) {
  while(true) {
    register_read_event(fd);
    wait_read_event();
    char buf[1024];
    int ret = recv(fd, buf.ptr, buf.length, 0);
    if (ret <= 0) {
       writefln("recv fail");
       remove_events(fd);
       close(fd);
       break;
    }
    writefln("received:", buf[0..ret]);
  }
}

void server_loop(int fd) {
  while(true) {
    register_read_event(fd);
    wait_read_event();
    int client = accept(fd);
    spawn(client_loop, client);
  }
}

假定它是以Coroutine方式来实现的，Coroutine这样来实现也是很简单的。对应的Callback是什么形式呢？我假想了一个：

void client_loop(int fd) {
    register_read_event(fd)
      .on_event({
        char buf[1024];
        int ret = recv(fd, buf.ptr, buf.length, 0);
        if (ret <= 0) {
           writefln("recv fail");
           remove_events(fd);
           close(fd);
        }
        else {
          writefln("received:", buf[0..ret]);
          client_loop(fd);
        }
      });
}

void server_loop(int fd) {
  register_read_event(fd)
    .on_event({
      int clientfd = accept(fd);
      client_loop(clientfd);
      server_loop(fd);
    });
}

注意这并不是递归，因为server_loop和client_loop是立即返回的，里面那个代码块只是closure。

这段代码可读性明显不如上面好，你必须时刻思考需要在哪里创建一个closure，只因为并发时需要切换，当然下面这个实现效率会好得多。有没有办法用Coroutine这种连贯的写法达到Callback这样的效率呢？这是最近想要研究的问题。。

评论

4 楼 qiezi 2008-04-01  

main里的ev没用的，测试时忘了删掉。。

Event对象是由gc来管理的，所以我需要在EventManager里面加个引用防止它在使用过程中被干掉。

3 楼 sleets 2008-03-31  

感觉闭包把可以省下很多函数或变量名，另外逻辑上也比较紧凑了。

2 楼 sleets 2008-03-31  

test2.d里动态生成的Event什么时候析构啊？

另外不明白在main里创建的Event ev是做什么用的。。

1 楼 qiezi 2008-03-31  

上面这段callback方式的D语言代码我用libevent简单测试过了，还没发现问题。

有兴趣可以看看代码：

http://jies.googlecode.com/svn/trunk/libeventd

里面的test2.d是个简单的并发echo服务器，需要DMD 2.0x以上版本，具体哪个版本我忘了，至少要支持closure的。

test2.d:

void server_loop(EventManager em, Socket sock) {
	em.add_read_event(sock.handle(), EVENT_MASK.PERSIST)
		.on_event((Event event){
			writefln("do accept");
			Socket client = sock.accept();
			em.add_read_event(client.handle(), EVENT_MASK.PERSIST, 3000)
				.on_event((Event event){
					writefln("do recv");
					char[] buf;
					buf.length = 1024;
					int ret = client.receive(buf);
					if (ret <= 0) {
						writefln("recv fail");
						em.remove_event(event);
						client.close();
					}
					buf.length = ret;
					em.add_write_event(client.handle())
						.on_event((Event event){
							writefln("do send");
							int ret = client.send(buf);
							if (ret <= 0) {
								writefln("send fail");
								em.remove_event(event);
								client.close();
							}
						});
				});
		});
}