tidy code for asynchronous IO

Question

Whilst asynchronous IO (non-blocking descriptors with select/poll/epoll/kqueue etc) is not the most documented thing on the web, there are a handful of good examples.

However, all these examples, having determined the handles that are returned by the call, just have a 'do_some_io(fd)' stub. They don't really explain how to best approach the actual asynchronous IO in such a method.

Blocking IO is very tidy and straightforward to read code. Non-blocking, async IO is, on the other hand, hairy and messy.

What approaches are there? What are robust and readable?

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:
       ...
  }
}

or perhaps (ab)using GCC's computed gotos:

#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));
}

Or perhaps C++ exceptions and a state machine, so worker functions can trigger the abort/resume bit, or perhaps a table-driven state-machine?

Its not how to make it work, its how to make it maintainable that I'm chasing!

Answer 1

I suggest take a look on: http://www.kegel.com/c10k.html, second take a look on existing libraries like libevent, Boost.Asio that already do the job and see how they work.

The point is that the approach may be different for each type of system call:

• select is simple reactor
• epoll have both edge or level triggered interface that require different approach
• iocp is proactor require other approach

Suggestion: use good existing library like Boost.Asio for C++ or libevent for C.

EDIT: This is how ASIO handles this

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();
   }

};

Because ASIO works as proactor it notifies you when operation is complete and handles EWOULDBLOCK internally.

If you word as reactor you may simulate this behavior:

 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);
    }
 }

Something like that.

Answer 2

You need to have a main loop that provides async_schedule(), async_foreach(), async_tick() etc. These functions in turn place entries into a global list of methods that will run upon next call to async_tick(). Then you can write code that is much more tidy and does not include any switch statements.

You can just write:

async_schedule(callback, arg, timeout); 

Or:

async_wait(condition, callback, arg, timeout); 

Then your condition can even be set in another thread (provided that you take care of thread safety when accessing that variable).

I have implemented an async framework in C for my embedded project because I wanted to have non-preemptive multitasking and async is perfect for doing many tasks by doing a little bit of work during every iteration of the main loop.

The code is here: https://github.com/mkschreder/fortmax-blocks/blob/master/common/kernel/async.c

Answer 3

Great design pattern "coroutine" exists to solve this problem.

It's the best of both worlds: tidy code, exactly like synchronous io flow and great performance without context switching, like async io gives. Coroutine looks inside like an odinary synchronous thread, with single instruction pointer. But many coroutines can run within one OS thread (so-called "cooperative multitasking").

Example coroutine code:

void do_some_io() {
   blocking_read(something,sizeof(something));
   blocking_read(something_else,sizeof(something_else));
   blocking_write(something,sizeof(something));
}

Looks like synchronous code, but in fact control flow use another way, like this:

void do_some_io() {
   // return control to network io scheduler, to handle another coroutine
   blocking_read(something,sizeof(something)); 
   // when "something" is read, scheduler fill given buffer and resume this coroutine 

   // return control to network io scheduler, to handle another coroutine
   CoroSleep( 1000 );
   // scheduler create async timer and when it fires, scheduler pass control to this coroutine
    ...
   // and so on 

So single threaded scheduler control many coroutines with user-defined code and tidy synchronous-like calls to io.

C++ coroutines implementation example is "boost.coroutine" (actually not a part of boost :) http://www.crystalclearsoftware.com/soc/coroutine/ This library fully implements coroutine mechanics and can use boost.asio as scheduler and async io layer.

Answer 4

You want to decouple "io" from processing, at which point the code you read will become very readable. Basically you have:

    int read_io_event(...) { /* triggers when we get a read event from epoll/poll/whatever */

     /* read data from "fd" into a vstr/buffer/whatever */

     if (/* read failed */) /* return failure code to event callback */ ;

     if (/* "message" received */) return process_io_event();

     if (/* we've read "too much" */) /* return failure code to event callback */ ;

     return /* keep going code for event callback */ ;
    }


    int process_io_event(...) {
       /* this is where you process the HTTP request/whatever */
    }

...then the real code is in process event, and even if you have multiple requests responses it's pretty readable, you just do "return read_io_event()" after setting a state or whatever.

Answer 5

State machines are one nice approach. It's a bit of complexity up front that'll save you headaches in the future, where the future starts really, really soon. ;-)

Another method is to use threads and do blocking I/O on a single fd in each thread. The trade-off here is that you make I/O simple but may introduce complexity in synchronization.