PERF: Don't allow a single user to monopolize the defer queue (discou…

…rse#25593)

lib/hijack.rb

@@ -29,6 +29,7 @@ def hijack(info: nil, &blk)
         Scheduler::Defer.later(
           "hijack #{params["controller"]} #{params["action"]} #{info}",
           force: false,
+          current_user: current_user&.id,
           &scheduled.method(:resolve)
         )
       rescue WorkQueue::WorkQueueFull

lib/scheduler/defer.rb

@@ -10,7 +10,9 @@ def initialize
       @async = !Rails.env.test?
       @queue =
         WorkQueue::ThreadSafeWrapper.new(
-          WorkQueue::FairQueue.new(500) { WorkQueue::BoundedQueue.new(100) },
+          WorkQueue::FairQueue.new(:site, 500) do
+            WorkQueue::FairQueue.new(:user, 100) { WorkQueue::BoundedQueue.new(50) }
+          end,
         )
 
       @mutex = Mutex.new
@@ -48,15 +50,21 @@ def async=(val)
       @async = val
     end
 
-    def later(desc = nil, db = RailsMultisite::ConnectionManagement.current_db, force: true, &blk)
+    def later(
+      desc = nil,
+      db = RailsMultisite::ConnectionManagement.current_db,
+      force: true,
+      current_user: nil,
+      &blk
+    )
       @stats_mutex.synchronize do
         stats = (@stats[desc] ||= { queued: 0, finished: 0, duration: 0, errors: 0 })
         stats[:queued] += 1
       end
 
       if @async
         start_thread if !@thread&.alive? && !@paused
-        @queue.push({ key: db, task: [db, blk, desc] }, force: force)
+        @queue.push({ site: db, user: current_user, db: db, job: blk, desc: desc }, force: force)
       else
         blk.call
       end
@@ -93,7 +101,7 @@ def start_thread
 
     # using non_block to match Ruby #deq
     def do_work(non_block = false)
-      db, job, desc = @queue.shift(block: !non_block)[:task]
+      db, job, desc = @queue.shift(block: !non_block).values_at(:db, :job, :desc)
 
       start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
       db ||= RailsMultisite::ConnectionManagement::DEFAULT

lib/work_queue.rb

@@ -51,15 +51,16 @@ def size
   class FairQueue
     attr_reader :size
 
-    def initialize(limit, &blk)
+    def initialize(key, limit, &blk)
       @limit = limit
       @size = 0
+      @key = key
       @elements = Hash.new { |h, k| h[k] = blk.call }
     end
 
     def push(task, force:)
       raise WorkQueueFull if !force && @size >= @limit
-      key, task = task.values_at(:key, :task)
+      key = task[@key]
       @elements[key].push(task, force: force)
       @size += 1
       nil
@@ -72,10 +73,8 @@ def shift
         task = queue.shift
 
         @elements[key] = queue unless queue.empty?
-
         @size -= 1
-
-        { key: key, task: task }
+        task
       end
     end
 

spec/lib/scheduler/defer_spec.rb

@@ -112,4 +112,40 @@ def wait_for(timeout, &blk)
 
     expect(s).to eq("good")
   end
+
+  describe "#later" do
+    let!(:ivar) { Concurrent::IVar.new }
+    let!(:responses) { Thread::Queue.new }
+
+    def later(db, current_user, request)
+      @defer.later(nil, db, current_user: current_user) do
+        ivar.value
+        responses.push([db, current_user, request])
+      end
+    end
+
+    it "runs jobs in a fair order" do
+      later("site1", 1, 1)
+      later("site1", 1, 2)
+      later("site1", 2, 3)
+      later("site2", 3, 4)
+      later("site2", 4, 5)
+      later("site2", 4, 6)
+
+      ivar.set(nil)
+
+      result = 6.times.map { responses.shift }
+
+      expect(result).to eq(
+        [
+          ["site1", 1, 1],
+          ["site2", 3, 4],
+          ["site1", 2, 3],
+          ["site2", 4, 5],
+          ["site1", 1, 2],
+          ["site2", 4, 6],
+        ],
+      )
+    end
+  end
 end

spec/lib/work_queue_spec.rb

@@ -74,7 +74,7 @@
 
 RSpec.describe WorkQueue::FairQueue do
   subject(:queue) do
-    WorkQueue::FairQueue.new(global_limit) { WorkQueue::BoundedQueue.new(per_key_limit) }
+    WorkQueue::FairQueue.new(:key, global_limit) { WorkQueue::BoundedQueue.new(per_key_limit) }
   end
 
   let(:global_limit) { 5 }