Merge pull request #417 from ikyn-inc/interruptible

Reimplement Interruptible

Author: rosa

.gitignore

@@ -14,5 +14,8 @@
 # Folder for Visual Studio Code
 /.vscode/
 
+# Files for RVM holdouts
+.ruby-gemset
+
 # misc
 .DS_Store

.rubocop.yml

@@ -4,7 +4,7 @@
 inherit_gem: { rubocop-rails-omakase: rubocop.yml }
 
 AllCops:
-  TargetRubyVersion: 3.0
+  TargetRubyVersion: 3.3
   Exclude:
     - "test/dummy/db/schema.rb"
     - "test/dummy/db/queue_schema.rb"

Gemfile.lock

@@ -75,6 +75,7 @@ GEM
       reline (>= 0.4.2)
     json (2.8.2)
     language_server-protocol (3.17.0.3)
+    logger (1.6.2)
     loofah (2.23.1)
       crass (~> 1.0.2)
       nokogiri (>= 1.12.0)
@@ -177,16 +178,19 @@ GEM
 PLATFORMS
   arm64-darwin-22
   arm64-darwin-23
+  arm64-darwin-24
   x86_64-darwin-21
   x86_64-darwin-23
   x86_64-linux
 
 DEPENDENCIES
-  debug
+  debug (~> 1.9)
+  logger
   mocha
   mysql2
   pg
   puma
+  rdoc
   rubocop-rails-omakase
   solid_queue!
   sqlite3

lib/solid_queue/processes/interruptible.rb

@@ -7,31 +7,23 @@ def wake_up
     end
 
     private
-      SELF_PIPE_BLOCK_SIZE = 11
 
       def interrupt
-        self_pipe[:writer].write_nonblock(".")
-      rescue Errno::EAGAIN, Errno::EINTR
-        # Ignore writes that would block and retry
-        # if another signal arrived while writing
-        retry
+        queue << true
       end
 
       def interruptible_sleep(time)
-        if time > 0 && self_pipe[:reader].wait_readable(time)
-          loop { self_pipe[:reader].read_nonblock(SELF_PIPE_BLOCK_SIZE) }
-        end
-      rescue Errno::EAGAIN, Errno::EINTR
+        # Invoking from the main thread can result in a 35% slowdown (at least when running the test suite).
+        # Using some form of Async (Futures) addresses this performance issue.
+        Concurrent::Promises.future(time) do |timeout|
+          if timeout > 0 && queue.pop(timeout:)
+            queue.clear
+          end
+        end.value
       end
 
-      # Self-pipe for signal-handling (http://cr.yp.to/docs/selfpipe.html)
-      def self_pipe
-        @self_pipe ||= create_self_pipe
-      end
-
-      def create_self_pipe
-        reader, writer = IO.pipe
-        { reader: reader, writer: writer }
+      def queue
+        @queue ||= Queue.new
       end
   end
 end

solid_queue.gemspec

@@ -37,11 +37,13 @@ Gem::Specification.new do |spec|
   spec.add_dependency "fugit", "~> 1.11.0"
   spec.add_dependency "thor", "~> 1.3.1"
 
-  spec.add_development_dependency "debug"
+  spec.add_development_dependency "debug", "~> 1.9"
   spec.add_development_dependency "mocha"
   spec.add_development_dependency "puma"
   spec.add_development_dependency "mysql2"
   spec.add_development_dependency "pg"
   spec.add_development_dependency "sqlite3"
   spec.add_development_dependency "rubocop-rails-omakase"
+  spec.add_development_dependency "rdoc"
+  spec.add_development_dependency "logger"
 end

test/dummy/config/initializers/enable_yjit.rb

@@ -0,0 +1,12 @@
+# frozen_string_literal: true
+
+# Ideally, tests should be configured as close to production settings as
+# possible and YJIT is likely to be enabled. While it's highly unlikely
+# YJIT would cause issues, enabling it confirms this assertion.
+#
+# Configured via initializer to align with Rails 7.1 default in gemspec
+if defined?(RubyVM::YJIT.enable)
+  Rails.application.config.after_initialize do
+    RubyVM::YJIT.enable
+  end
+end

test/integration/concurrency_controls_test.rb

@@ -85,7 +85,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   test "run several jobs over the same record sequentially, with some of them failing" do
     ("A".."F").each_with_index do |name, i|
       # A, C, E will fail, for i= 0, 2, 4
-      SequentialUpdateResultJob.perform_later(@result, name: name, pause: 0.2.seconds, exception: (RuntimeError if i.even?))
+      SequentialUpdateResultJob.perform_later(@result, name: name, pause: 0.2.seconds, exception: (ExpectedTestError if i.even?))
     end
 
     ("G".."K").each do |name|

test/integration/instrumentation_test.rb

@@ -162,7 +162,7 @@ class InstrumentationTest < ActiveSupport::TestCase
 
   test "errors when deregistering processes are included in deregister_process events" do
     previous_thread_report_on_exception, Thread.report_on_exception = Thread.report_on_exception, false
-    error = RuntimeError.new("everything is broken")
+    error = ExpectedTestError.new("everything is broken")
     SolidQueue::Process.any_instance.expects(:destroy!).raises(error).at_least_once
 
     events = subscribed("deregister_process.solid_queue") do
@@ -182,7 +182,7 @@ class InstrumentationTest < ActiveSupport::TestCase
   end
 
   test "retrying failed job emits retry event" do
-    RaisingJob.perform_later(RuntimeError, "A")
+    RaisingJob.perform_later(ExpectedTestError, "A")
     job = SolidQueue::Job.last
 
     worker = SolidQueue::Worker.new.tap(&:start)
@@ -198,7 +198,7 @@ class InstrumentationTest < ActiveSupport::TestCase
   end
 
   test "retrying failed jobs in bulk emits retry_all" do
-    3.times { RaisingJob.perform_later(RuntimeError, "A") }
+    3.times { RaisingJob.perform_later(ExpectedTestError, "A") }
     AddToBufferJob.perform_later("A")
 
     jobs = SolidQueue::Job.last(4)
@@ -392,7 +392,7 @@ class InstrumentationTest < ActiveSupport::TestCase
   test "thread errors emit thread_error events" do
     previous_thread_report_on_exception, Thread.report_on_exception = Thread.report_on_exception, false
 
-    error = RuntimeError.new("everything is broken")
+    error = ExpectedTestError.new("everything is broken")
     SolidQueue::ClaimedExecution::Result.expects(:new).raises(error).at_least_once
 
     AddToBufferJob.perform_later "hey!"

test/integration/jobs_lifecycle_test.rb

@@ -4,11 +4,13 @@
 
 class JobsLifecycleTest < ActiveSupport::TestCase
   setup do
+    SolidQueue.on_thread_error = silent_on_thread_error_for([ ExpectedTestError, RaisingJob::DefaultError ])
     @worker = SolidQueue::Worker.new(queues: "background", threads: 3)
     @dispatcher = SolidQueue::Dispatcher.new(batch_size: 10, polling_interval: 0.2)
   end
 
   teardown do
+    SolidQueue.on_thread_error = @on_thread_error
     @worker.stop
     @dispatcher.stop
 
@@ -29,16 +31,16 @@ class JobsLifecycleTest < ActiveSupport::TestCase
   end
 
   test "enqueue and run jobs that fail without retries" do
-    RaisingJob.perform_later(RuntimeError, "A")
-    RaisingJob.perform_later(RuntimeError, "B")
+    RaisingJob.perform_later(ExpectedTestError, "A")
+    RaisingJob.perform_later(ExpectedTestError, "B")
     jobs = SolidQueue::Job.last(2)
 
     @dispatcher.start
     @worker.start
 
     wait_for_jobs_to_finish_for(3.seconds)
 
-    message = "raised RuntimeError for the 1st time"
+    message = "raised ExpectedTestError for the 1st time"
     assert_equal [ "A: #{message}", "B: #{message}" ], JobBuffer.values.sort
 
     assert_empty SolidQueue::Job.finished

test/integration/processes_lifecycle_test.rb

@@ -144,11 +144,11 @@ class ProcessesLifecycleTest < ActiveSupport::TestCase
   test "process some jobs that raise errors" do
     2.times { enqueue_store_result_job("no error", :background) }
     2.times { enqueue_store_result_job("no error", :default) }
-    error1 = enqueue_store_result_job("error", :background, exception: RuntimeError)
+    error1 = enqueue_store_result_job("error", :background, exception: ExpectedTestError)
     enqueue_store_result_job("no error", :background, pause: 0.03)
-    error2 = enqueue_store_result_job("error", :background, exception: RuntimeError, pause: 0.05)
+    error2 = enqueue_store_result_job("error", :background, exception: ExpectedTestError, pause: 0.05)
     2.times { enqueue_store_result_job("no error", :default, pause: 0.01) }
-    error3 = enqueue_store_result_job("error", :default, exception: RuntimeError)
+    error3 = enqueue_store_result_job("error", :default, exception: ExpectedTestError)
 
     wait_for_jobs_to_finish_for(2.second, except: [ error1, error2, error3 ])