- mutex to correctly access hash with fiber timeouts (concurrency happens between fiber itself and iodine worker)

- kill fiber forcefully if it still alive after error was raised

Author: serhii-sadovskyi

lib/rage/fiber_scheduler.rb

@@ -11,6 +11,7 @@ def initialize
     @dns_cache = {}
 
     @alive_fibers = Hash.new { |h, k| h[k] = {} }
+    @fibers_mutex = Mutex.new
 
     start_timeout_worker
   end
@@ -75,17 +76,21 @@ def timeout_after(duration, exception_class = Timeout::Error, *exception_argumen
     fiber = Fiber.current
     timeout_deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + duration
 
-    @alive_fibers[fiber.__get_id][timeout_deadline] = {
-      fiber: fiber,
-      timeout_deadline: timeout_deadline,
-      exception_class: exception_class,
-      exception_arguments: exception_arguments
-    }
+    @fibers_mutex.synchronize do
+      @alive_fibers[fiber.__get_id][timeout_deadline] = {
+        fiber: fiber,
+        timeout_deadline: timeout_deadline,
+        exception_class: exception_class,
+        exception_arguments: exception_arguments
+      }
+    end
 
     begin
       block.call
     ensure
-      @alive_fibers[fiber.__get_id].delete(timeout_deadline)
+      @fibers_mutex.synchronize do
+        @alive_fibers[fiber.__get_id].delete(timeout_deadline)
+      end
     end
   end
 
@@ -166,25 +171,31 @@ def start_timeout_worker
   end
 
   def check_timeouts
-    @alive_fibers.delete_if do |fiber_id, timeouts|
-      timeouts.delete_if do |timeout_key, fiber_hash|
-        current_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    @fibers_mutex.synchronize do
+      @alive_fibers.delete_if do |fiber_id, timeouts|
+        timeouts.delete_if do |timeout_key, fiber_hash|
+          current_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+          next false if current_time < fiber_hash[:timeout_deadline]
 
-        return false if current_time < fiber_hash[:timeout_deadline]
+          fiber = fiber_hash[:fiber]
+          unblock(nil, fiber)
 
-        fiber = fiber_hash[:fiber]
-        unblock(nil, fiber)
+          if fiber.alive?
+            fiber.raise(RageTimeout)
 
-        if fiber.alive?
-          fiber.raise(RageTimeout)
-        else
-          timeouts.delete(timeout_key)
+            ::Iodine.run_after(1000) do
+              fiber.kill if fiber.alive?
+            end
+          else
+            timeouts.delete(timeout_key)
+          end
+
+          true
         end
 
-        true
+        timeouts.length == 0
       end
-
-      timeouts.length == 0
     end
   end
 end