Support combine with other event loops

asio/include/asio/detail/impl/kqueue_reactor.ipp

@@ -444,8 +444,18 @@ void kqueue_reactor::run(long usec, op_queue<operation>& ops)
   lock.unlock();
 
   // Block on the kqueue descriptor.
-  struct kevent events[128];
-  int num_events = kevent(kqueue_fd_, 0, 0, events, 128, timeout);
+  struct kevent *events = event_loop_field_.events;
+  int &num_events = event_loop_field_.num_events;
+
+  if (!event_loop_field_.has_event) {
+    num_events = kevent(kqueue_fd_, 0, 0, events, 128, timeout);
+    if (event_loop_field_.is_wait) {
+      event_loop_field_.has_event = true;
+      return;
+    }
+  } else {
+    event_loop_field_.has_event = false;
+  }
 
 #if defined(ASIO_ENABLE_HANDLER_TRACKING)
   // Trace the waiting events.
@@ -542,6 +552,10 @@ void kqueue_reactor::run(long usec, op_queue<operation>& ops)
   timer_queues_.get_ready_timers(ops);
 }
 
+void kqueue_reactor::mask_wait_only() {
+  event_loop_field_.is_wait = true;
+}
+
 void kqueue_reactor::interrupt()
 {
   interrupter_.interrupt();

asio/include/asio/detail/impl/scheduler.ipp

@@ -230,6 +230,12 @@ std::size_t scheduler::run_one(asio::error_code& ec)
   return do_run_one(lock, this_thread, ec);
 }
 
+std::size_t scheduler::wait_event(long usec, asio::error_code &ec)
+{
+  task_->mask_wait_only();
+  return wait_one(usec, ec);
+}
+
 std::size_t scheduler::wait_one(long usec, asio::error_code& ec)
 {
   ec = asio::error_code();

asio/include/asio/detail/kqueue_reactor.hpp

@@ -208,6 +208,8 @@ class kqueue_reactor
   // Run the kqueue loop.
   ASIO_DECL void run(long usec, op_queue<operation>& ops);
 
+  ASIO_DECL void mask_wait_only();
+
   // Interrupt the kqueue loop.
   ASIO_DECL void interrupt();
 
@@ -254,6 +256,14 @@ class kqueue_reactor
 
   // Keep track of all registered descriptors.
   object_pool<descriptor_state> registered_descriptors_;
+
+  // Save fired events
+  struct {
+    struct kevent events[128];
+    int num_events;
+    std::atomic_bool has_event;
+    std::atomic_bool is_wait;
+  } event_loop_field_ {};
 };
 
 } // namespace detail

asio/include/asio/detail/scheduler.hpp

@@ -77,6 +77,9 @@ class scheduler
   // Poll for one operation without blocking.
   ASIO_DECL std::size_t poll_one(asio::error_code& ec);
 
+  // Wait until timeout, interrupted, or one operation event.
+  ASIO_DECL std::size_t wait_event(long usec, asio::error_code& ec);
+
   // Interrupt the event processing loop.
   ASIO_DECL void stop();
 

asio/include/asio/detail/scheduler_task.hpp

@@ -31,6 +31,11 @@ class scheduler_task
   // Run the task once until interrupted or events are ready to be dispatched.
   virtual void run(long usec, op_queue<scheduler_operation>& ops) = 0;
 
+  // Mask just wait event for this time call run
+  virtual void mask_wait_only() {
+    throw_error(std::make_error_code(std::errc::operation_not_supported));
+  };
+
   // Interrupt the task.
   virtual void interrupt() = 0;
 

asio/src/examples/cpp11/other_eventloop/with_libuv.cpp

@@ -0,0 +1,56 @@
+#include <asio.hpp>
+#include <iostream>
+#include <uv.h>
+
+//#define USE_LOG
+#ifdef USE_LOG
+#include "log.h"
+#else
+#define LOG printf
+#endif
+
+int main() {
+  uv_loop_t *loop = uv_default_loop();
+  static asio::io_context io_context;
+  asio::io_context::work work(io_context);
+
+  uv_timer_t timer_req;
+  uv_timer_init(loop, &timer_req);
+  uv_timer_start(
+      &timer_req, [](uv_timer_t *handle) { LOG("libuv Timer fired!\n"); },
+      10000, 10000);
+
+  asio::steady_timer asio_timer(io_context);
+  std::function<void()> start_asio_timer;
+  start_asio_timer = [&] {
+    asio_timer.expires_after(asio::chrono::milliseconds(1000));
+    asio_timer.async_wait([&](const asio::error_code &ec) {
+      LOG("asio Timer fired!\n");
+      start_asio_timer();
+    });
+  };
+  start_asio_timer();
+
+  static std::condition_variable async_done;
+  std::mutex async_done_lock;
+  uv_async_t async;
+  uv_async_init(loop, &async, [](uv_async_t *handle) {
+    io_context.poll_one();
+    async_done.notify_one();
+  });
+
+  std::thread([&] {
+    for (;;) {
+      LOG("wait_event...");
+      asio::error_code ec;
+      auto &service = asio::use_service<asio::detail::scheduler>(io_context);
+      service.wait_event(INTMAX_MAX, ec);
+      uv_async_send(&async);
+      std::unique_lock<std::mutex> lock(async_done_lock);
+      async_done.wait(lock);
+    }
+  }).detach();
+
+  uv_run(loop, UV_RUN_DEFAULT);
+  return 0;
+}