Make Semian work in non-main Ractors

ext/semian/semian.c

@@ -7,6 +7,8 @@ void Init_semian()
   VALUE cSemian, cResource;
   struct seminfo info_buf;
 
+  rb_ext_ractor_safe(true);
+
   cSemian = rb_const_get(rb_cObject, rb_intern("Semian"));
 
   /*

gemfiles/redis_client.gemfile

@@ -10,7 +10,7 @@ gem "toxiproxy"
 gem "webrick"
 gem "concurrent-ruby"
 
-gem "hiredis-client", github: "redis-rb/redis-client"
+gem "hiredis-client", github: "redis-rb/redis-client", tag: "v0.30.1"
 gem "hiredis", "~> 0.6"
 gem "redis", github: "redis/redis-rb"
 

lib/semian.rb

@@ -106,7 +106,17 @@ module Semian
   OpenCircuitError = Class.new(BaseError)
   SemaphoreMissingError = Class.new(BaseError)
 
-  attr_accessor :maximum_lru_size, :minimum_lru_time, :default_permissions, :namespace, :default_force_config_validation
+  def main_ractor?
+    !defined?(Ractor.main?) || Ractor.main?
+  end
+  private :main_ractor?
+
+  attr_accessor :maximum_lru_size, :minimum_lru_time, :default_permissions, :default_force_config_validation
+  attr_reader :namespace
+
+  def namespace=(namespace)
+    @namespace = namespace.is_a?(String) ? -namespace : namespace
+  end
 
   self.maximum_lru_size = 500
   self.minimum_lru_time = 300 # 300 seconds / 5 minutes
@@ -116,9 +126,7 @@ module Semian
   # We only allow disabling thread-safety for parts of the code that are on the hot path.
   # Since locking there could have a significant impact. Everything else is enforced thread safety
   def thread_safe?
-    return @thread_safe if defined?(@thread_safe)
-
-    @thread_safe = true
+    defined?(@thread_safe) ? @thread_safe : true
   end
 
   def thread_safe=(thread_safe)
@@ -128,9 +136,15 @@ def thread_safe=(thread_safe)
   @reset_mutex = Mutex.new
 
   def issue_disabled_semaphores_warning
-    return if defined?(@warning_issued)
+    if main_ractor?
+      return if defined?(@warning_issued)
 
-    @warning_issued = true
+      @warning_issued = true
+    else
+      return if Ractor[:semian_warning_issued]
+
+      Ractor[:semian_warning_issued] = true
+    end
     if !sysv_semaphores_supported?
       logger.info("Semian sysv semaphores are not supported on #{RUBY_PLATFORM} - all operations will no-op")
     elsif disabled?
@@ -155,9 +169,22 @@ def to_s
     end
   end
 
-  attr_accessor :logger
+  def logger
+    return @logger if main_ractor?
+
+    Ractor[:semian_logger] || shared_logger || Ractor.store_if_absent(:semian_default_logger) { Logger.new($stderr) }
+  end
+
+  def logger=(logger)
+    if main_ractor?
+      @logger_set = true
+      @logger = logger
+    else
+      Ractor[:semian_logger] = logger
+    end
+  end
 
-  self.logger = Logger.new($stderr)
+  @logger = Logger.new($stderr)
 
   # Registers a resource.
   #
@@ -296,6 +323,11 @@ def unregister_all_resources
   end
 
   def reset!
+    unless main_ractor?
+      Ractor[:semian_consumers] = Concurrent::Map.new
+      return Ractor[:semian_resources] = LRUHash.new
+    end
+
     @reset_mutex.synchronize do
       @consumers = Concurrent::Map.new
       @resources = LRUHash.new
@@ -318,6 +350,7 @@ def disable_bulkheads_for_thread(thread)
   end
 
   def resources
+    return Ractor.store_if_absent(:semian_resources) { LRUHash.new } unless main_ractor?
     return @resources if defined?(@resources) && @resources
 
     @reset_mutex.synchronize do
@@ -326,6 +359,7 @@ def resources
   end
 
   def consumers
+    return Ractor.store_if_absent(:semian_consumers) { Concurrent::Map.new } unless main_ractor?
     return @consumers if defined?(@consumers) && @consumers
 
     @reset_mutex.synchronize do
@@ -335,6 +369,12 @@ def consumers
 
   private
 
+  def shared_logger
+    @logger if @logger_set
+  rescue Ractor::IsolationError
+    nil
+  end
+
   def create_dual_circuit_breaker(name, **options)
     return if ENV.key?("SEMIAN_CIRCUIT_BREAKER_DISABLED")
 

lib/semian/instrumentable.rb

@@ -5,18 +5,29 @@
 module Semian
   module Instrumentable
     SUBSCRIBERS = Concurrent::Map.new
+    SHAREABLE_SUBSCRIBERS_LOCK = Mutex.new
+    private_constant :SHAREABLE_SUBSCRIBERS_LOCK
 
     def subscribe(name = rand, &block)
-      SUBSCRIBERS[name] = block
+      if main_ractor?
+        SUBSCRIBERS[name] = block
+        update_shareable_subscribers
+      else
+        ractor_subscribers[name] = block
+      end
       name
     end
 
     def unsubscribe(name)
-      SUBSCRIBERS.delete(name)
+      return ractor_subscribers.delete(name) unless main_ractor?
+
+      subscriber = SUBSCRIBERS.delete(name)
+      update_shareable_subscribers
+      subscriber
     end
 
     def subscribers
-      SUBSCRIBERS
+      main_ractor? ? SUBSCRIBERS : ractor_subscribers
     end
 
     # Args:
@@ -26,7 +37,28 @@ def subscribers
     #   adapter (string)
     #   payload (optional)
     def notify(*args)
-      SUBSCRIBERS.values.each { |subscriber| subscriber.call(*args) }
+      if main_ractor?
+        SUBSCRIBERS.values.each { |subscriber| subscriber.call(*args) }
+      else
+        @shareable_subscribers&.each_value { |subscriber| subscriber.call(*args) }
+        Ractor[:semian_subscribers]&.values&.each { |subscriber| subscriber.call(*args) }
+      end
+    end
+
+    private
+
+    def ractor_subscribers
+      Ractor.store_if_absent(:semian_subscribers) { Concurrent::Map.new }
+    end
+
+    def update_shareable_subscribers
+      SHAREABLE_SUBSCRIBERS_LOCK.synchronize do
+        shareable = {}
+        SUBSCRIBERS.each_pair do |name, block|
+          shareable[name] = block if Ractor.shareable?(name) && Ractor.shareable?(block)
+        end
+        @shareable_subscribers = Ractor.make_shareable(shareable)
+      end
     end
   end
 end

test/adapters/redis_client_test.rb

@@ -5,6 +5,7 @@
 require "hiredis-client"
 require "semian/redis_client"
 require "benchmark"
+require "helpers/ractor_helper"
 
 module RedisClientTests
   include BackgroundHelper
@@ -629,12 +630,44 @@ def assert_redis_timeout_in_delta(expected_timeout:, delta: 0.1, &block)
 
 class TestRedisClient < Minitest::Test
   include RedisClientTests
+  include RactorHelper
+
+  def test_connection_errors_open_the_circuit_in_a_non_main_ractor
+    skip("Semian supports non-main Ractors from Ruby 3.4") unless Ractor.respond_to?(:main?)
+
+    assert_equal(["PONG"], ping_in_a_ractor(1))
+    @proxy.down do
+      errors = [RedisClient::CannotConnectError] * ERROR_THRESHOLD + [RedisClient::CircuitOpenError]
+
+      assert_equal(errors, ping_in_a_ractor(ERROR_THRESHOLD + 1))
+    end
+  end
 
   private
 
   def redis_driver
     :ruby
   end
+
+  def ping_in_a_ractor(times)
+    options = {
+      host: SemianConfig["toxiproxy_upstream_host"],
+      port: SemianConfig["redis_toxiproxy_port"],
+      reconnect_attempts: 0,
+      timeout: REDIS_TIMEOUT,
+      semian: SEMIAN_OPTIONS.merge(bulkhead: false),
+    }
+    ractor = Ractor.new(options, times) do |ractor_options, count|
+      Semian.logger = Logger.new(nil)
+      client = RedisClient.new(**ractor_options, driver: RedisClient::RubyConnection)
+      Array.new(count) do
+        client.call("PING")
+      rescue RedisClient::ConnectionError => error
+        error.class
+      end
+    end
+    ractor_value(ractor)
+  end
 end
 
 class TestHiredisClient < Minitest::Test

test/helpers/ractor_helper.rb

@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+module RactorHelper
+  private
+
+  def ractor_value(ractor)
+    ractor.respond_to?(:value) ? ractor.value : ractor.take
+  end
+end