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

Author: ysbaddaden

src/crystal/event_loop/io_uring.cr

@@ -44,6 +44,9 @@ class Crystal::EventLoop::IoUring < Crystal::EventLoop
   # we require are only available in Linux 5.18. The event loop is thus
   # incompatible with:
   #
+  # FIXME: while IORING_OP_ASYNC_CANCEL was introduced in Linux 5.5, the
+  # IORING_ASYNC_CANCEL_FD flag was only introduced in Linux 5.19.
+  #
   # - Linux 5.4 LTS (EOL Dec 2025)
   # - Linux 5.10 SLTS (EOL Jan 2031)
   # - Linux 5.15 LTS (EOL Dec 2026)
@@ -70,10 +73,9 @@ class Crystal::EventLoop::IoUring < Crystal::EventLoop
       System::IoUring.supports_opcode?(LibC::IORING_OP_MSG_RING)
   end
 
-  DEFAULT_SQ_ENTRIES = 16
-  DEFAULT_CQ_ENTRIES = 128
-  # DEFAULT_SQ_THREAD_IDLE = 2000
-  DEFAULT_SQ_THREAD_IDLE = nil
+  DEFAULT_SQ_ENTRIES     =  16
+  DEFAULT_CQ_ENTRIES     = 128
+  DEFAULT_SQ_THREAD_IDLE = nil # 200 (in milliseconds)
 
   CLOSE_RING_EVENT = GC.malloc(sizeof(Event)).as(Event*)
 
@@ -377,13 +379,23 @@ class Crystal::EventLoop::IoUring < Crystal::EventLoop
     # search a waiting ring to wakeup
     return unless waiting_ring = @rings.find(&.try(&.waiting?))
 
-    # try to use the local ring and fallback to the actual ring (we might
-    # interrupt from a raw thread)
-    ring = ring? || waiting_ring
+    # try to notify the waiting ring through the local ring (every scheduler
+    # should have one), fallback to a syscall (Linux 6.13) or to a cross ring
+    # submit for older kernels
+    ring = ring?
+    ring = waiting_ring if ring.nil? && !System::IoUring.supports_register_send_msg_ring?
 
-    ring.submit do |sqe|
-      sqe.value.opcode = LibC::IORING_OP_MSG_RING
-      sqe.value.fd = waiting_ring.fd
+    if ring
+      ring.submit do |sqe|
+        sqe.value.opcode = LibC::IORING_OP_MSG_RING
+        sqe.value.fd = waiting_ring.fd
+      end
+    else
+      sqe = LibC::IoUringSqe.new
+      sqe.opcode = LibC::IORING_OP_MSG_RING
+      sqe.fd = waiting_ring.fd
+      ret = System::Syscall.io_uring_register(-1, LibC::IORING_REGISTER_SEND_MSG_RING, pointerof(sqe).as(Void*), 1)
+      raise RuntimeError.from_os_error("io_uring_register(IORING_REGISTER_SEND_MSG_RING)", Errno.new(-ret)) if ret < 0
     end
   end
 
@@ -487,9 +499,19 @@ class Crystal::EventLoop::IoUring < Crystal::EventLoop
     return unless fd = file_descriptor.close_volatile_fd?
 
     # FIXME: we must submit an IORING_OP_ASYNC_CANCEL to every ring (across
-    # execution contexts) that has a pending READ/WRITE/POLL operation on the
-    # IO::FileDescriptor; we can't just link CANCEL to CLOSE on the local
-    # ring...
+    # execution contexts) that has a pending IORING_OP_READ or IORING_OP_WRITE
+    # operation on the IO::FileDescriptor; we can't just link CANCEL to CLOSE on
+    # the local ring (requires SQ lock so any thread can submit to the ring).
+    #
+    # NOTE: IORING_OP_POLL_ADD operations don't need to be canceled (POLLERR
+    # and/or POLLHUP / POLLRDHUP are enough).
+    #
+    # TODO: consider IORING_REGISTER_SYNC_CANCEL (Linux 6.0) that is synchronous
+    # (bad) but allows to target any ring safely and to skip the SQ lock for
+    # _every_ operation (good).
+    #
+    # OPTIMIZE: if the pending ops are on this ring (lucky) we can keep
+    # submitting an IORING_OP_ASYNC_CANCEL linked to an IORING_OP_CLOSE.
 
     async_close(fd) do |sqe|
       # one thread closing a fd won't interrupt reads or writes happening in

src/crystal/event_loop/io_uring/ring.cr

@@ -4,27 +4,43 @@ 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
-    # 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 initialize
+      # TODO: not needed after <https://github.com/crystal-lang/crystal/issues/16157>
+      @rng = Random::PCG32.new
+
+      {% if flag?(:preview_mt) %}
+        # unless IORING_REGISTER_SYNC_CANCEL (Linux 6.0) and
+        # IORING_REGISTER_SEND_MSG_RING (Linux 6.13) are supported we may have
+        # multiple threads trying to submit to the ring (on evloop interrupt and
+        # before closing an IO::FileDescriptor)
+        unless System::IoUring.supports_register_sync_cancel? && System::IoUring.supports_register_send_msg_ring?
+          @sq_lock = Thread::Mutex.new
+        end
+      {% end %}
+
+      {% if flag?(:execution_context) %}
+        @cq_lock = Thread::Mutex.new
+      {% end %}
+    end
+
     def waiting(&)
       @waiting = true
       yield
     ensure
       @waiting = false
     end
 
-    # Acquires the SQ lock for the duration of the block.
+    # Acquires the SQ lock for the duration of the block if required.
     def sq_lock(&)
-      {% if flag?(:execution_context) %}
-        @sq_lock.synchronize { yield }
-      {% else %}
-        yield
+      {% if flag?(:preview_mt) %}
+        if sq_lock = @sq_lock
+          return sq_lock.synchronize { yield }
+        end
       {% end %}
+
+      yield
     end
 
     # Acquires the CQ lock for the duration of the block.

src/crystal/system/unix/io_uring.cr

@@ -1,6 +1,5 @@
 require "c/linux/io_uring"
 require "./syscall"
-require "./eventfd"
 
 # WARNING: while the syscalls are thread safe, the rings and the overall
 # abstraction are not: accesses to the SQ and CQ rings aren't synchronized!
@@ -14,6 +13,8 @@ class Crystal::System::IoUring
   @@opcodes : Slice(LibC::IoUringProbeOp)?
 
   class_getter?(supported : Bool) { check_kernel_support? }
+  class_getter? supports_register_send_msg_ring : Bool = false
+  class_getter? supports_register_sync_cancel : Bool = false
 
   def self.check_kernel_support? : Bool
     # try with "no sqarray" flag first (available since linux 6.6)
@@ -32,20 +33,35 @@ class Crystal::System::IoUring
       @@no_sqarray = false
     end
 
+    # record supported features
     @@features = params.features
 
     begin
+      # probe supported opcodes
       probe_sz = sizeof(LibC::IoUringProbe) + LibC::IORING_OP_LAST * sizeof(LibC::IoUringProbeOp)
-      probe = GC.malloc(probe_sz).as(LibC::IoUringProbe*)
+      probe = GC.malloc_atomic(probe_sz).as(LibC::IoUringProbe*)
       raise RuntimeError.from_errno("malloc") unless probe
       LibIntrinsics.memset(probe, 0_u8, probe_sz, false)
 
       ret = Syscall.io_uring_register(fd, LibC::IORING_REGISTER_PROBE, probe.as(Void*), LibC::IORING_OP_LAST)
       raise RuntimeError.from_os_error("io_uring_register", Errno.new(-ret)) if ret < 0
 
       @@opcodes = Slice(LibC::IoUringProbeOp).new(probe.value.ops.to_unsafe, LibC::IORING_OP_LAST)
+
+      # probe supported register opcodes (must test one by one)
+      sqe = LibC::IoUringSqe.new
+      sqe.opcode = LibC::IORING_OP_MSG_RING
+      sqe.fd = fd
+      ret = System::Syscall.io_uring_register(-1, LibC::IORING_REGISTER_SEND_MSG_RING, pointerof(sqe).as(Void*), 1)
+      @@supports_register_send_msg_ring = ret == 0
+
+      reg = LibC::IoUringSyncCancelReg.new
+      reg.flags = LibC::IORING_ASYNC_CANCEL_FD
+      reg.fd = 1
+      ret = System::Syscall.io_uring_register(fd, LibC::IORING_REGISTER_SYNC_CANCEL, pointerof(reg).as(Void*), 1)
+      @@supports_register_sync_cancel = ret != -LibC::ENOENT
     ensure
-      LibC.close(fd) # if fd > 0
+      LibC.close(fd)
     end
 
     true
@@ -55,6 +71,13 @@ class Crystal::System::IoUring
     (@@features.not_nil! & feature) == feature
   end
 
+  def self.supports_register_send_msg_ring? : Bool
+    sqe = LibC::IoUringSqe.new
+    sqe.opcode = LibC::IORING_OP_MSG_RING
+    sqe.fd = waiting_ring.fd
+    ret = System::Syscall.io_uring_register(-1, LibC::IORING_REGISTER_SEND_MSG_RING, pointerof(sqe).as(Void*), 1)
+  end
+
   def self.supports_opcode?(opcode : UInt32) : Bool
     (@@opcodes.not_nil![opcode].flags & LibC::IO_URING_OP_SUPPORTED) == LibC::IO_URING_OP_SUPPORTED
   end
@@ -192,7 +215,6 @@ class Crystal::System::IoUring
     LibC.munmap(@sqes, @sqes_size)
 
     LibC.close(fd)
-    @eventfd.try(&.close)
   end
 
   def sq_poll? : Bool
@@ -204,17 +226,18 @@ class Crystal::System::IoUring
     (sq_flags & LibC::IORING_SQ_NEED_WAKEUP) == LibC::IORING_SQ_NEED_WAKEUP
   end
 
-  # Call `io_uring_register` syscall, and raises on errno.
+  # Call `io_uring_register` syscall. Raises on error.
   def register(opcode : UInt32, arg : Pointer | Nil = nil, arg_sz = 0) : Nil
-    argp = arg ? arg.as(Void*) : Pointer(Void).null
-    err = Syscall.io_uring_register(@fd, opcode, argp, arg_sz.to_u32)
-    raise RuntimeError.from_os_error("io_uring_register", Errno.new(-err)) if err < 0
+    if errno = register?(opcode, arg, arg_sz)
+      raise RuntimeError.from_os_error("io_uring_register", errno)
+    end
   end
 
-  # Register an `EventFD`.
-  def register(@eventfd : EventFD) : Nil
-    efd = eventfd.fd
-    register(LibC::IORING_REGISTER_EVENTFD, pointerof(efd), 1)
+  # Call `io_uring_register` syscall. Returns Errno on error.
+  def register?(opcode : UInt32, arg : Pointer | Nil = nil, arg_sz = 0) : Errno?
+    argp = arg ? arg.as(Void*) : Pointer(Void).null
+    err = Syscall.io_uring_register(@fd, opcode, argp, arg_sz.to_u32)
+    Errno.new(-err) if err < 0
   end
 
   # Makes sure there is at least *count* SQE available in the SQ ring so we can

src/lib_c/x86_64-linux-gnu/c/linux/io_uring.cr

@@ -1,3 +1,5 @@
+require "../time"
+
 lib LibC
   # IORING_FILE_INDEX_ALLOC = ~0_u32
 
@@ -25,7 +27,8 @@ lib LibC
   # IORING_SETUP_DEFER_TASKRUN = 1_u32 << 13
   # IORING_SETUP_NO_MMAP = 	1_u32 << 14
   # IORING_SETUP_REGISTERED_FD_ONLY = 1_u32 << 15
-  IORING_SETUP_NO_SQARRAY = 1_u32 << 16
+  IORING_SETUP_NO_SQARRAY    = 1_u32 << 16
+  IORING_SETUP_HYBRID_IOPOLL = 1_u32 << 17
 
   # IORING_TIMEOUT_ABS = 1_u32 << 0
   # IORING_TIMEOUT_UPDATE = 1_u32 << 1
@@ -86,6 +89,7 @@ lib LibC
     futex_flags : UInt32
     install_fd_flags : UInt32
     nop_flags : UInt32
+    pipe_flags : UInt32
   end
 
   struct IoUringSqe
@@ -155,6 +159,16 @@ lib LibC
     ts : UInt64
   end
 
+  struct IoUringSyncCancelReg
+    addr : UInt64
+    fd : Int32
+    flags : UInt32
+    timeout : KernelTimespec
+    opcode : UInt8
+    pad : UInt8[7]
+    pad2 : UInt64[3]
+  end
+
   IORING_FEAT_SINGLE_MMAP     = 1_u32 << 0
   IORING_FEAT_NODROP          = 1_u32 << 1
   IORING_FEAT_SUBMIT_STABLE   = 1_u32 << 2
@@ -169,6 +183,10 @@ lib LibC
   IORING_FEAT_CQE_SKIP        = 1_u32 << 11
   IORING_FEAT_LINKED_FILE     = 1_u32 << 12
   IORING_FEAT_REG_REG_RING    = 1_u32 << 13
+  IORING_FEAT_RECVSEND_BUNDLE = 1_u32 << 14
+  IORING_FEAT_MIN_TIMEOUT     = 1_u32 << 15
+  IORING_FEAT_RW_ATTR         = 1_u32 << 16
+  IORING_FEAT_NO_IOWAIT       = 1_u32 << 17
 
   IORING_OP_NOP              =  0_u32
   IORING_OP_READV            =  1_u32
@@ -228,7 +246,12 @@ lib LibC
   IORING_OP_FTRUNCATE        = 55_u32
   IORING_OP_BIND             = 56_u32
   IORING_OP_LISTEN           = 57_u32
-  IORING_OP_LAST             = 58_u32
+  IORING_OP_RECV_ZC          = 59_u32
+  IORING_OP_EPOLL_WAIT       = 60_u32
+  IORING_OP_READV_FIXED      = 61_u32
+  IORING_OP_WRITEV_FIXED     = 62_u32
+  IORING_OP_PIPE             = 63_u32
+  IORING_OP_LAST             = 64_u32
 
   IORING_OFF_SQ_RING =          0_u64
   IORING_OFF_CQ_RING =  0x8000000_u64
@@ -266,7 +289,13 @@ lib LibC
   IORING_REGISTER_PBUF_STATUS         = 26_u32
   IORING_REGISTER_NAPI                = 27_u32
   IORING_UNREGISTER_NAPI              = 28_u32
-  IORING_REGISTER_LAST                = 29_u32
+  IORING_REGISTER_CLOCK               = 29_u32
+  IORING_REGISTER_CLONE_BUFFERS       = 30_u32
+  IORING_REGISTER_SEND_MSG_RING       = 31_u32
+  IORING_REGISTER_ZCRX_IFQ            = 32_u32
+  IORING_REGISTER_RESIZE_RINGS        = 33_u32
+  IORING_REGISTER_MEM_REGION          = 34_u32
+  IORING_REGISTER_LAST                = 35_u32
   IORING_REGISTER_USE_REGISTERED_RING = 1_u32 << 31
 
   IO_URING_OP_SUPPORTED = 1_u32

src/lib_c/x86_64-linux-gnu/c/time.cr

@@ -18,6 +18,11 @@ lib LibC
     tm_zone : Char*
   end
 
+  struct KernelTimespec
+    tv_sec : LongLong
+    tv_nsec : LongLong
+  end
+
   struct Timespec
     tv_sec : TimeT
     tv_nsec : Long