Add EventLoop#run(blocking) and EventLoop#interrupt

src/crystal/scheduler.cr

@@ -149,7 +149,7 @@ class Crystal::Scheduler
         resume(runnable) unless runnable == @thread.current_fiber
         break
       else
-        @event_loop.run_once
+        @event_loop.run(blocking: true)
       end
     end
   end

src/crystal/system/event_loop.cr

@@ -7,8 +7,28 @@ abstract class Crystal::EventLoop
     Crystal::Scheduler.event_loop
   end
 
-  # Runs the event loop.
-  abstract def run_once : Nil
+  # Runs the loop.
+  #
+  # Returns immediately if events are activable. Set `blocking` to false to
+  # return immediately if there are no activable events; set it to true to wait
+  # for activable events, which will block the current thread until then.
+  #
+  # Returns `true` on normal returns (e.g. has activated events, has pending
+  # events but blocking was false) and `false` when there are no registered
+  # events.
+  abstract def run(blocking : Bool) : Bool
+
+  # Tells a blocking run loop to no longer wait for events to activate. It may
+  # for example enqueue a NOOP event with an immediate (or past) timeout. Having
+  # activated an event, the loop shall return, allowing the blocked thread to
+  # continue.
+  #
+  # Should be a NOOP when the loop isn't running or is running in a nonblocking
+  # mode.
+  #
+  # NOTE: we assume that multiple threads won't run the event loop at the same
+  #       time in parallel, but this assumption may change in the future!
+  abstract def interrupt : Nil
 
   # Create a new resume event for a fiber.
   abstract def create_resume_event(fiber : Fiber) : Event

src/crystal/system/unix/event_libevent.cr

@@ -59,18 +59,23 @@ module Crystal::LibEvent
         Crystal::LibEvent::Event.new(event)
       end
 
-      def run_loop : Nil
-        LibEvent2.event_base_loop(@base, LibEvent2::EventLoopFlags::None)
-      end
-
-      def run_once : Nil
-        LibEvent2.event_base_loop(@base, LibEvent2::EventLoopFlags::Once)
+      # NOTE: may return `true` even if no event has been triggered (e.g.
+      #       nonblocking), but `false` means that nothing was processed.
+      def loop(once : Bool, nonblock : Bool) : Bool
+        flags = LibEvent2::EventLoopFlags::None
+        flags |= LibEvent2::EventLoopFlags::Once if once
+        flags |= LibEvent2::EventLoopFlags::NonBlock if nonblock
+        LibEvent2.event_base_loop(@base, flags) == 0
       end
 
       def loop_break : Nil
         LibEvent2.event_base_loopbreak(@base)
       end
 
+      def loop_exit : Nil
+        LibEvent2.event_base_loopexit(@base, nil)
+      end
+
       def new_dns_base(init = true)
         DnsBase.new LibEvent2.evdns_base_new(@base, init ? 1 : 0)
       end

src/crystal/system/unix/event_loop_libevent.cr

@@ -18,9 +18,12 @@ class Crystal::LibEvent::EventLoop < Crystal::EventLoop
     end
   {% end %}
 
-  # Runs the event loop.
-  def run_once : Nil
-    event_base.run_once
+  def run(blocking : Bool) : Bool
+    event_base.loop(once: true, nonblock: !blocking)
+  end
+
+  def interrupt : Nil
+    event_base.loop_exit
   end
 
   # Create a new resume event for a fiber.

src/crystal/system/unix/lib_event2.cr

@@ -47,6 +47,7 @@ lib LibEvent2
   fun event_base_dispatch(eb : EventBase) : Int
   fun event_base_loop(eb : EventBase, flags : EventLoopFlags) : Int
   fun event_base_loopbreak(eb : EventBase) : Int
+  fun event_base_loopexit(EventBase, LibC::Timeval*) : LibC::Int
   fun event_set_log_callback(callback : (Int, UInt8*) -> Nil)
   fun event_enable_debug_mode
   fun event_reinit(eb : EventBase) : Int

src/crystal/system/wasi/event_loop.cr

@@ -8,8 +8,12 @@ end
 # :nodoc:
 class Crystal::Wasi::EventLoop < Crystal::EventLoop
   # Runs the event loop.
-  def run_once : Nil
-    raise NotImplementedError.new("Crystal::Wasi::EventLoop.run_once")
+  def run(blocking : Bool) : Bool
+    raise NotImplementedError.new("Crystal::Wasi::EventLoop.run")
+  end
+
+  def interrupt : Nil
+    raise NotImplementedError.new("Crystal::Wasi::EventLoop.interrupt")
   end
 
   # Create a new resume event for a fiber.

src/crystal/system/win32/event_loop_iocp.cr

@@ -13,6 +13,10 @@ class Crystal::Iocp::EventLoop < Crystal::EventLoop
   # This is a list of resume and timeout events managed outside of IOCP.
   @queue = Deque(Crystal::Iocp::Event).new
 
+  @lock = Crystal::SpinLock.new
+  @interrupted = Atomic(Bool).new(false)
+  @blocked_thread = Atomic(Thread?).new(nil)
+
   # Returns the base IO Completion Port
   getter iocp : LibC::HANDLE do
     create_completion_port(LibC::INVALID_HANDLE_VALUE, nil)
@@ -37,33 +41,49 @@ class Crystal::Iocp::EventLoop < Crystal::EventLoop
 
   # Runs the event loop and enqueues the fiber for the next upcoming event or
   # completion.
-  def run_once : Nil
+  def run(blocking : Bool) : Bool
     # Pull the next upcoming event from the event queue. This determines the
     # timeout for waiting on the completion port.
     # OPTIMIZE: Implement @queue as a priority queue in order to avoid this
     # explicit search for the lowest value and dequeue more efficient.
     next_event = @queue.min_by?(&.wake_at)
 
-    unless next_event
-      Crystal::System.print_error "Warning: No runnables in scheduler. Exiting program.\n"
-      ::exit
-    end
+    # no registered events: nothing to wait for
+    return false unless next_event
 
     now = Time.monotonic
 
     if next_event.wake_at > now
-      wait_time = next_event.wake_at - now
-      # There is no event ready to wake. So we wait for completions with a
-      # timeout for the next event wake time.
+      # There is no event ready to wake. We wait for completions until the next
+      # event wake time, unless nonblocking or already interrupted (timeout
+      # immediately).
+      if blocking
+        @lock.sync do
+          if @interrupted.get(:acquire)
+            blocking = false
+          else
+            # memorize the blocked thread (so we can alert it)
+            @blocked_thread.set(Thread.current, :release)
+          end
+        end
+      end
 
-      timed_out = IO::Overlapped.wait_queued_completions(wait_time.total_milliseconds) do |fiber|
+      wait_time = blocking ? (next_event.wake_at - now).total_milliseconds : 0
+      timed_out = IO::Overlapped.wait_queued_completions(wait_time, alertable: blocking) do |fiber|
         # This block may run multiple times. Every single fiber gets enqueued.
         fiber.enqueue
       end
 
-      # If the wait for completion timed out we've reached the wake time and
-      # continue with waking `next_event`.
-      return unless timed_out
+      @blocked_thread.set(nil, :release)
+      @interrupted.set(false, :release)
+
+      # The wait for completion enqueued events.
+      return true unless timed_out
+
+      # Wait for completion timed out but it may have been interrupted or we ask
+      # for immediate timeout (nonblocking), so we check for the next event
+      # readyness again:
+      return false if next_event.wake_at > Time.monotonic
     end
 
     # next_event gets activated because its wake time is passed, either from the
@@ -81,7 +101,7 @@ class Crystal::Iocp::EventLoop < Crystal::EventLoop
     # This would avoid the scheduler needing to looking at runnable again just
     # to notice it's still empty. The lock involved there should typically be
     # uncontested though, so it's probably not a big deal.
-    return if fiber.dead?
+    return false if fiber.dead?
 
     # A timeout event needs special handling because it does not necessarily
     # means to resume the fiber directly, in case a different select branch
@@ -92,6 +112,22 @@ class Crystal::Iocp::EventLoop < Crystal::EventLoop
     else
       fiber.enqueue
     end
+
+    # We enqueued a fiber.
+    true
+  end
+
+  def interrupt : Nil
+    thread = nil
+
+    @lock.sync do
+      @interrupted.set(true)
+      thread = @blocked_thread.swap(nil, :acquire)
+    end
+    return unless thread
+
+    # alert the thread to interrupt GetQueuedCompletionStatusEx
+    LibC.QueueUserAPC(->(ptr : LibC::ULONG_PTR) {}, thread, LibC::ULONG_PTR.new(0))
   end
 
   def enqueue(event : Crystal::Iocp::Event)

src/io/overlapped.cr

@@ -8,19 +8,21 @@ module IO::Overlapped
     property fiber : Fiber?
   end
 
-  def self.wait_queued_completions(timeout, &)
+  def self.wait_queued_completions(timeout, alertable = false, &)
     overlapped_entries = uninitialized LibC::OVERLAPPED_ENTRY[1]
 
     if timeout > UInt64::MAX
       timeout = LibC::INFINITE
     else
       timeout = timeout.to_u64
     end
-    result = LibC.GetQueuedCompletionStatusEx(Crystal::EventLoop.current.iocp, overlapped_entries, overlapped_entries.size, out removed, timeout, false)
+    result = LibC.GetQueuedCompletionStatusEx(Crystal::EventLoop.current.iocp, overlapped_entries, overlapped_entries.size, out removed, timeout, alertable)
     if result == 0
       error = WinError.value
       if timeout && error.wait_timeout?
         return true
+      elsif alertable && error.value == LibC::WAIT_IO_COMPLETION
+        return true
       else
         raise IO::Error.from_os_error("GetQueuedCompletionStatusEx", error)
       end

src/lib_c/x86_64-windows-msvc/c/processthreadsapi.cr

@@ -57,6 +57,7 @@ lib LibC
   fun GetProcessTimes(hProcess : HANDLE, lpCreationTime : FILETIME*, lpExitTime : FILETIME*,
                       lpKernelTime : FILETIME*, lpUserTime : FILETIME*) : BOOL
   fun SwitchToThread : BOOL
+  fun QueueUserAPC(pfnAPC : PAPCFUNC, hThread : HANDLE, dwData : ULONG_PTR) : DWORD
 
   PROCESS_QUERY_INFORMATION = 0x0400
 end

src/lib_c/x86_64-windows-msvc/c/winnt.cr

@@ -14,6 +14,8 @@ lib LibC
   alias HANDLE = Void*
   alias HMODULE = Void*
 
+  alias PAPCFUNC = ULONG_PTR ->
+
   INVALID_FILE_ATTRIBUTES      = DWORD.new!(-1)
   FILE_ATTRIBUTE_DIRECTORY     =  0x10
   FILE_ATTRIBUTE_HIDDEN        =   0x2