整洁的异步IO代码

Question

虽然异步IO(带有select/poll/epoll/kqueue等的非阻塞描述符)并不是Web上记录最多的东西,但有一些很好的例子.

但是,所有这些示例在确定了调用返回的句柄后,只有一个do_some_io(fd)"存根".它们并没有真正解释如何在这种方法中最好地接近实际的异步IO.

阻止IO非常整洁,直接读取代码.另一方面,非阻塞,异步IO是毛茸茸的,凌乱的.

有什么办法？什么是健壮和可读的？

void do_some_io(int fd) {
  switch(state) {
    case STEP1:
       ... async calls
       if(io_would_block)
          return;
       state = STEP2;
    case STEP2:
       ... more async calls
       if(io_would_block)
          return;
       state = STEP3;
    case STEP3:
       ...
  }
}

或者(ab)使用GCC的计算得到的:

#define concatentate(x,y) x##y
#define async_read_xx(var,bytes,line)       \
   concatentate(jmp,line):                  \
   if(!do_async_read(bytes,&var)) {         \
       schedule(EPOLLIN);                   \
       jmp_read = &&concatentate(jmp,line); \
       return;                              \
}

// macros for making async code read like sync code
#define async_read(var,bytes) \
    async_read_xx(var,bytes,__LINE__)

#define async_resume()            \
     if(jmp_read) {               \
         void* target = jmp_read; \
         jmp_read = NULL;         \
         goto *target;            \
     }

void do_some_io() {
   async_resume();
   async_read(something,sizeof(something));
   async_read(something_else,sizeof(something_else));
}

或者可能是C++异常和状态机,因此工作函数可以触发中止/恢复位,或者可能是表驱动的状态机？

它不是如何让它工作,它是如何让它可以维持我追逐!

Answer 1

我建议看看:http://www.kegel.com/c10k.html,第二,看看现有的库,比如libevent,Boost.Asio已经完成这项工作,看看它们是如何工作的.

关键是每种类型的系统调用方法可能不同:

• 选择是简单的反应堆
• epoll有边缘或水平触发的接口,需要不同的方法
• iocp是proactor需要其他方法

建议:使用良好的现有库,如Boost.Asio for C++或libevent for C.

编辑:这是ASIO如何处理这个问题

class connection {
   boost::asio:ip::tcp::socket socket_;
public:
   void run()
   {
         // for variable length chunks
         async_read_until(socket_,resizable_buffer,'\n',
               boost::bind(&run::on_line_recieved,this,errorplacehplder);
         // or constant length chunks
         async_read(socket_,buffer(some_buf,buf_size),
               boost::bind(&run::on_line_recieved,this,errorplacehplder);
   }
   void on_line_recieved(error e)
   {
        // handle it
        run();
   }

};

因为ASIO作为proactor工作,它会在操作完成时通知您并在内部处理EWOULDBLOCK.

如果你称为反应堆,你可以模拟这种行为:

 class conn {
    // Application logic

    void run() {
       read_chunk(&conn::on_chunk_read,size);
    }
    void on_chunk_read() {
         /* do something;*/
    }

    // Proactor wrappers

    void read_chunk(void (conn::*callback),int size, int start_point=0) {
       read(socket,buffer+start,size)
       if( complete )
          (this->*callback()
       else {
          this -> tmp_size-=size-read;
          this -> tmp_start=start+read;
          this -> tmp_callback=callback
          your_event_library_register_op_on_readable(callback,socket,this);
       }
    }
    void callback()
    {
       read_chunk(tmp_callback,tmp_size,tmp_start);
    }
 }

这样的事情.

Answer 2

状态机是一种很好的方法.事先有点复杂性,这将使您在未来真正地,非常快地开始未来的头痛.;-)

另一种方法是使用线程并在每个线程中的单个fd上执行阻塞I/O. 这里的权衡是你使I/O变得简单,但可能会引入同步的复杂性.