Add Fiber::Execution.syscall(&)

Marks the scheduler has running a blocking syscall.

The monitor thread now ticks every 10ms to check if any scheduler in any
ST or MT context is blocked on a syscall, and if so tries to detach the
scheduler from the thread. On success the scheduler is moved to another
thread, taken from the thread pool.

The fiber doing a blocking syscall will still be blocked, but other
fibers may be resumed by the scheduler.

When the blocking syscall returns, the thread will try to unmark the
scheduler as running a blocking syscall. Upon success it returns. Upon
failure, it enqueues the current fiber back into its execution context,
and checks itself back into the thread pool.

Author: ysbaddaden

src/fiber.cr

@@ -355,6 +355,15 @@ class Fiber
     {% end %}
   end
 
+  # :nodoc:
+  def self.syscall(&)
+    {% if flag?(:execution_context) %}
+      ExecutionContext::Scheduler.current.syscall { yield }
+    {% else %}
+      yield
+    {% end %}
+  end
+
   def to_s(io : IO) : Nil
     io << "#<" << self.class.name << ":0x"
     object_id.to_s(io, 16)

src/fiber/execution_context/isolated.cr

@@ -233,11 +233,27 @@ module Fiber::ExecutionContext
     def status : String
       if @waiting
         "event-loop"
+      elsif @syscall == SYSCALL_FLAG
+        "syscall"
       elsif @running
         "running"
       else
         "shutdown"
       end
     end
+
+    protected def enter_syscall : UInt32
+      @syscall.lazy_set(SYSCALL_FLAG)
+    end
+
+    protected def leave_syscall?(value : UInt32) : Bool
+      @syscall.lazy_set(0_u32)
+      true
+    end
+
+    # :nodoc:
+    def syscall(& : -> U) : U forall U
+      yield
+    end
   end
 end

src/fiber/execution_context/monitor.cr

@@ -1,11 +1,25 @@
 module Fiber::ExecutionContext
   # :nodoc:
   class Monitor
-    DEFAULT_EVERY = 5.seconds
+    # :nodoc:
+    struct Timer
+      def initialize(@every : Time::Span)
+        @last = Time.monotonic
+      end
+
+      def elapsed?(now : Time::Span) : Bool
+        ret = @last + @every <= now
+        @last = now if ret
+        ret
+      end
+    end
+
+    DEFAULT_EVERY = 10.milliseconds
 
     @thread : Thread?
 
     def initialize(@every = DEFAULT_EVERY)
+      @stack_collect_timer = Timer.new(5.seconds)
       @thread = Thread.new(name: "SYSMON") { run_loop }
     end
 
@@ -35,7 +49,8 @@ module Fiber::ExecutionContext
     # OS.
     private def run_loop : Nil
       every do |now|
-        collect_stacks
+        transfer_schedulers_blocked_on_syscall
+        collect_stacks if @stack_collect_timer.elapsed?(now)
       end
     end
 
@@ -55,6 +70,30 @@ module Fiber::ExecutionContext
       end
     end
 
+    # Iterates each ExecutionContext::Scheduler and transfers the Scheduler for
+    # any Thread currently blocked on a syscall.
+    #
+    # OPTIMIZE: a scheduler in a MT context might not need to be transferred if
+    # its queue is empty and another scheduler in the context is blocked on the
+    # event loop.
+    private def transfer_schedulers_blocked_on_syscall : Nil
+      ExecutionContext.each do |execution_context|
+        if execution_context.responds_to?(:each_scheduler)
+          execution_context.each_scheduler do |scheduler|
+            next unless scheduler.detach_syscall?
+
+            Crystal.trace :sched, "reassociate",
+              scheduler: scheduler,
+              syscall: scheduler.thread.current_fiber
+
+            pool = ExecutionContext.thread_pool
+            pool.detach(scheduler.thread)
+            pool.checkout(scheduler)
+          end
+        end
+      end
+    end
+
     # Iterates each execution context and collects unused fiber stacks.
     #
     # OPTIMIZE: should maybe happen during GC collections (?)

src/fiber/execution_context/multi_threaded/scheduler.cr

@@ -126,6 +126,10 @@ module Fiber::ExecutionContext
 
         # nothing to do: start spinning
         spinning do
+          # usually empty but the scheduler may have been transferred to another
+          # thread with queued fibers
+          yield @runnables.shift?
+
           yield @global_queue.grab?(@runnables, divisor: @execution_context.size)
 
           if @execution_context.lock_evloop? { @event_loop.run(pointerof(list), blocking: false) }
@@ -257,6 +261,8 @@ module Fiber::ExecutionContext
           "event-loop"
         elsif @parked
           "parked"
+        elsif @syscall.get(:relaxed).bits_set?(SYSCALL_FLAG)
+          "syscall"
         else
           "running"
         end

src/fiber/execution_context/scheduler.cr

@@ -6,6 +6,11 @@ module Fiber::ExecutionContext
       Thread.current.scheduler
     end
 
+    @[AlwaysInline]
+    def self.current? : Scheduler?
+      Thread.current.scheduler?
+    end
+
     protected abstract def thread : Thread
     protected abstract def execution_context : ExecutionContext
 
@@ -68,5 +73,57 @@ module Fiber::ExecutionContext
     # Returns the current status of the scheduler. For example `"running"`,
     # `"event-loop"` or `"parked"`.
     abstract def status : String
+
+    # :nodoc:
+    SYSCALL_FLAG = 1_u32
+
+    # :nodoc:
+    # use to increment the counter to avoid ABA issues
+    SYSCALL_INCREMENT = 2_u32
+
+    @syscall = Atomic(UInt32).new(0_u32)
+
+    # :nodoc:
+    #
+    # Marks the current scheduler as doing a syscall that may block the current
+    # thread for an unknown time. This allows the monitor thread to move the
+    # scheduler to another thread, without waiting for the syscall to complete.
+    #
+    # When the syscall eventually completes, the thread will check if the
+    # scheduler has been detached and either continue running the current fiber if
+    # the scheduler is still attached, or enqueue the current fiber back into
+    # its execution context, then returns itself back into the thread pool.
+    def syscall(& : -> U) : U forall U
+      scheduler = Scheduler.current
+      fiber = Fiber.current
+      value = scheduler.enter_syscall
+
+      ret = yield
+
+      if !scheduler.leave_syscall?(value)
+        # the scheduler has been detached: enqueue the current fiber back into
+        # its execution context, and return the thread back into the thread pool
+        scheduler.execution_context.enqueue(fiber)
+        ExecutionContext.thread_pool.checkin
+      end
+
+      ret
+    end
+
+    protected def enter_syscall : UInt32
+      old_value = @syscall.add(SYSCALL_FLAG | SYSCALL_INCREMENT, :acquire_release)
+      old_value += SYSCALL_FLAG | SYSCALL_INCREMENT
+    end
+
+    protected def leave_syscall?(value : UInt32) : Bool
+      new_value = (value & ~SYSCALL_FLAG) &+ SYSCALL_INCREMENT
+      _, success = @syscall.compare_and_set(value, new_value, :acquire_release, :relaxed)
+      success
+    end
+
+    protected def detach_syscall? : Bool
+      value = @syscall.get(:relaxed)
+      value.bits_set?(SYSCALL_FLAG) && leave_syscall?(value)
+    end
   end
 end

src/fiber/execution_context/single_threaded.cr

@@ -91,6 +91,10 @@ module Fiber::ExecutionContext
       ExecutionContext.thread_pool.checkout(self)
     end
 
+    protected def each_scheduler(& : Scheduler ->) : Nil
+      yield self.as(Scheduler)
+    end
+
     # :nodoc:
     def spawn(*, name : String? = nil, same_thread : Bool, &block : ->) : Fiber
       # whatever the value of same_thread: the fibers will always run on the
@@ -122,14 +126,24 @@ module Fiber::ExecutionContext
       end
     end
 
+    # FIXME: duplicates MultiThreaded::Scheduler#resume
     protected def resume(fiber : Fiber) : Nil
-      unless fiber.resumable?
+      Crystal.trace :sched, "resume", fiber: fiber
+
+      # fibers should always be ready to be resumed in the ST environment,
+      # unless SYSMON moved the scheduler to another thread while the fiber was
+      # blocked on a syscall; upon return the original thread might have
+      # enqueued the fiber, but hasn't swapcontext while this thread already
+      # tries to resume it
+      attempts = 0
+
+      until fiber.resumable?
         if fiber.dead?
           raise "BUG: tried to resume dead fiber #{fiber} (#{inspect})"
-        else
-          raise "BUG: can't resume running fiber #{fiber} (#{inspect})"
         end
+        attempts = Thread.delay(attempts)
       end
+
       swapcontext(fiber)
     end
 
@@ -186,6 +200,10 @@ module Fiber::ExecutionContext
 
       # nothing to do: start spinning
       spinning do
+        # usually empty but the scheduler may have been transferred to another
+        # thread with queued fibers
+        yield @runnables.shift?
+
         yield @global_queue.grab?(@runnables, divisor: 1)
 
         @event_loop.run(pointerof(list), blocking: false)
@@ -272,6 +290,8 @@ module Fiber::ExecutionContext
         "spinning"
       elsif @waiting.get(:relaxed)
         "event-loop"
+      elsif @syscall.get(:relaxed).bits_set?(SYSCALL_FLAG)
+        "syscall"
       else
         "running"
       end