Avoid ScheduledTask timer lock inversion

lib/concurrent-ruby/concurrent/scheduled_task.rb

@@ -187,7 +187,7 @@ def initialize(delay, opts = {}, &task)
         @args = get_arguments_from(opts)
         @delay = delay.to_f
         @task = task
-        @time = nil
+        self.schedule_time = nil
         @executor = Options.executor_from_options(opts) || Concurrent.global_io_executor
         self.observers = Collection::CopyOnNotifyObserverSet.new
       end
@@ -203,9 +203,8 @@ def initial_delay
     # The monotonic time at which the the task is scheduled to be executed.
     #
     # @return [Float] the schedule time or nil if `unscheduled`
-    def schedule_time
-      synchronize { @time }
-    end
+    attr_volatile :schedule_time # read by TimerSet under its own lock, so it must not take the task lock
+    private :schedule_time=
 
     # Comparator which orders by schedule time.
     #
@@ -311,7 +310,7 @@ def process_task
     # @!visibility private
     def ns_schedule(delay)
       @delay = delay
-      @time = Concurrent.monotonic_time + @delay
+      self.schedule_time = Concurrent.monotonic_time + @delay
       @parent.send(:post_task, self)
     end
 

spec/concurrent/executor/timer_set_spec.rb

@@ -306,6 +306,32 @@ module Concurrent
         expect(success).to be false
         expect(task.schedule_time).to be_within(0.01).of(expected)
       end
+
+      it 'does not deadlock when the timer reads the schedule time during rescheduling' do
+        task = subject.post(10){ nil }
+        task_locked = Concurrent::CountDownLatch.new
+        timer_locked = Concurrent::CountDownLatch.new
+
+        rescheduler = in_thread do
+          task.send(:synchronize) do
+            task_locked.count_down
+            timer_locked.wait(2)
+            task.reschedule(20)
+          end
+        end
+        reader = in_thread do
+          task_locked.wait(2)
+          subject.send(:synchronize) do
+            timer_locked.count_down
+            task.schedule_time
+          end
+        end
+
+        rescheduled = rescheduler.join(2)
+        [rescheduler, reader].each(&:kill) # cleanup, so the timer set can shut down after a deadlock
+        expect(rescheduled).not_to be_nil
+        expect(rescheduler.value).to be true
+      end
     end
 
     context 'task resetting' do