New kqueue reactor implementation using one-shot event registration.
diff --git a/asio/include/asio/detail/kqueue_reactor.hpp b/asio/include/asio/detail/kqueue_reactor.hpp index ee3017c..1e118b3 100644 --- a/asio/include/asio/detail/kqueue_reactor.hpp +++ b/asio/include/asio/detail/kqueue_reactor.hpp
@@ -34,10 +34,10 @@ #include "asio/error.hpp" #include "asio/io_service.hpp" #include "asio/system_error.hpp" +#include "asio/detail/hash_map.hpp" #include "asio/detail/mutex.hpp" #include "asio/detail/op_queue.hpp" #include "asio/detail/reactor_op.hpp" -#include "asio/detail/reactor_op_queue.hpp" #include "asio/detail/select_interrupter.hpp" #include "asio/detail/service_base.hpp" #include "asio/detail/socket_types.hpp" @@ -58,15 +58,24 @@ : public 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(asio::io_service& io_service) : asio::detail::service_base<kqueue_reactor>(io_service), @@ -74,34 +83,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() { - 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); } @@ -114,11 +127,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; } @@ -129,66 +147,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; - } - - 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) { - asio::error_code ec(errno, + op->ec_ = asio::error_code(errno, asio::error::get_system_category()); - cancel_ops_unlocked(descriptor, ec); + descriptor_data->op_queue_[op_type].pop(); + io_service_.post_deferred_completion(op); } } } @@ -196,33 +205,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) { - asio::detail::mutex::scoped_lock lock(mutex_); - cancel_ops_unlocked(descriptor, 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_ = 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) { - 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 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, asio::error::operation_aborted); + // 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_ = asio::error::operation_aborted; + descriptor_data->op_queue_[i].pop(); + ops.push(op); + } + } + + descriptor_lock.unlock(); + + registered_descriptors_.erase(descriptor); + + 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) { - asio::detail::mutex::scoped_lock lock(mutex_); + mutex::scoped_lock lock(mutex_); timer_queues_.insert(&timer_queue); } @@ -230,7 +269,7 @@ template <typename Time_Traits> void remove_timer_queue(timer_queue<Time_Traits>& timer_queue) { - asio::detail::mutex::scoped_lock lock(mutex_); + mutex::scoped_lock lock(mutex_); timer_queues_.erase(&timer_queue); } @@ -240,13 +279,13 @@ void schedule_timer(timer_queue<Time_Traits>& timer_queue, const typename Time_Traits::time_type& time, timer_op* op, void* token) { - 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(); } } @@ -255,7 +294,7 @@ template <typename Time_Traits> std::size_t cancel_timer(timer_queue<Time_Traits>& timer_queue, void* token) { - 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(); @@ -266,13 +305,7 @@ // Run the kqueue loop. void run(bool block, op_queue<operation>& ops) { - 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 }; @@ -282,103 +315,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) + 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) { - asio::error_code error( - events[i].data, asio::error::get_system_category()); - op_queue_[except_op].perform_operations(descriptor, ops); - op_queue_[read_op].perform_operations(descriptor, ops); + 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_ = asio::error_code(events[i].data, + 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; + } + } + } } - else if (events[i].flags & EV_OOBAND) + + // Renew registration for event notifications. + struct kevent event; + switch (events[i].filter) { - 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); + 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 - more_reads = op_queue_[read_op].has_operation(descriptor); + 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; } - else - { - more_reads = op_queue_[read_op].perform_operations(descriptor, ops); - more_except = op_queue_[except_op].has_operation(descriptor); - } - - // Update the descriptor in the kqueue. - 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) { asio::error_code error(errno, 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) - { - asio::error_code error( - events[i].data, 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); - } - - // 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) - { - asio::error_code error(errno, - 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: @@ -409,22 +443,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 asio::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. - asio::detail::mutex mutex_; + mutex mutex_; // The kqueue file descriptor. int kqueue_fd_; @@ -432,17 +455,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