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

.github/workflows/linux.yml

@@ -82,6 +82,27 @@ jobs:
       - name: Test
         run: bin/ci with_build_env 'CRYSTAL_WORKERS=4 make std_spec threads=1 FLAGS="-D preview_mt"'
 
+  test_force_iouring:
+    env:
+      ARCH: x86_64
+      ARCH_CMD: linux64
+    runs-on: ubuntu-latest
+    steps:
+      - name: Download Crystal source
+        uses: actions/checkout@v2
+
+      - name: Prepare System
+        run: bin/ci prepare_system
+
+      - name: Prepare Build
+        run: bin/ci prepare_build
+
+      - name: Make Crystal
+        run: bin/ci with_build_env 'make crystal'
+
+      - name: Test
+        run: bin/ci with_build_env 'make std_spec threads=1 FLAGS="-D force_iouring"'
+
   check_format:
     env:
       ARCH: x86_64

spec/std/syscall_spec.cr

@@ -0,0 +1,27 @@
+{% skip_file unless flag?(:linux) %}
+
+require "spec"
+require "syscall"
+
+private Syscall.def_syscall pipe2, Int32, pipefd : Int32[2]*, flags : Int32
+private Syscall.def_syscall write, Int32, fd : Int32, buf : UInt8*, count : LibC::SizeT
+private Syscall.def_syscall read, Int32, fd : Int32, buf : UInt8*, count : LibC::SizeT
+private Syscall.def_syscall close, Int32, fd : Int32
+
+describe Syscall do
+  it "can call into the system successfully" do
+    pair = uninitialized Int32[2]
+    pipe2(pointerof(pair), 0).should eq(0)
+
+    str = "Hello"
+    write(pair[1], str.to_unsafe, LibC::SizeT.new(str.bytesize)).should eq(str.bytesize)
+
+    buf = Bytes.new(64)
+    read(pair[0], buf.to_unsafe, LibC::SizeT.new(buf.size)).should eq(str.bytesize)
+
+    String.new(buf.to_unsafe, str.bytesize).should eq(str)
+
+    close(pair[0]).should eq(0)
+    close(pair[1]).should eq(0)
+  end
+end

src/crystal/scheduler.cr

@@ -12,6 +12,12 @@ 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
+
+  def self.event_loop : Crystal::EventLoop
+    Thread.current.scheduler.@event_loop
+  end
+
   def self.current_fiber : Fiber
     Thread.current.scheduler.@current
   end
@@ -148,7 +154,7 @@ class Crystal::Scheduler
         end
         break
       else
-        Crystal::EventLoop.run_once
+        @event_loop.run_once
       end
     end
 

src/crystal/system/event_loop.cr

@@ -1,35 +1,41 @@
-module Crystal::EventLoop
+abstract class Crystal::EventLoop
+  # Creates an event loop instance
+  # def self.create : Crystal::EventLoop
+
   # Runs the event loop.
-  # def self.run_once : Nil
+  abstract def run_once : Nil
 
   {% unless flag?(:preview_mt) %}
     # Reinitializes the event loop after a fork.
-    # def self.after_fork : Nil
+    abstract def after_fork : Nil
   {% end %}
 
   # Create a new resume event for a fiber.
-  # def self.create_resume_event(fiber : Fiber) : Crystal::Event
+  abstract def create_resume_event(fiber : Fiber) : Crystal::Event
 
   # Creates a timeout_event.
-  # def self.create_timeout_event(fiber) : Crystal::Event
+  abstract def create_timeout_event(fiber : Fiber) : Crystal::Event
 
   # Creates a write event for a file descriptor.
-  # def self.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) : Crystal::Event
 
   # Creates a read event for a file descriptor.
-  # def self.create_fd_read_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::Event
+  abstract def create_fd_read_event(io : IO::Evented, edge_triggered : Bool = false) : Crystal::Event
 end
 
-struct Crystal::Event
+# :nodoc:
+abstract struct Crystal::Event
   # Frees the event.
-  # def free : Nil
+  abstract def free : Nil
 
   # Adds a new timeout to this event.
-  # def add(time_span : Time::Span?) : Nil
+  abstract def add(time_span : Time::Span?) : Nil
 end
 
 {% if flag?(:unix) %}
+  require "./unix/event_loop_io_uring"
   require "./unix/event_loop_libevent"
+  require "./unix/event_loop"
 {% elsif flag?(:win32) %}
   require "./win32/event_loop_iocp"
 {% else %}

src/crystal/system/unix/event_io_uring.cr

@@ -0,0 +1,43 @@
+{% skip_file unless flag?(:linux) %}
+
+require "./io_uring"
+
+# :nodoc:
+struct Crystal::IoUringEvent < Crystal::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)
+  end
+
+  def free : Nil
+  end
+
+  def delete : Nil
+    if @type.timeout?
+      @io_uring.timeout_remove(@callback)
+    end
+  end
+
+  def add(time_span : Time::Span?) : Nil
+    time_span = nil if time_span == Time::Span::ZERO
+
+    case @type
+    when .resume?, .timeout?
+      if time_span
+        @io_uring.timeout(time_span, @callback)
+      else
+        @io_uring.nop(@callback)
+      end
+    when .readable_fd?
+      @io_uring.wait_readable(@fd, @callback, timeout: time_span)
+    when .writable_fd?
+      @io_uring.wait_writable(@fd, @callback, timeout: time_span)
+    end
+  end
+end

src/crystal/system/unix/event_libevent.cr

@@ -1,11 +1,13 @@
+{% skip_file if flag?(:linux) && flag?(:force_iouring) %}
+
 require "./lib_event2"
 
 {% if flag?(:preview_mt) %}
   LibEvent2.evthread_use_pthreads
 {% end %}
 
 # :nodoc:
-struct Crystal::Event
+struct Crystal::LibEventEvent < Crystal::Event
   VERSION = String.new(LibEvent2.event_get_version)
 
   def self.callback(&block : Int32, LibEvent2::EventFlags, Void* ->)
@@ -16,23 +18,19 @@ struct Crystal::Event
     @freed = false
   end
 
-  def add(timeout : LibC::Timeval? = nil)
+  def add(timeout : Time::Span?) : Nil
     if timeout
-      timeout_copy = timeout
-      LibEvent2.event_add(@event, pointerof(timeout_copy))
+      timeval = LibC::Timeval.new(
+        tv_sec: LibC::TimeT.new(timeout.total_seconds),
+        tv_usec: timeout.nanoseconds // 1_000
+      )
+      LibEvent2.event_add(@event, pointerof(timeval))
     else
       LibEvent2.event_add(@event, nil)
     end
   end
 
-  def add(timeout : Time::Span)
-    add LibC::Timeval.new(
-      tv_sec: LibC::TimeT.new(timeout.total_seconds),
-      tv_usec: timeout.nanoseconds // 1_000
-    )
-  end
-
-  def free
+  def free : Nil
     LibEvent2.event_free(@event) unless @freed
     @freed = true
   end
@@ -49,26 +47,26 @@ struct Crystal::Event
       @base = LibEvent2.event_base_new
     end
 
-    def reinit
+    def reinit : Nil
       unless LibEvent2.event_reinit(@base) == 0
         raise "Error reinitializing libevent"
       end
     end
 
     def new_event(s : Int32, flags : LibEvent2::EventFlags, data, &callback : LibEvent2::Callback)
       event = LibEvent2.event_new(@base, s, flags, callback, data.as(Void*))
-      Event.new(event)
+      LibEventEvent.new(event)
     end
 
-    def run_loop
+    def run_loop : Nil
       LibEvent2.event_base_loop(@base, LibEvent2::EventLoopFlags::None)
     end
 
-    def run_once
+    def run_once : Nil
       LibEvent2.event_base_loop(@base, LibEvent2::EventLoopFlags::Once)
     end
 
-    def loop_break
+    def loop_break : Nil
       LibEvent2.event_base_loopbreak(@base)
     end
 

src/crystal/system/unix/event_loop.cr

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

src/crystal/system/unix/event_loop_io_uring.cr

@@ -0,0 +1,62 @@
+{% skip_file unless flag?(:linux) %}
+
+require "./event_io_uring"
+
+# :nodoc:
+class Crystal::IoUringEventLoop < Crystal::EventLoop
+  private getter(io_uring) { Crystal::System::IoUring.new(128) }
+
+  {% unless flag?(:preview_mt) %}
+    # Reinitializes the event loop after a fork.
+    def after_fork : Nil
+      @io_uring = nil
+    end
+  {% end %}
+
+  # Runs the event loop.
+  def run_once : Nil
+    io_uring.process_completion_events(blocking: true)
+  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|
+      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|
+      next if res == -Errno::ECANCELED.value
+      if (select_action = fiber.timeout_select_action)
+        fiber.timeout_select_action = nil
+        select_action.time_expired(fiber)
+      else
+        Crystal::Scheduler.enqueue fiber
+      end
+    end
+  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
+    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
+    end
+  end
+end

src/crystal/system/unix/event_loop_libevent.cr

@@ -1,37 +1,32 @@
-require "./event_libevent"
+{% skip_file if flag?(:linux) && flag?(:force_iouring) %}
 
-class Thread
-  # :nodoc:
-  getter(event_base) { Crystal::Event::Base.new }
-end
+require "./event_libevent"
 
 # :nodoc:
-module Crystal::EventLoop
+class Crystal::LibEventEventLoop < Crystal::EventLoop
+  private getter(event_base) { Crystal::LibEventEvent::Base.new }
+
   {% unless flag?(:preview_mt) %}
     # Reinitializes the event loop after a fork.
-    def self.after_fork : Nil
+    def after_fork : Nil
       event_base.reinit
     end
   {% end %}
 
   # Runs the event loop.
-  def self.run_once
+  def run_once : Nil
     event_base.run_once
   end
 
-  private def self.event_base
-    Thread.current.event_base
-  end
-
   # Create a new resume event for a fiber.
-  def self.create_resume_event(fiber : Fiber) : Crystal::Event
+  def create_resume_event(fiber : Fiber) : Crystal::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 self.create_timeout_event(fiber) : Crystal::Event
+  def create_timeout_event(fiber) : Crystal::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)
@@ -44,7 +39,7 @@ module Crystal::EventLoop
   end
 
   # Creates a write event for a file descriptor.
-  def self.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::Event
     flags = LibEvent2::EventFlags::Write
     flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET if edge_triggered
 
@@ -59,7 +54,7 @@ module Crystal::EventLoop
   end
 
   # Creates a read event for a file descriptor.
-  def self.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::Event
     flags = LibEvent2::EventFlags::Read
     flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET if edge_triggered