Linux's io_uring IO interface (2x performance vs libevent)

.github/workflows/linux.yml

@@ -113,7 +113,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Download Crystal source
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
 
       - name: Prepare System
         run: bin/ci prepare_system

src/crystal/scheduler.cr

@@ -12,7 +12,7 @@ require "crystal/system/thread"
 # Only the class methods are public and safe to use. Instance methods are
 # protected and must never be called directly.
 class Crystal::Scheduler
-  protected getter event_loop = Crystal::EventLoop.create
+  @event_loop = Crystal::EventLoop.create
 
   def self.event_loop
     Thread.current.scheduler.@event_loop

src/crystal/system/event_loop.cr

@@ -11,25 +11,24 @@ abstract class Crystal::EventLoop
   {% end %}
 
   # Create a new resume event for a fiber.
-  abstract def create_resume_event(fiber : Fiber) : Crystal::Event
+  abstract def create_resume_event(fiber : Fiber) : Event
 
   # Creates a timeout_event.
-  abstract def create_timeout_event(fiber : Fiber) : Crystal::Event
+  abstract def create_timeout_event(fiber : Fiber) : Event
 
   # Creates a write event for a file descriptor.
-  abstract def create_fd_write_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::Event
+  abstract def create_fd_write_event(io : IO::Evented, edge_triggered : Bool = false) : Event
 
   # Creates a read event for a file descriptor.
-  abstract def create_fd_read_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::Event
-end
+  abstract def create_fd_read_event(io : IO::Evented, edge_triggered : Bool = false) : Event
 
-# :nodoc:
-abstract struct Crystal::Event
-  # Frees the event.
-  abstract def free : Nil
+  abstract struct Event
+    # Frees the event.
+    abstract def free : Nil
 
-  # Adds a new timeout to this event.
-  abstract def add(timeout : Time::Span?) : Nil
+    # Adds a new timeout to this event.
+    abstract def add(timeout : Time::Span?) : Nil
+  end
 end
 
 {% if flag?(:wasi) %}

src/crystal/system/unix/event_io_uring.cr

@@ -1,43 +1,48 @@
-{% skip_file unless flag?(:linux) %}
-
 require "./io_uring"
 
 # :nodoc:
-struct Crystal::IoUringEvent < Crystal::Event
+struct Crystal::IoUring::Event < Crystal::EventLoop::Event
   enum Type
     Resume
     Timeout
     ReadableFd
     WritableFd
   end
 
-  def initialize(@io_uring : Crystal::System::IoUring, @type : Type, @fd : Int32, &callback : Int32 ->)
-    @callback = Box.box(callback)
+  def initialize(@io_uring : Crystal::System::IoUring, @type : Type, @fd : Int32 = -1, &@callback : Int32 ->)
+    @action_id = 0u64
   end
 
   def free : Nil
+    delete
   end
 
   def delete : Nil
+    return if @action_id == 0u64
+    @io_uring.delete_completion_action(@action_id)
     if @type.timeout?
-      @io_uring.timeout_remove(@callback)
+      @io_uring.submit_timeout_remove(@action_id)
     end
+    @action_id = 0u64
   end
 
   def add(timeout : Time::Span?) : Nil
+    delete
+    @action_id = @io_uring.register_completion_action(@callback)
+
     timeout = nil if timeout == Time::Span::ZERO
 
     case @type
     in .resume?, .timeout?
       if timeout
-        @io_uring.timeout(timeout, @callback)
+        @io_uring.submit_timeout(timeout, action_id: @action_id)
       else
-        @io_uring.nop(@callback)
+        @io_uring.submit_nop(action_id: @action_id)
       end
     in .readable_fd?
-      @io_uring.wait_readable(@fd, @callback, timeout: timeout)
+      @io_uring.submit_poll_add(@fd, Crystal::System::Syscall::POLLIN, action_id: @action_id, timeout: timeout)
     in .writable_fd?
-      @io_uring.wait_writable(@fd, @callback, timeout: timeout)
+      @io_uring.submit_poll_add(@fd, Crystal::System::Syscall::POLLOUT, action_id: @action_id, timeout: timeout)
     end
   end
 end

src/crystal/system/unix/event_libevent.cr

@@ -5,7 +5,7 @@ require "./lib_event2"
 {% end %}
 
 # :nodoc:
-struct Crystal::LibEventEvent < Crystal::Event
+struct Crystal::LibEvent::Event < Crystal::EventLoop::Event
   VERSION = String.new(LibEvent2.event_get_version)
 
   def self.callback(&block : Int32, LibEvent2::EventFlags, Void* ->)
@@ -53,7 +53,7 @@ struct Crystal::LibEventEvent < Crystal::Event
 
     def new_event(s : Int32, flags : LibEvent2::EventFlags, data, &callback : LibEvent2::Callback)
       event = LibEvent2.event_new(@base, s, flags, callback, data.as(Void*))
-      LibEventEvent.new(event)
+      Crystal::LibEvent::Event.new(event)
     end
 
     def run_loop : Nil

src/crystal/system/unix/event_loop.cr

@@ -7,18 +7,18 @@
   require "./event_loop_libevent"
 {% end %}
 
-class Crystal::EventLoop
+abstract class Crystal::EventLoop
   def self.create : Crystal::EventLoop
     {% if flag?(:linux) && !flag?(:interpreted) && flag?(:force_iouring) %}
-      return IoUringEventLoop.new
+      return Crystal::IoUring::EventLoop.new
     {% else %}
       {% if flag?(:linux) && !flag?(:interpreted) && flag?(:preview_iouring) %}
         if ENV["CRYSTAL_EVENTLOOP"]? == "io_uring" || Crystal::System::IoUring.available?
-          return IoUringEventLoop.new
+          return Crystal::IoUring::EventLoop.new
         end
       {% end %}
 
-      return LibEventEventLoop.new
+      return Crystal::LibEvent::EventLoop.new
     {% end %}
   end
 end

src/crystal/system/unix/event_loop_io_uring.cr

@@ -1,9 +1,8 @@
-{% skip_file unless flag?(:linux) %}
-
 require "./event_io_uring"
+require "weak_ref"
 
 # :nodoc:
-class Crystal::IoUringEventLoop < Crystal::EventLoop
+class Crystal::IoUring::EventLoop < Crystal::EventLoop
   private getter(io_uring) { Crystal::System::IoUring.new(128) }
 
   {% unless flag?(:preview_mt) %}
@@ -19,15 +18,15 @@ class Crystal::IoUringEventLoop < Crystal::EventLoop
   end
 
   # Create a new resume event for a fiber.
-  def create_resume_event(fiber : Fiber) : Crystal::Event
-    Crystal::IoUringEvent.new(io_uring, :resume, 0) do |res|
+  def create_resume_event(fiber : Fiber) : Crystal::EventLoop::Event
+    Crystal::IoUring::Event.new(io_uring, :resume) do |res|
       Crystal::Scheduler.enqueue fiber
     end
   end
 
   # Creates a timeout_event.
-  def create_timeout_event(fiber) : Crystal::Event
-    Crystal::IoUringEvent.new(io_uring, :timeout, 0) do |res|
+  def create_timeout_event(fiber) : Crystal::EventLoop::Event
+    Crystal::IoUring::Event.new(io_uring, :timeout) do |res|
       next if res == -Errno::ECANCELED.value
       if select_action = fiber.timeout_select_action
         fiber.timeout_select_action = nil
@@ -39,24 +38,18 @@ class Crystal::IoUringEventLoop < Crystal::EventLoop
   end
 
   # Creates a write event for a file descriptor.
-  def create_fd_write_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::Event
-    Crystal::IoUringEvent.new(io_uring, :writable_fd, io.fd) do |res|
-      if res == -Errno::ECANCELED.value
-        io.resume_write(timed_out: true)
-      else
-        io.resume_write
-      end
+  def create_fd_write_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::EventLoop::Event
+    io_ref = WeakRef.new(io)
+    Crystal::IoUring::Event.new(io_uring, :writable_fd, io.fd) do |res|
+      io_ref.value.try &.resume_write(timed_out: res == -Errno::ECANCELED.value)
     end
   end
 
   # Creates a read event for a file descriptor.
-  def create_fd_read_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::Event
-    Crystal::IoUringEvent.new(io_uring, :readable_fd, io.fd) do |res|
-      if res == -Errno::ECANCELED.value
-        io.resume_read(timed_out: true)
-      else
-        io.resume_read
-      end
+  def create_fd_read_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::EventLoop::Event
+    io_ref = WeakRef.new(io)
+    Crystal::IoUring::Event.new(io_uring, :readable_fd, io.fd) do |res|
+      io_ref.value.try &.resume_read(timed_out: res == -Errno::ECANCELED.value)
     end
   end
 end

src/crystal/system/unix/event_loop_libevent.cr

@@ -1,10 +1,8 @@
-{% skip_file if flag?(:linux) && !flag?(:interpreted) && flag?(:force_iouring) %}
-
 require "./event_libevent"
 
 # :nodoc:
-class Crystal::LibEventEventLoop < Crystal::EventLoop
-  private getter(event_base) { Crystal::LibEventEvent::Base.new }
+class Crystal::LibEvent::EventLoop < Crystal::EventLoop
+  private getter(event_base) { Crystal::LibEvent::Event::Base.new }
 
   {% unless flag?(:preview_mt) %}
     # Reinitializes the event loop after a fork.
@@ -19,14 +17,14 @@ class Crystal::LibEventEventLoop < Crystal::EventLoop
   end
 
   # Create a new resume event for a fiber.
-  def create_resume_event(fiber : Fiber) : Crystal::Event
+  def create_resume_event(fiber : Fiber) : Crystal::EventLoop::Event
     event_base.new_event(-1, LibEvent2::EventFlags::None, fiber) do |s, flags, data|
       Crystal::Scheduler.enqueue data.as(Fiber)
     end
   end
 
   # Creates a timeout_event.
-  def create_timeout_event(fiber) : Crystal::Event
+  def create_timeout_event(fiber) : Crystal::EventLoop::Event
     event_base.new_event(-1, LibEvent2::EventFlags::None, fiber) do |s, flags, data|
       f = data.as(Fiber)
       if (select_action = f.timeout_select_action)
@@ -39,7 +37,7 @@ class Crystal::LibEventEventLoop < Crystal::EventLoop
   end
 
   # Creates a write event for a file descriptor.
-  def create_fd_write_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::Event
+  def create_fd_write_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::EventLoop::Event
     flags = LibEvent2::EventFlags::Write
     flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET if edge_triggered
 
@@ -54,7 +52,7 @@ class Crystal::LibEventEventLoop < Crystal::EventLoop
   end
 
   # Creates a read event for a file descriptor.
-  def create_fd_read_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::Event
+  def create_fd_read_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::EventLoop::Event
     flags = LibEvent2::EventFlags::Read
     flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET if edge_triggered