EventLoop: direct epoll/kqueue integration

src/channel/select/timeout_action.cr

@@ -58,9 +58,11 @@ class Channel(T)
     end
 
     def time_expired(fiber : Fiber) : Nil
-      if @select_context.try &.try_trigger
-        fiber.enqueue
-      end
+      fiber.enqueue if time_expired?
+    end
+
+    def time_expired? : Bool
+      @select_context.try &.try_trigger || false
     end
   end
 end

src/crystal/arena.cr

@@ -0,0 +1,152 @@
+# OPTIMIZE: can the generation help to avoid the mutation lock (atomic)?
+# OPTIMIZE: consider a memory map (mmap, VirtualAlloc) with a maximum capacity
+class Crystal::Arena(T)
+  struct Allocation(T)
+    property generation = 0_u32
+    property? allocated = false
+    @object = uninitialized T
+
+    def pointer : Pointer(T)
+      pointerof(@object)
+    end
+
+    def free : Nil
+      @generation &+= 1_u32
+      @allocated = false
+      pointer.clear(1)
+    end
+  end
+
+  @buffer : Slice(Allocation(T))
+
+  def initialize
+    @lock = SpinLock.new
+    @buffer = Pointer(Allocation(T)).malloc(32).to_slice(32)
+  end
+
+  private def grow_buffer(capacity)
+    buffer = Pointer(Allocation(T)).malloc(capacity).to_slice(capacity)
+    buffer.to_unsafe.copy_from(@buffer.to_unsafe, @buffer.size)
+    @buffer = buffer
+  end
+
+  # Returns a pointer to the object allocated at *gen_idx* (generation index).
+  # Raises if the object isn't allocated.
+  # Raises if the generation has changed (i.e. the object has been freed then reallocated)
+  # Raises if *index* is negative.
+  def get(gen_idx : Int64) : Pointer(T)
+    index, generation = from_gen_index(gen_idx)
+
+    in_bounds!(index)
+    allocation = @buffer.to_unsafe + index
+
+    unless allocation.value.allocated?
+      raise RuntimeError.new("#{self.class.name}: object not allocated at index #{index}")
+    end
+
+    unless (actual = allocation.value.generation) == generation
+      raise RuntimeError.new("#{self.class.name}: object generation changed at index #{index} (#{generation} => #{actual})")
+    end
+
+    allocation.value.pointer
+  end
+
+  # Yields and allocates the object at *index* unless already allocated.
+  # Returns a pointer to the object at *index* and the generation index.
+  #
+  # There are no generational checks.
+  # Raises if *index* is negative.
+  def lazy_allocate(index : Int32, &) : {Pointer(T), Int64}
+    # fast-path: check if already allocated
+    if in_bounds?(index)
+      allocation = @buffer.to_unsafe + index
+
+      if allocation.value.allocated?
+        return {allocation.value.pointer, to_gen_index(index, allocation)}
+      end
+    end
+
+    # slow-path: allocate
+    @lock.sync do
+      if index >= @buffer.size
+        # slowest-path: grow the buffer
+        grow_buffer(Math.pw2ceil(Math.max(index, @buffer.size * 2)))
+      end
+
+      unsafe_allocate(index) do |pointer, gen_index|
+        yield pointer, gen_index
+      end
+    end
+  end
+
+  private def unsafe_allocate(index : Int32, &) : {Pointer(T), Int64}
+    allocation = @buffer.to_unsafe + index
+    pointer = allocation.value.pointer
+    gen_index = to_gen_index(index, allocation)
+
+    unless allocation.value.allocated?
+      allocation.value.allocated = true
+      yield pointer, gen_index
+    end
+
+    {pointer, gen_index}
+  end
+
+  # Yields the object allocated at *index* then releases it.
+  # Does nothing if the object wasn't allocated.
+  #
+  # Raises if *index* is negative.
+  def free(index : Int32, &) : Nil
+    return unless in_bounds?(index)
+
+    @lock.sync do
+      allocation = @buffer.to_unsafe + index
+      return unless allocation.value.allocated?
+
+      yield allocation.value.pointer
+      allocation.value.free
+    end
+  end
+
+  private def in_bounds?(index : Int32) : Bool
+    if index.negative?
+      raise ArgumentError.new("#{self.class.name}: negative index #{index}")
+    else
+      index < @buffer.size
+    end
+  end
+
+  private def in_bounds!(index : Int32) : Nil
+    if index.negative?
+      raise ArgumentError.new("#{self.class.name}: negative index #{index}")
+    elsif index >= @buffer.size
+      raise IndexError.new("#{self.class.name}: out of bounds index #{index}")
+    end
+  end
+
+  # Iterates all allocated objects, yields the actual index as well as the
+  # generation index.
+  def each(&) : Nil
+    ptr = @buffer.to_unsafe
+
+    @buffer.size.times do |index|
+      allocation = ptr + index
+
+      if allocation.value.allocated?
+        yield index, to_gen_index(index, allocation)
+      end
+    end
+  end
+
+  private def to_gen_index(index : Int32, allocation : Pointer(Allocation(T))) : Int64
+    to_gen_index(index, allocation.value.generation)
+  end
+
+  private def to_gen_index(index : Int32, generation : UInt32) : Int64
+    (index.to_i64! << 32) | generation.to_u64!
+  end
+
+  private def from_gen_index(gen_index : Int64) : {Int32, UInt32}
+    {(gen_index >> 32).to_i32!, gen_index.to_u32!}
+  end
+end

src/crystal/pointer_linked_list.cr

@@ -86,4 +86,8 @@ struct Crystal::PointerLinkedList(T)
     each { |node| yield node }
     @head = Pointer(T).null
   end
+
+  def clear : Nil
+    @head = Pointer(T).null
+  end
 end

src/crystal/spin_lock.cr

@@ -1,5 +1,5 @@
 # :nodoc:
-class Crystal::SpinLock
+struct Crystal::SpinLock
   private UNLOCKED = 0
   private LOCKED   = 1
 

src/crystal/system/event_loop.cr

@@ -4,7 +4,15 @@ abstract class Crystal::EventLoop
     {% if flag?(:wasi) %}
       Crystal::Wasi::EventLoop.new
     {% elsif flag?(:unix) %}
-      Crystal::LibEvent::EventLoop.new
+      {% if flag?(:evloop_libevent) %}
+        Crystal::LibEvent::EventLoop.new
+      {% elsif flag?(:bsd) || flag?(:darwin) %}
+        Crystal::Kqueue::EventLoop.new
+      {% elsif flag?(:linux) || flag?(:solaris) %}
+        Crystal::Epoll::EventLoop.new
+      {% else %}
+        Crystal::LibEvent::EventLoop.new
+      {% end %}
     {% elsif flag?(:win32) %}
       Crystal::IOCP::EventLoop.new
     {% else %}
@@ -73,7 +81,15 @@ end
 {% if flag?(:wasi) %}
   require "./wasi/event_loop"
 {% elsif flag?(:unix) %}
-  require "./unix/event_loop_libevent"
+  {% if flag?(:evloop_libevent) %}
+    require "./unix/event_loop_libevent"
+  {% elsif flag?(:bsd) || flag?(:darwin) %}
+    require "./unix/kqueue/event_loop"
+  {% elsif flag?(:linux) || flag?(:solaris) %}
+    require "./unix/epoll/event_loop"
+  {% else %}
+    require "./unix/event_loop_libevent"
+  {% end %}
 {% elsif flag?(:win32) %}
   require "./win32/event_loop_iocp"
 {% else %}

src/crystal/system/event_loop/file_descriptor.cr

@@ -19,5 +19,13 @@ abstract class Crystal::EventLoop
 
     # Closes the file descriptor resource.
     abstract def close(file_descriptor : Crystal::System::FileDescriptor) : Nil
+
+    # Removes the file descriptor from the event loop. Can be used to free up
+    # memory resources associated with the file descriptor, as well as removing
+    # the file descriptor from kernel data structures.
+    #
+    # Called by `::IO::FileDescriptor#finalize` before closing the file
+    # descriptor. Errors shall be silently ignored.
+    abstract def delete(file_descriptor : Crystal::System::FileDescriptor) : Nil
   end
 end

src/crystal/system/event_loop/socket.cr

@@ -62,5 +62,13 @@ abstract class Crystal::EventLoop
 
     # Closes the socket.
     abstract def close(socket : ::Socket) : Nil
+
+    # Removes the socket from the event loop. Can be used to free up memory
+    # resources associated with the socket, as well as removing the socket from
+    # kernel data structures.
+    #
+    # Called by `::Socket#finalize` before closing the socket. Errors shall be
+    # silently ignored.
+    abstract def delete(socket : ::Socket) : Nil
   end
 end

src/crystal/system/unix/epoll.cr

@@ -0,0 +1,57 @@
+{% skip_file unless flag?(:linux) || flag?(:solaris) %}
+
+require "c/sys/epoll"
+
+struct Crystal::System::Epoll
+  def initialize
+    @epfd = LibC.epoll_create1(LibC::EPOLL_CLOEXEC)
+    raise RuntimeError.from_errno("epoll_create1") if @epfd == -1
+  end
+
+  def fd : Int32
+    @epfd
+  end
+
+  def add(fd : Int32, epoll_event : LibC::EpollEvent*) : Nil
+    if LibC.epoll_ctl(@epfd, LibC::EPOLL_CTL_ADD, fd, epoll_event) == -1
+      raise RuntimeError.from_errno("epoll_ctl(EPOLL_CTL_ADD)") unless Errno.value == Errno::EPERM
+    end
+  end
+
+  def add(fd : Int32, events : UInt32, u64 : UInt64) : Nil
+    epoll_event = uninitialized LibC::EpollEvent
+    epoll_event.events = events
+    epoll_event.data.u64 = u64
+    add(fd, pointerof(epoll_event))
+  end
+
+  def modify(fd : Int32, epoll_event : LibC::EpollEvent*) : Nil
+    if LibC.epoll_ctl(@epfd, LibC::EPOLL_CTL_MOD, fd, epoll_event) == -1
+      raise RuntimeError.from_errno("epoll_ctl(EPOLL_CTL_MOD)")
+    end
+  end
+
+  # OPTIMIZE: we might be able to spare the errno checks for EPERM and ENOENT
+  def delete(fd : Int32) : Nil
+    delete(fd) do
+      raise RuntimeError.from_errno("epoll_ctl(EPOLL_CTL_DEL)") unless Errno.value.in?(Errno::EPERM, Errno::ENOENT)
+    end
+  end
+
+  def delete(fd : Int32, &) : Nil
+    if LibC.epoll_ctl(@epfd, LibC::EPOLL_CTL_DEL, fd, nil) == -1
+      yield
+    end
+  end
+
+  # `timeout` is in milliseconds; -1 will wait indefinitely; 0 will never wait.
+  def wait(events : Slice(LibC::EpollEvent), timeout : Int32) : Slice(LibC::EpollEvent)
+    count = LibC.epoll_wait(@epfd, events.to_unsafe, events.size, timeout)
+    raise RuntimeError.from_errno("epoll_wait") if count == -1 && Errno.value != Errno::EINTR
+    events[0, count.clamp(0..)]
+  end
+
+  def close : Nil
+    LibC.close(@epfd)
+  end
+end