Register execution context schedulers with the event loop (#16519)

Some event loops need to be aware of the execution context schedulers, to allocate and cleanup resources as schedulers are started and stopped.

For example the io_uring evloop will need to keep a ring per thread, yet can reuse the main ring for the first scheduler (index 0), and also needs to keep a list of all the active rings.

- Add `Crystal::EventLoop#initialize(parallelism)`.
- Add `Crystal::EventLoop#register(scheduler, index)`.
- Add `Crystal::EventLoop#unregister(scheduler)`.

Author: ysbaddaden

src/crystal/event_loop.cr

@@ -20,9 +20,14 @@ abstract class Crystal::EventLoop
     {% end %}
   end
 
-  # Creates an event loop instance
-  def self.create : self
-    backend_class.new
+  # Creates an event loop instance.
+  #
+  # The *parallelism* arg is informational. It reports how many schedulers are
+  # expected to register with the event loop instance. Because schedulers are
+  # dynamically started and execution contexts can be resized, more or less
+  # schedulers may really register in practice.
+  def self.create(parallelism : Int32 = 1) : self
+    backend_class.new(parallelism)
   end
 
   def self.default_file_blocking? : Bool
@@ -67,6 +72,14 @@ abstract class Crystal::EventLoop
     # enqueueing in parallel, so the caller is responsible and in control for
     # when and how the fibers will be enqueued.
     abstract def run(queue : Fiber::List*, blocking : Bool) : Nil
+
+    # Called once before *scheduler* is started. Optional hook.
+    def register(scheduler : Fiber::ExecutionContext::Scheduler, index : Int32) : Nil
+    end
+
+    # Called once before *scheduler* is shut down. Optional hook.
+    def unregister(scheduler : Fiber::ExecutionContext::Scheduler) : Nil
+    end
   {% end %}
 
   # Tells a blocking run loop to no longer wait for events to activate. It may

src/crystal/event_loop/epoll.cr

@@ -4,7 +4,7 @@ require "../system/unix/eventfd"
 require "../system/unix/timerfd"
 
 class Crystal::EventLoop::Epoll < Crystal::EventLoop::Polling
-  def initialize
+  def initialize(parallelism : Int32)
     # the epoll instance
     @epoll = System::Epoll.new
 

src/crystal/event_loop/iocp.cr

@@ -26,7 +26,7 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
   @timer_packet = LibC::HANDLE.null
   @timer_key : System::IOCP::CompletionKey?
 
-  def initialize
+  def initialize(parallelism : Int32)
     @mutex = Thread::Mutex.new
     @timers = Timers(Timer).new
 

src/crystal/event_loop/kqueue.cr

@@ -10,7 +10,7 @@ class Crystal::EventLoop::Kqueue < Crystal::EventLoop::Polling
     @pipe = uninitialized LibC::Int[2]
   {% end %}
 
-  def initialize
+  def initialize(parallelism : Int32)
     # the kqueue instance
     @kqueue = System::Kqueue.new
 

src/crystal/event_loop/libevent.cr

@@ -12,6 +12,9 @@ class Crystal::EventLoop::LibEvent < Crystal::EventLoop
 
   private getter(event_base) { Crystal::EventLoop::LibEvent::Event::Base.new }
 
+  def initialize(parallelism : Int32)
+  end
+
   {% unless flag?(:preview_mt) %}
     # Reinitializes the event loop after a fork.
     def after_fork : Nil

src/crystal/event_loop/wasi.cr

@@ -8,6 +8,9 @@ class Crystal::EventLoop::Wasi < Crystal::EventLoop
     false
   end
 
+  def initialize(parallelism : Int32)
+  end
+
   # Runs the event loop.
   def run(blocking : Bool) : Bool
     raise NotImplementedError.new("Crystal::Wasi::EventLoop.run")

src/fiber/execution_context/isolated.cr

@@ -48,7 +48,11 @@ module Fiber::ExecutionContext
     @main_fiber : Fiber
 
     # :nodoc:
-    getter(event_loop : Crystal::EventLoop) { Crystal::EventLoop.create }
+    getter(event_loop : Crystal::EventLoop) do
+      evloop = Crystal::EventLoop.create(parallelism: 1)
+      evloop.register(self, index: 0)
+      evloop
+    end
 
     getter? running : Bool = true
     @enqueued = false
@@ -207,7 +211,12 @@ module Fiber::ExecutionContext
         @running = false
         @wait_list.consume_each(&.value.enqueue)
       end
-      ExecutionContext.execution_contexts.delete(self)
+
+      begin
+        @event_loop.try(&.unregister(self))
+      ensure
+        ExecutionContext.execution_contexts.delete(self)
+      end
     end
 
     # Blocks the calling fiber until the isolated context fiber terminates.

src/fiber/execution_context/parallel.cr

@@ -68,7 +68,7 @@ module Fiber::ExecutionContext
     getter stack_pool : Fiber::StackPool = Fiber::StackPool.new
 
     # :nodoc:
-    getter event_loop : Crystal::EventLoop = Crystal::EventLoop.create
+    getter event_loop : Crystal::EventLoop
     @event_loop_lock = Atomic(Bool).new(false)
 
     @parked = Atomic(Int32).new(0)
@@ -107,6 +107,7 @@ module Fiber::ExecutionContext
       @global_queue = GlobalQueue.new(@mutex)
       @schedulers = Array(Scheduler).new(capacity)
       @threads = Array(Thread).new(capacity)
+      @event_loop = Crystal::EventLoop.create(capacity)
 
       @rng = Random::PCG32.new
 
@@ -144,9 +145,13 @@ 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 do |index|
-        @schedulers << Scheduler.new(self, "#{@name}-#{index}")
-      end
+      capacity.times { |index| @schedulers << start_scheduler(index) }
+    end
+
+    private def start_scheduler(index)
+      scheduler = Scheduler.new(self, "#{@name}-#{index}")
+      @event_loop.register(scheduler, index)
+      scheduler
     end
 
     # Attaches *scheduler* to the current `Thread`, usually the process' main
@@ -204,7 +209,7 @@ module Fiber::ExecutionContext
 
         if new_capacity > old_capacity
           @schedulers = Array(Scheduler).new(new_capacity) do |index|
-            old_schedulers[index]? || Scheduler.new(self, "#{@name}-#{index}")
+            old_schedulers[index]? || start_scheduler(index)
           end
           threads = Array(Thread).new(new_capacity)
           old_threads.each { |thread| threads << thread }

src/fiber/execution_context/parallel/scheduler.cr

@@ -152,6 +152,8 @@ module Fiber::ExecutionContext
           Crystal.print_error_buffered("BUG: %s#run_loop [%s] crashed",
             self.class.name, @name, exception: exception)
         end
+      ensure
+        @event_loop.unregister(self)
       end
 
       private def find_next_runnable : Fiber?