[Boost-commit] svn:boost r60940 - trunk/boost/asio/detail

Subject: [Boost-commit] svn:boost r60940 - trunk/boost/asio/detail
From: chris_at_[hidden]
Date: 2010-03-30 08:31:52

Author: chris_kohlhoff
Date: 2010-03-30 08:31:51 EDT (Tue, 30 Mar 2010)
New Revision: 60940
URL: http://svn.boost.org/trac/boost/changeset/60940

Log:
New kqueue reactor implementation using one-shot event registration.

Text files modified: 
   trunk/boost/asio/detail/kqueue_reactor.hpp |   370 +++++++++++++++++++++------------------ 
   1 files changed, 200 insertions(+), 170 deletions(-)

Modified: trunk/boost/asio/detail/kqueue_reactor.hpp
==============================================================================
--- trunk/boost/asio/detail/kqueue_reactor.hpp	(original)
+++ trunk/boost/asio/detail/kqueue_reactor.hpp	2010-03-30 08:31:51 EDT (Tue, 30 Mar 2010)
@@ -34,10 +34,10 @@
 
 #include <boost/asio/error.hpp>
 #include <boost/asio/io_service.hpp>
+#include <boost/asio/detail/hash_map.hpp>
 #include <boost/asio/detail/mutex.hpp>
 #include <boost/asio/detail/op_queue.hpp>
 #include <boost/asio/detail/reactor_op.hpp>
-#include <boost/asio/detail/reactor_op_queue.hpp>
 #include <boost/asio/detail/select_interrupter.hpp>
 #include <boost/asio/detail/service_base.hpp>
 #include <boost/asio/detail/socket_types.hpp>
@@ -59,15 +59,24 @@
   : public boost::asio::detail::service_base<kqueue_reactor>
 {
 public:
-  enum { read_op = 0, write_op = 1,
+  enum op_types { read_op = 0, write_op = 1,
     connect_op = 1, except_op = 2, max_ops = 3 };
 
-  // Per-descriptor data.
-  struct per_descriptor_data
+  // Per-descriptor queues.
+  struct descriptor_state
   {
-    bool allow_speculative[max_ops];
+    descriptor_state() {}
+    descriptor_state(const descriptor_state&) {}
+    void operator=(const descriptor_state&) {}
+
+    mutex mutex_;
+    op_queue<reactor_op> op_queue_[max_ops];
+    bool shutdown_;
   };
 
+  // Per-descriptor data.
+  typedef descriptor_state* per_descriptor_data;
+
   // Constructor.
   kqueue_reactor(boost::asio::io_service& io_service)
     : boost::asio::detail::service_base<kqueue_reactor>(io_service),
@@ -75,34 +84,38 @@
       mutex_(),
       kqueue_fd_(do_kqueue_create()),
       interrupter_(),
-      shutdown_(false),
-      need_kqueue_wait_(true)
+      shutdown_(false)
   {
-    // Add the interrupter's descriptor to the kqueue.
-    struct kevent event;
-    EV_SET(&event, interrupter_.read_descriptor(),
-        EVFILT_READ, EV_ADD, 0, 0, 0);
-    ::kevent(kqueue_fd_, &event, 1, 0, 0, 0);
+    // The interrupter is put into a permanently readable state. Whenever we
+    // want to interrupt the blocked kevent call we register a one-shot read
+    // operation against the descriptor.
+    interrupter_.interrupt();
   }
 
   // Destructor.
   ~kqueue_reactor()
   {
-    shutdown_service();
     close(kqueue_fd_);
   }
 
   // Destroy all user-defined handler objects owned by the service.
   void shutdown_service()
   {
-    boost::asio::detail::mutex::scoped_lock lock(mutex_);
+    mutex::scoped_lock lock(mutex_);
     shutdown_ = true;
     lock.unlock();
 
     op_queue<operation> ops;
 
-    for (int i = 0; i < max_ops; ++i)
-      op_queue_[i].get_all_operations(ops);
+    descriptor_map::iterator iter = registered_descriptors_.begin();
+    descriptor_map::iterator end = registered_descriptors_.end();
+    while (iter != end)
+    {
+      for (int i = 0; i < max_ops; ++i)
+        ops.push(iter->second.op_queue_[i]);
+      iter->second.shutdown_ = true;
+      ++iter;
+    }
 
     timer_queues_.get_all_timers(ops);
   }
@@ -115,11 +128,16 @@
 
   // Register a socket with the reactor. Returns 0 on success, system error
   // code on failure.
-  int register_descriptor(socket_type, per_descriptor_data& descriptor_data)
+  int register_descriptor(socket_type descriptor,
+      per_descriptor_data& descriptor_data)
   {
-    descriptor_data.allow_speculative[read_op] = true;
-    descriptor_data.allow_speculative[write_op] = true;
-    descriptor_data.allow_speculative[except_op] = true;
+    mutex::scoped_lock lock(registered_descriptors_mutex_);
+
+    descriptor_map::iterator new_entry = registered_descriptors_.insert(
+          std::make_pair(descriptor, descriptor_state())).first;
+    descriptor_data = &new_entry->second;
+
+    descriptor_data->shutdown_ = false;
 
     return 0;
   }
@@ -130,66 +148,57 @@
       per_descriptor_data& descriptor_data,
       reactor_op* op, bool allow_speculative)
   {
-    if (allow_speculative && descriptor_data.allow_speculative[op_type])
-    {
-      if (op->perform())
-      {
-        io_service_.post_immediate_completion(op);
-        return;
-      }
-
-      // We only get one shot at a speculative read in this function.
-      allow_speculative = false;
-    }
-
-    boost::asio::detail::mutex::scoped_lock lock(mutex_);
-
-    if (shutdown_)
+    mutex::scoped_lock descriptor_lock(descriptor_data->mutex_);
+    if (descriptor_data->shutdown_)
       return;
 
-    if (!allow_speculative)
-      need_kqueue_wait_ = true;
-    else if (!op_queue_[op_type].has_operation(descriptor))
+    bool first = descriptor_data->op_queue_[op_type].empty();
+    if (first)
     {
-      // Speculative reads are ok as there are no queued read operations.
-      descriptor_data.allow_speculative[op_type] = true;
-
-      if (op->perform())
+      if (allow_speculative)
       {
-        lock.unlock();
-        io_service_.post_immediate_completion(op);
-        return;
+        if (op_type != read_op || descriptor_data->op_queue_[except_op].empty())
+        {
+          if (op->perform())
+          {
+            descriptor_lock.unlock();
+            io_service_.post_immediate_completion(op);
+            return;
+          }
+        }
       }
     }
 
-    // Speculative reads are not ok as there will be queued read operations.
-    descriptor_data.allow_speculative[op_type] = false;
-
-    bool first = op_queue_[op_type].enqueue_operation(descriptor, op);
+    descriptor_data->op_queue_[op_type].push(op);
     io_service_.work_started();
+
     if (first)
     {
       struct kevent event;
       switch (op_type)
       {
       case read_op:
-        EV_SET(&event, descriptor, EVFILT_READ, EV_ADD, 0, 0, 0);
+        EV_SET(&event, descriptor, EVFILT_READ,
+            EV_ADD | EV_ONESHOT, 0, 0, descriptor_data);
         break;
       case write_op:
-        EV_SET(&event, descriptor, EVFILT_WRITE, EV_ADD, 0, 0, 0);
+        EV_SET(&event, descriptor, EVFILT_WRITE,
+            EV_ADD | EV_ONESHOT, 0, 0, descriptor_data);
         break;
       case except_op:
-        if (op_queue_[read_op].has_operation(descriptor))
-          EV_SET(&event, descriptor, EVFILT_READ, EV_ADD, 0, 0, 0);
-        else
-          EV_SET(&event, descriptor, EVFILT_WRITE, EV_ADD, EV_OOBAND, 0, 0);
+        if (!descriptor_data->op_queue_[read_op].empty())
+          return; // Already registered for read events.
+        EV_SET(&event, descriptor, EVFILT_READ,
+            EV_ADD | EV_ONESHOT, EV_OOBAND, 0, descriptor_data);
         break;
       }
+
       if (::kevent(kqueue_fd_, &event, 1, 0, 0, 0) == -1)
       {
-        boost::system::error_code ec(errno,
+        op->ec_ = boost::system::error_code(errno,
             boost::asio::error::get_system_category());
-        cancel_ops_unlocked(descriptor, ec);
+        descriptor_data->op_queue_[op_type].pop();
+        io_service_.post_deferred_completion(op);
       }
     }
   }
@@ -197,33 +206,63 @@
   // Cancel all operations associated with the given descriptor. The
   // handlers associated with the descriptor will be invoked with the
   // operation_aborted error.
-  void cancel_ops(socket_type descriptor, per_descriptor_data&)
+  void cancel_ops(socket_type descriptor, per_descriptor_data& descriptor_data)
   {
-    boost::asio::detail::mutex::scoped_lock lock(mutex_);
-    cancel_ops_unlocked(descriptor, boost::asio::error::operation_aborted);
+    mutex::scoped_lock descriptor_lock(descriptor_data->mutex_);
+
+    op_queue<operation> ops;
+    for (int i = 0; i < max_ops; ++i)
+    {
+      while (reactor_op* op = descriptor_data->op_queue_[i].front())
+      {
+        op->ec_ = boost::asio::error::operation_aborted;
+        descriptor_data->op_queue_[i].pop();
+        ops.push(op);
+      }
+    }
+
+    descriptor_lock.unlock();
+
+    io_service_.post_deferred_completions(ops);
   }
 
   // Cancel any operations that are running against the descriptor and remove
   // its registration from the reactor.
-  void close_descriptor(socket_type descriptor, per_descriptor_data&)
+  void close_descriptor(socket_type descriptor,
+      per_descriptor_data& descriptor_data)
   {
-    boost::asio::detail::mutex::scoped_lock lock(mutex_);
+    mutex::scoped_lock descriptor_lock(descriptor_data->mutex_);
+    mutex::scoped_lock descriptors_lock(registered_descriptors_mutex_);
+
+    // Remove the descriptor from the set of known descriptors. The descriptor
+    // will be automatically removed from the kqueue set when it is closed.
+    descriptor_data->shutdown_ = true;
+
+    op_queue<operation> ops;
+    for (int i = 0; i < max_ops; ++i)
+    {
+      while (reactor_op* op = descriptor_data->op_queue_[i].front())
+      {
+        op->ec_ = boost::asio::error::operation_aborted;
+        descriptor_data->op_queue_[i].pop();
+        ops.push(op);
+      }
+    }
+
+    descriptor_lock.unlock();
+
+    registered_descriptors_.erase(descriptor);
 
-    // Remove the descriptor from kqueue.
-    struct kevent event[2];
-    EV_SET(&event[0], descriptor, EVFILT_READ, EV_DELETE, 0, 0, 0);
-    EV_SET(&event[1], descriptor, EVFILT_WRITE, EV_DELETE, 0, 0, 0);
-    ::kevent(kqueue_fd_, event, 2, 0, 0, 0);
-    
-    // Cancel any outstanding operations associated with the descriptor.
-    cancel_ops_unlocked(descriptor, boost::asio::error::operation_aborted);
+    descriptors_lock.unlock();
+
+    io_service_.post_deferred_completions(ops);
   }
 
   // Add a new timer queue to the reactor.
   template <typename Time_Traits>
   void add_timer_queue(timer_queue<Time_Traits>& timer_queue)
   {
-    boost::asio::detail::mutex::scoped_lock lock(mutex_);
+    mutex::scoped_lock lock(mutex_);
     timer_queues_.insert(&timer_queue);
   }
 
@@ -231,7 +270,7 @@
   template <typename Time_Traits>
   void remove_timer_queue(timer_queue<Time_Traits>& timer_queue)
   {
-    boost::asio::detail::mutex::scoped_lock lock(mutex_);
+    mutex::scoped_lock lock(mutex_);
     timer_queues_.erase(&timer_queue);
   }
 
@@ -241,13 +280,13 @@
   void schedule_timer(timer_queue<Time_Traits>& timer_queue,
       const typename Time_Traits::time_type& time, timer_op* op, void* token)
   {
-    boost::asio::detail::mutex::scoped_lock lock(mutex_);
+    mutex::scoped_lock lock(mutex_);
     if (!shutdown_)
     {
       bool earliest = timer_queue.enqueue_timer(time, op, token);
       io_service_.work_started();
       if (earliest)
-        interrupter_.interrupt();
+        interrupt();
     }
   }
 
@@ -256,7 +295,7 @@
   template <typename Time_Traits>
   std::size_t cancel_timer(timer_queue<Time_Traits>& timer_queue, void* token)
   {
-    boost::asio::detail::mutex::scoped_lock lock(mutex_);
+    mutex::scoped_lock lock(mutex_);
     op_queue<operation> ops;
     std::size_t n = timer_queue.cancel_timer(token, ops);
     lock.unlock();
@@ -267,13 +306,7 @@
   // Run the kqueue loop.
   void run(bool block, op_queue<operation>& ops)
   {
-    boost::asio::detail::mutex::scoped_lock lock(mutex_);
-
-    // We can return immediately if there's no work to do and the reactor is
-    // not supposed to block.
-    if (!block && op_queue_[read_op].empty() && op_queue_[write_op].empty()
-        && op_queue_[except_op].empty() && timer_queues_.all_empty())
-      return;
+    mutex::scoped_lock lock(mutex_);
 
     // Determine how long to block while waiting for events.
     timespec timeout_buf = { 0, 0 };
@@ -283,103 +316,104 @@
 
     // Block on the kqueue descriptor.
     struct kevent events[128];
-    int num_events = (block || need_kqueue_wait_)
-      ? kevent(kqueue_fd_, 0, 0, events, 128, timeout)
-      : 0;
-
-    lock.lock();
+    int num_events = kevent(kqueue_fd_, 0, 0, events, 128, timeout);
 
     // Dispatch the waiting events.
     for (int i = 0; i < num_events; ++i)
     {
       int descriptor = events[i].ident;
-      if (descriptor == interrupter_.read_descriptor())
+      void* ptr = events[i].udata;
+      if (ptr == &interrupter_)
       {
-        interrupter_.reset();
+        // No need to reset the interrupter since we're leaving the descriptor
+        // in a ready-to-read state and relying on one-shot notifications.
       }
-      else if (events[i].filter == EVFILT_READ)
+      else
       {
-        // Dispatch operations associated with the descriptor.
-        bool more_reads = false;
-        bool more_except = false;
-        if (events[i].flags & EV_ERROR)
-        {
-          boost::system::error_code error(
-              events[i].data, boost::asio::error::get_system_category());
-          op_queue_[except_op].perform_operations(descriptor, ops);
-          op_queue_[read_op].perform_operations(descriptor, ops);
-        }
-        else if (events[i].flags & EV_OOBAND)
-        {
-          more_except
-            = op_queue_[except_op].perform_operations(descriptor, ops);
-          if (events[i].data > 0)
-            more_reads = op_queue_[read_op].perform_operations(descriptor, ops);
-          else
-            more_reads = op_queue_[read_op].has_operation(descriptor);
-        }
-        else
+        descriptor_state* descriptor_data = static_cast<descriptor_state*>(ptr);
+        mutex::scoped_lock descriptor_lock(descriptor_data->mutex_);
+
+        // Exception operations must be processed first to ensure that any
+        // out-of-band data is read before normal data.
+        static const int filter[max_ops] =
+          { EVFILT_READ, EVFILT_WRITE, EVFILT_READ };
+        for (int j = max_ops - 1; j >= 0; --j)
         {
-          more_reads = op_queue_[read_op].perform_operations(descriptor, ops);
-          more_except = op_queue_[except_op].has_operation(descriptor);
+          if (events[i].filter == filter[j])
+          {
+            if (j != except_op || events[i].flags & EV_OOBAND)
+            {
+              while (reactor_op* op = descriptor_data->op_queue_[j].front())
+              {
+                if (events[i].flags & EV_ERROR)
+                {
+                  op->ec_ = boost::system::error_code(events[i].data,
+                      boost::asio::error::get_system_category());
+                  descriptor_data->op_queue_[j].pop();
+                  ops.push(op);
+                }
+                if (op->perform())
+                {
+                  descriptor_data->op_queue_[j].pop();
+                  ops.push(op);
+                }
+                else
+                  break;
+              }
+            }
+          }
         }
 
-        // Update the descriptor in the kqueue.
+        // Renew registration for event notifications.
         struct kevent event;
-        if (more_reads)
-          EV_SET(&event, descriptor, EVFILT_READ, EV_ADD, 0, 0, 0);
-        else if (more_except)
-          EV_SET(&event, descriptor, EVFILT_READ, EV_ADD, EV_OOBAND, 0, 0);
-        else
-          EV_SET(&event, descriptor, EVFILT_READ, EV_DELETE, 0, 0, 0);
-        if (::kevent(kqueue_fd_, &event, 1, 0, 0, 0) == -1)
-        {
-          boost::system::error_code error(errno,
-              boost::asio::error::get_system_category());
-          op_queue_[except_op].cancel_operations(descriptor, ops, error);
-          op_queue_[read_op].cancel_operations(descriptor, ops, error);
-        }
-      }
-      else if (events[i].filter == EVFILT_WRITE)
-      {
-        // Dispatch operations associated with the descriptor.
-        bool more_writes = false;
-        if (events[i].flags & EV_ERROR)
+        switch (events[i].filter)
         {
-          boost::system::error_code error(
-              events[i].data, boost::asio::error::get_system_category());
-          op_queue_[write_op].cancel_operations(descriptor, ops, error);
-        }
-        else
-        {
-          more_writes = op_queue_[write_op].perform_operations(descriptor, ops);
+        case EVFILT_READ:
+          if (!descriptor_data->op_queue_[read_op].empty())
+            EV_SET(&event, descriptor, EVFILT_READ,
+                EV_ADD | EV_ONESHOT, 0, 0, descriptor_data);
+          else if (!descriptor_data->op_queue_[except_op].empty())
+            EV_SET(&event, descriptor, EVFILT_READ,
+                EV_ADD | EV_ONESHOT, EV_OOBAND, 0, descriptor_data);
+          else
+            continue;
+        case EVFILT_WRITE:
+          if (!descriptor_data->op_queue_[write_op].empty())
+            EV_SET(&event, descriptor, EVFILT_WRITE,
+                EV_ADD | EV_ONESHOT, 0, 0, descriptor_data);
+          else
+            continue;
+        default:
+          break;
         }
-
-        // Update the descriptor in the kqueue.
-        struct kevent event;
-        if (more_writes)
-          EV_SET(&event, descriptor, EVFILT_WRITE, EV_ADD, 0, 0, 0);
-        else
-          EV_SET(&event, descriptor, EVFILT_WRITE, EV_DELETE, 0, 0, 0);
         if (::kevent(kqueue_fd_, &event, 1, 0, 0, 0) == -1)
         {
           boost::system::error_code error(errno,
               boost::asio::error::get_system_category());
-          op_queue_[write_op].cancel_operations(descriptor, ops, error);
+          for (int j = 0; j < max_ops; ++j)
+          {
+            while (reactor_op* op = descriptor_data->op_queue_[j].front())
+            {
+              op->ec_ = error;
+              descriptor_data->op_queue_[j].pop();
+              ops.push(op);
+            }
+          }
         }
       }
     }
-    timer_queues_.get_ready_timers(ops);
 
-    // Determine whether kqueue needs to be called next time the reactor is run.
-    need_kqueue_wait_ = !op_queue_[read_op].empty()
-      || !op_queue_[write_op].empty() || !op_queue_[except_op].empty();
+    lock.lock();
+    timer_queues_.get_ready_timers(ops);
   }
 
-  // Interrupt the select loop.
+  // Interrupt the kqueue loop.
   void interrupt()
   {
-    interrupter_.interrupt();
+    struct kevent event;
+    EV_SET(&event, interrupter_.read_descriptor(),
+        EVFILT_READ, EV_ADD | EV_ONESHOT, 0, 0, &interrupter_);
+    ::kevent(kqueue_fd_, &event, 1, 0, 0, 0);
   }
 
 private:
@@ -410,22 +444,11 @@
     return &ts;
   }
 
-  // Cancel all operations associated with the given descriptor. This function
-  // does not acquire the kqueue_reactor's mutex.
-  void cancel_ops_unlocked(socket_type descriptor,
-      const boost::system::error_code& ec)
-  {
-    op_queue<operation> ops;
-    for (int i = 0; i < max_ops; ++i)
-      op_queue_[i].cancel_operations(descriptor, ops, ec);
-    io_service_.post_deferred_completions(ops);
-  }
-
   // The io_service implementation used to post completions.
   io_service_impl& io_service_;
 
   // Mutex to protect access to internal data.
-  boost::asio::detail::mutex mutex_;
+  mutex mutex_;
 
   // The kqueue file descriptor.
   int kqueue_fd_;
@@ -433,17 +456,24 @@
   // The interrupter is used to break a blocking kevent call.
   select_interrupter interrupter_;
 
-  // The queues of read, write and except operations.
-  reactor_op_queue<socket_type> op_queue_[max_ops];
-
   // The timer queues.
   timer_queue_set timer_queues_;
 
   // Whether the service has been shut down.
   bool shutdown_;
 
-  // Whether we need to call kqueue the next time the reactor is run.
-  bool need_kqueue_wait_;
+  // Mutex to protect access to the registered descriptors.
+  mutex registered_descriptors_mutex_;
+
+  // Keep track of all registered descriptors. This code relies on the fact that
+  // the hash_map implementation pools deleted nodes, meaning that we can assume
+  // our descriptor_state pointer remains valid even after the entry is removed.
+  // Technically this is not true for C++98, as that standard says that spliced
+  // elements in a list are invalidated. However, C++0x fixes this shortcoming
+  // so we'll just assume that C++98 std::list implementations will do the right
+  // thing anyway.
+  typedef detail::hash_map<socket_type, descriptor_state> descriptor_map;
+  descriptor_map registered_descriptors_;
 };
 
 } // namespace detail