WIP: add MT support to Crystal::EventLoop::IoUring - part 2 [skip ci]

Now with timers and a bunch of fixes to make it compile.

Author: ysbaddaden

src/crystal/event_loop/io_uring.cr

@@ -1,3 +1,5 @@
+require "crystal/pointer_pairing_heap"
+
 struct Crystal::EventLoop::IoUring::Event
   enum Type
     Async
@@ -12,6 +14,10 @@ struct Crystal::EventLoop::IoUring::Event
   # property! flags : UInt32
 
   getter? timeout : Time::Span?
+  property! wake_at : Time::Span?
+
+  # The event can be added to the `Timers` list.
+  include PointerPairingHeap::Node
 
   # When using SQPOLL (or when IORING_FEAT_SUBMIT_STABLE is missing) the pointer
   # to the timespec struct must be reachable until the SQE has been successfully
@@ -33,4 +39,8 @@ struct Crystal::EventLoop::IoUring::Event
   protected def timespec : Pointer(LibC::Timespec)
     pointerof(@timespec)
   end
+
+  def heap_compare(other : Pointer(self)) : Bool
+    wake_at < other.value.wake_at
+  end
 end

src/crystal/event_loop/io_uring/event.cr

@@ -7,13 +7,14 @@ class Crystal::EventLoop::IoUring::FiberEvent
   end
 
   def add(timeout : Time::Span) : Nil
-    @event.timeout = timeout
+    s, ns = System::Time.monotonic
+    @event.wake_at = Time::Span.new(seconds: s, nanoseconds: ns) + timeout
     EventLoop.current.as(IoUring).add_timer(pointerof(@event))
   end
 
   # select timeout has been cancelled
   def delete : Nil
-    return unless @event.timeout?
+    return unless @event.wake_at?
     EventLoop.current.as(IoUring).delete_timer(pointerof(@event))
     clear
   end
@@ -25,6 +26,6 @@ class Crystal::EventLoop::IoUring::FiberEvent
 
   # the timer triggered (already dequeued from eventloop)
   def clear : Nil
-    @event.timeout = nil
+    @event.wake_at = nil
   end
 end

src/crystal/event_loop/io_uring/fiber_event.cr

@@ -1,25 +1,24 @@
 require "../../system/unix/io_uring"
 
 class Crystal::EventLoop::IoUring < Crystal::EventLoop
+  # Extends the system abstraction with additional data and helpers tailored
+  # for the event loop implementation.
   class Ring < System::IoUring
-    @tick = 0_u32
+    # TODO: not needed after <https://github.com/crystal-lang/crystal/issues/16157>
     @rng = Random::PCG32.new
+
     @sq_lock = Thread::Mutex.new
     @cq_lock = Thread::Mutex.new
-
     getter? waiting : Bool = false
 
-    def once_in_a_while? : Bool
-      (@tick &+= 1) == 53
-    end
-
     def waiting(&)
       @waiting = true
       yield
     ensure
       @waiting = false
     end
 
+    # Acquires the SQ lock for the duration of the block.
     def sq_lock(&)
       {% if flag?(:execution_context) %}
         @sq_lock.synchronize { yield }
@@ -28,6 +27,7 @@ class Crystal::EventLoop::IoUring < Crystal::EventLoop
       {% end %}
     end
 
+    # Acquires the CQ lock for the duration of the block.
     def cq_lock(&)
       {% if flag?(:execution_context) %}
         @cq_lock.synchronize { yield }
@@ -36,6 +36,8 @@ class Crystal::EventLoop::IoUring < Crystal::EventLoop
       {% end %}
     end
 
+    # Tries to acquire the CQ lock for the duration of the block. Returns
+    # immediately if the CQ lock couldn't be acquired.
     def cq_trylock?(&)
       {% if flag?(:execution_context) %}
         if @cq_lock.try_lock
@@ -50,6 +52,7 @@ class Crystal::EventLoop::IoUring < Crystal::EventLoop
       {% end %}
     end
 
+    # Locks the SQ ring, reserves exactly one SQE and submits before returning.
     def submit(&)
       sq_lock do
         sqe = next_sqe
@@ -60,6 +63,8 @@ class Crystal::EventLoop::IoUring < Crystal::EventLoop
       end
     end
 
+    # Locks the SQ ring, reserves as many SQE as needed to fill *sqes* and
+    # submits before returning.
     def submit(sqes, &)
       sq_lock do
         reserve(sqes.size)

src/crystal/event_loop/io_uring/ring.cr

@@ -18,13 +18,20 @@ struct Crystal::EventLoop::Timers(T)
     @heap.empty?
   end
 
-  # Returns the time of the next ready timer (if any).
+  # Returns the absolute time of the next ready timer (if any).
   def next_ready? : Time::Span?
     if event = @heap.first?
       event.value.wake_at
     end
   end
 
+  # Returns the relative time until the next ready timer (if any).
+  def next_ready_as_timeout? : Time::Span?
+    if abstime = next_ready?
+      abstime - now
+    end
+  end
+
   # Dequeues and yields each ready timer (their `#wake_at` is lower than
   # `System::Time.monotonic`) from the oldest to the most recent (i.e. time
   # ascending).

src/crystal/event_loop/timers.cr

@@ -59,7 +59,7 @@ class Crystal::System::IoUring
     (@@opcodes.not_nil![opcode].flags & LibC::IO_URING_OP_SUPPORTED) == LibC::IO_URING_OP_SUPPORTED
   end
 
-  @fd : Int32
+  getter fd : Int32
   @flags : UInt32
   @sq_size : UInt32
   @cq_size : UInt32
@@ -90,6 +90,10 @@ class Crystal::System::IoUring
       params.flags |= LibC::IORING_SETUP_ATTACH_WQ
     end
 
+    # The following is boilerplate code to map the rings from kernel land to
+    # user land, with some slight differences based on the running kernel
+    # supported features (for example single mmap, no SQ array).
+
     Crystal.trace :evloop, "io_uring_setup"
     @fd = Syscall.io_uring_setup(sq_entries.to_u32, pointerof(params))
     raise RuntimeError.from_os_error("io_uring_setup", Errno.new(-@fd)) if @fd < 0
@@ -278,17 +282,15 @@ class Crystal::System::IoUring
   # Call `io_uring_enter` syscall. Panics on EBADR (can't recover from lost
   # CQE), returns -EINTR or -EBUSY, and raises on other errnos, otherwise
   # returns the int returned by the syscall.
-  def enter(to_submit : UInt32 = 0, min_complete : UInt32 = 0, flags : UInt32 = 0, timeout : Time::Span?) : Int32
+  def enter(to_submit : Int = 0, min_complete : Int = 0, flags : UInt32 = 0, timeout : ::Time::Span? = nil) : Int32
     if timeout
       flags |= LibC::IORING_ENTER_EXT_ARG
 
       ts = uninitialized LibC::Timespec
       ts.tv_sec = typeof(ts.tv_sec).new(timeout.@seconds)
       ts.tv_nsec = typeof(ts.tv_nsec).new(timeout.@nanoseconds)
 
-      args = LibC::IoUringGetEventsArg.new(
-        ts : pointerof(ts).address.to_u64!
-      )
+      args = LibC::IoUringGetEventsArg.new(ts: pointerof(ts).address.to_u64!)
       arg = pointerof(args).as(Void*)
       argsz = LibC::SizeT.new(sizeof(LibC::IoUringGetEventsArg))
     else
@@ -303,7 +305,7 @@ class Crystal::System::IoUring
       timeout: timeout,
       flags: ENTERS.new(flags).to_s
 
-    ret = Syscall.io_uring_enter(@fd, to_submit, min_complete, flags, arg, argsz)
+    ret = Syscall.io_uring_enter(@fd, to_submit.to_u32, min_complete.to_u32, flags, arg, argsz)
     return ret if ret >= 0
 
     case ret

src/crystal/system/unix/io_uring.cr

@@ -145,7 +145,7 @@ module Fiber::ExecutionContext
     # use an atomic/fence to make sure that @size can only be incremented
     # *after* the value has been written to @buffer.
     private def start_schedulers(capacity)
-      capacity.times { |index| @schedulers << start_scheduler }
+      capacity.times { |index| @schedulers << start_scheduler(index) }
     end
 
     private def start_scheduler(index)

src/fiber/execution_context/parallel.cr

@@ -6,6 +6,11 @@ module Fiber::ExecutionContext
       Thread.current.scheduler
     end
 
+    @[AlwaysInline]
+    def self.current? : Scheduler?
+      Thread.current.scheduler?
+    end
+
     protected abstract def thread : Thread
     protected abstract def execution_context : ExecutionContext