Register a descriptor's kqueue event filters once only i.e. when descriptor is first created.
diff --git a/asio/include/asio/detail/impl/kqueue_reactor.ipp b/asio/include/asio/detail/impl/kqueue_reactor.ipp index a5da0f0..b11d226 100644 --- a/asio/include/asio/detail/impl/kqueue_reactor.ipp +++ b/asio/include/asio/detail/impl/kqueue_reactor.ipp
@@ -46,10 +46,15 @@ interrupter_(), shutdown_(false) { - // The interrupter is put into a permanently readable state. Whenever we want - // to interrupt the blocked kevent call we register a read operation against - // the descriptor. - interrupter_.interrupt(); + struct kevent event; + ASIO_KQUEUE_EV_SET(&event, interrupter_.read_descriptor(), + EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, &interrupter_); + if (::kevent(kqueue_fd_, &event, 1, 0, 0, 0) == -1) + { + asio::error_code error(errno, + asio::error::get_system_category()); + asio::detail::throw_error(error); + } } kqueue_reactor::~kqueue_reactor() @@ -88,26 +93,27 @@ interrupter_.recreate(); + struct kevent event; + ASIO_KQUEUE_EV_SET(&event, interrupter_.read_descriptor(), + EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, &interrupter_); + if (::kevent(kqueue_fd_, &event, 1, 0, 0, 0) == -1) + { + asio::error_code error(errno, + asio::error::get_system_category()); + asio::detail::throw_error(error); + } + // Re-register all descriptors with kqueue. mutex::scoped_lock descriptors_lock(registered_descriptors_mutex_); for (descriptor_state* state = registered_descriptors_.first(); state != 0; state = state->next_) { struct kevent events[2]; - int num_events = 0; - - if (!state->op_queue_[read_op].empty()) - ASIO_KQUEUE_EV_SET(&events[num_events++], state->descriptor_, - EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, state); - else if (!state->op_queue_[except_op].empty()) - ASIO_KQUEUE_EV_SET(&events[num_events++], state->descriptor_, - EVFILT_READ, EV_ADD | EV_CLEAR, EV_OOBAND, 0, state); - - if (!state->op_queue_[write_op].empty()) - ASIO_KQUEUE_EV_SET(&events[num_events++], state->descriptor_, - EVFILT_WRITE, EV_ADD | EV_CLEAR, 0, 0, state); - - if (num_events && ::kevent(kqueue_fd_, events, num_events, 0, 0, 0) == -1) + ASIO_KQUEUE_EV_SET(&events[0], state->descriptor_, + EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, state); + ASIO_KQUEUE_EV_SET(&events[1], state->descriptor_, + EVFILT_WRITE, EV_ADD | EV_CLEAR, 0, 0, state); + if (::kevent(kqueue_fd_, events, 2, 0, 0, 0) == -1) { asio::error_code error(errno, asio::error::get_system_category()); @@ -132,6 +138,14 @@ descriptor_data->descriptor_ = descriptor; descriptor_data->shutdown_ = false; + struct kevent events[2]; + ASIO_KQUEUE_EV_SET(&events[0], descriptor, EVFILT_READ, + EV_ADD | EV_CLEAR, 0, 0, descriptor_data); + ASIO_KQUEUE_EV_SET(&events[1], descriptor, EVFILT_WRITE, + EV_ADD | EV_CLEAR, 0, 0, descriptor_data); + if (::kevent(kqueue_fd_, events, 2, 0, 0, 0) == -1) + return errno; + return 0; } @@ -147,23 +161,13 @@ descriptor_data->shutdown_ = false; descriptor_data->op_queue_[op_type].push(op); - struct kevent event; - switch (op_type) - { - case read_op: - ASIO_KQUEUE_EV_SET(&event, descriptor, EVFILT_READ, - EV_ADD | EV_CLEAR, 0, 0, descriptor_data); - break; - case write_op: - ASIO_KQUEUE_EV_SET(&event, descriptor, EVFILT_WRITE, - EV_ADD | EV_CLEAR, 0, 0, descriptor_data); - break; - case except_op: - ASIO_KQUEUE_EV_SET(&event, descriptor, EVFILT_READ, - EV_ADD | EV_CLEAR, EV_OOBAND, 0, descriptor_data); - break; - } - ::kevent(kqueue_fd_, &event, 1, 0, 0, 0); + struct kevent events[2]; + ASIO_KQUEUE_EV_SET(&events[0], descriptor, EVFILT_READ, + EV_ADD | EV_CLEAR, 0, 0, descriptor_data); + ASIO_KQUEUE_EV_SET(&events[1], descriptor, EVFILT_WRITE, + EV_ADD | EV_CLEAR, 0, 0, descriptor_data); + if (::kevent(kqueue_fd_, events, 2, 0, 0, 0) == -1) + return errno; return 0; } @@ -176,7 +180,7 @@ source_descriptor_data = 0; } -void kqueue_reactor::start_op(int op_type, socket_type descriptor, +void kqueue_reactor::start_op(int op_type, socket_type /*descriptor*/, kqueue_reactor::per_descriptor_data& descriptor_data, reactor_op* op, bool is_continuation, bool allow_speculative) { @@ -214,36 +218,6 @@ descriptor_data->op_queue_[op_type].push(op); io_service_.work_started(); - - if (first) - { - struct kevent event; - switch (op_type) - { - case read_op: - ASIO_KQUEUE_EV_SET(&event, descriptor, EVFILT_READ, - EV_ADD | EV_CLEAR, 0, 0, descriptor_data); - break; - case write_op: - ASIO_KQUEUE_EV_SET(&event, descriptor, EVFILT_WRITE, - EV_ADD | EV_CLEAR, 0, 0, descriptor_data); - break; - case except_op: - if (!descriptor_data->op_queue_[read_op].empty()) - return; // Already registered for read events. - ASIO_KQUEUE_EV_SET(&event, descriptor, EVFILT_READ, - EV_ADD | EV_CLEAR, EV_OOBAND, 0, descriptor_data); - break; - } - - if (::kevent(kqueue_fd_, &event, 1, 0, 0, 0) == -1) - { - op->ec_ = asio::error_code(errno, - asio::error::get_system_category()); - descriptor_data->op_queue_[op_type].pop(); - io_service_.post_deferred_completion(op); - } - } } void kqueue_reactor::cancel_ops(socket_type, @@ -366,12 +340,10 @@ // Dispatch the waiting events. for (int i = 0; i < num_events; ++i) { - int descriptor = static_cast<int>(events[i].ident); void* ptr = reinterpret_cast<void*>(events[i].udata); if (ptr == &interrupter_) { - // No need to reset the interrupter since we're leaving the descriptor - // in a ready-to-read state and relying on edge-triggered notifications. + interrupter_.reset(); } else { @@ -413,45 +385,6 @@ } } } - - // Renew registration for event notifications. - struct kevent event; - switch (events[i].filter) - { - case EVFILT_READ: - if (!descriptor_data->op_queue_[read_op].empty()) - ASIO_KQUEUE_EV_SET(&event, descriptor, EVFILT_READ, - EV_ADD | EV_CLEAR, 0, 0, descriptor_data); - else if (!descriptor_data->op_queue_[except_op].empty()) - ASIO_KQUEUE_EV_SET(&event, descriptor, EVFILT_READ, - EV_ADD | EV_CLEAR, EV_OOBAND, 0, descriptor_data); - else - continue; - break; - case EVFILT_WRITE: - if (!descriptor_data->op_queue_[write_op].empty()) - ASIO_KQUEUE_EV_SET(&event, descriptor, EVFILT_WRITE, - EV_ADD | EV_CLEAR, 0, 0, descriptor_data); - else - continue; - break; - default: - break; - } - if (::kevent(kqueue_fd_, &event, 1, 0, 0, 0) == -1) - { - asio::error_code error(errno, - asio::error::get_system_category()); - 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); - } - } - } } } @@ -461,10 +394,7 @@ void kqueue_reactor::interrupt() { - struct kevent event; - ASIO_KQUEUE_EV_SET(&event, interrupter_.read_descriptor(), - EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, &interrupter_); - ::kevent(kqueue_fd_, &event, 1, 0, 0, 0); + interrupter_.interrupt(); } int kqueue_reactor::do_kqueue_create()