[boost] async socket i/o [long]

From: Michel André (michel.andre_at_[hidden])
Date: 2005-04-11 16:58:00

There has been some discussion on this on the list lately. Is there 
anyone working on it. I have been pondering an interface that supports 
both async and blocking io combined. And an event dispatching model that 
can handle several types of dispatching mechanism through a the same 
interface to provide for a lot of differnet implementations.

At heart of this is an io_stream base class that should serve as base 
for classes like tcp_stream and udp_stream.

class io_stream :
   public event::event_handler
{
public:			
   // blocking read
   virtual io_result read(
     void* buffer,
     size_t byteCount) = 0;

   // blocking write
   virtual io_result write(
     const void* buffer,
     size_t byteCount) = 0;

   // asynchrounous read.
   virtual io_result async_read(
     void* buffer,
     size_t byteCount,
     boost::function<void(const io_result&)>) = 0;

   // asynchrounous write.
   virtual io_result async_write(
     const void* buffer,
     size_t byteCount,
     boost::function<void(const io_result&)>) = 0;

   virtual void close() = 0;
protected:
};

the event_handler interface is at heart of event handling and dispatching.

class event_handler
{
  ublic:
   virtual ~event_handler();

   // returns a pointer to the dispatcher the source
   // is attached to or NULL if the source isn't attached.
   event_dispatcher* dispatcher();

protected:
   enum event_type_t
   {
     read_event, // A read can be done without blocking
     read_done_event, // An async read is completely done
     write_event, // A write can be done without blocking
     write_done_event, // An async write is completely done
     connect_event,
     accept_event
   };

   typedef std::bitset<6> event_set_t;

   // returns the current set of events we want to receive
   event_set_t event_filter() const;

   // updates the current event set by oring
   // current filter and the set provided
   void update_event_filter(event_set_t set);

   // an opaque handle to the underlying source
   virtual void* handle() = 0;

   /// callack when event occurs.
   /// event - type of event dispatched
   /// bytes - number of bytes affected by the event
   /// size_t(-1) if unknown.
   /// error - an event specific error code 0 if no error
   /// eventData - a void* pointer that contains opaque
   /// data supplied when initiating the operation NULL of none
   /// is supplied or not available
   void dispatch_event(
     event_type_t event,
     size_t bytes,
     unsigned int error,
     void* eventData) = 0;

private:
   event_handler(const event_handler&);
   event_handler& operator=(const event_handler&);
};

 From this we could be notified both of full async completes or just 
ready to write and ready to receive. And when a concrete io_stream 
implementation receives notification it can check on the flags if the 
read have completed or of we must do a read or write to fullfill the 
pending operation that must be stored within the implementation. This 
makes for implementations of the event dispatcher (including simple 
select, kqueue, iocp or some other mechanism both edge and level triggered).

The event dispatcher interface looks like:

class event_dispatcher
{
public:
   /// attaches an event handler to an dispatcher
   /// all dispatchable events for the event handlers
   /// handle will go through this dispatcher. an event handler
   /// and its associated handle is attached
   /// to a specific dispatcher throughout its life time
   virtual void attach(event_handler& source);

   /// Removes an event handler from set. notifications
   /// wont be sent to this dispatcher to the event handler
   /// after this point.
   virtual void detach(event_handler& source);

   /// notify the event_dispatcher that the event filter
   /// has changed for the given source
   virtual void notify(event_handler& source);

   /// dispatch pending events blocks timeoutSecs waiting
   /// for event. returns true if an event was dispatched
   /// during the call.
   virtual bool dispatch_event(unsigned int timeoutMsecs);

   /// adds a callback to be called asynchrounously the next
   /// as soon as possible.
   virtual void asynch_dispatch(boost::function<void> callback);
};

This can be implemented in various ways through simple select for single 
threaded version, and leader follower select for multithreaded with 
several threads hanging in dispatch_events and handling and event and 
promoting a new leader to catch next set. And it could be implemented 
using io completion ports or kqueue for scalable variants.

To top this of we have connector and acceptor base classes
class connector :
   public event::event_handler
{
public:
   std::auto_ptr<net::io_stream> connect_stream() = 0;
   void async_connect_stream(
     boost::function<void(const io_result&,
     std::auto_ptr<net::io_stream>)> callback) = 0;
};

class acceptor :
   public event::event_handler
{
public:
   std::auto_ptr<net::io_stream> accept_stream() = 0;
   void async_accept_stream(
     boost::function<
     void(const io_result&, std::auto_ptr<net::io_stream>)> callback) = 0;
};

And tcp_acceptor and tcp_connector to implement these for tcp.

Eg the tcp acceptor:

class tcp_acceptor :
   public acceptor
{
public:
   acceptor(const ip_endpoint& localEndPoint);
   acceptor(event::event_dispatcher& dispatcher, const ip_endpoint& 
localEndPoint);

   std::auto_ptr<net::tcp_stream> accept() = 0;
   void async_accept(
     boost::function<void(const io_result&,
     std::auto_ptr<net::tcp_stream>)> callback) = 0;

   /// Acceptor functions
   std::auto_ptr<net::io_stream> connect_stream() = 0;
   void async_connect_stream(
     boost::function<void(const io_result&,
     std::auto_ptr<net::io_stream>)> callback) = 0;

};

I think these interfaces would allow for a range of implementations 
under the same umbrella. The os dependent sockets part should be handled 
by a socket facade class.

The io_result class is a helper class that handles reporting of the 
actual result and eventual error outcome of an operation. It combines 
error codes and exception throwing.

// encapulates a result of an net operation
// throws an io_error if none of the following
// members haven't been called before destruction:
// failed,success,error_code,throw_exception.
struct io_result
{
   enum error_code_t
   {
     e_success,
     e_cancelled, // operation was cancelled
     e_pending_io // an net operation of the specified type is already 
pending
   };

   error_code_t error_code() const;
   bool failed() const;
   bool success() const;

   void throw_exception() const;

   // bytes transferred by the last operation
   size_t byte_count() const;
};

So this is the outline of an interface that I think could serve as a 
foundation for a net library. I will try to go about and implement 
prototype for this using both select and iocp. But there are some things 
to iron out such as thread saftey and such (since it will necessary to 
at least having several threads blocking in dispatch_event at least the 
event_dispatcher would have to be thread safe and probably parts of the 
event source d).

What are your initial thaugths on this, implementatble? flexible enough? 
efficient enough?

What are you others doing in this field? And maybe we should do 
something together instead if we are sitting in different cornes hacking 
on something similar. I know for one i need help with boost.build, how 
to produce documentation according to boost standard, multi platform 
testing and the like ;).

Regards
/Michel