Fix execution context queue stress tests failures (#16472)

Refactors and abstracts the stress test specs of `Fiber::ExecutionContext::Runnables` and `Fiber::ExecutionContext::GlobalQueue` so no loop will run forever: the "thread setup" part has been removed, the main fiber won't block waiting for the threads to be ready, and the threads' loop will eventually timeout, and the thread return, so the main fiber won't block while joining the threads.

I abstracted a helper because the different tests used the same structure, and it was painful & noisy to dup the logic.

This fixes the regular CI failures that often occurred on Darwin on CI, and that I just reproduced on Linux when running these specs in tight loops multiple times in parallel to overload the CPU cores.

Co-authored-by: Johannes Müller <[email protected]>

Author: ysbaddaden

spec/std/fiber/execution_context/global_queue_spec.cr

@@ -91,131 +91,112 @@ describe Fiber::ExecutionContext::GlobalQueue do
   # interpreter doesn't support threads yet (#14287)
   pending_interpreted describe: "thread safety" do
     it "one by one" do
+      {% if flag?(:win32) && flag?(:aarch64) %}
+        pending! "CI/WIN32/CLANGARM64 always fails"
+      {% end %}
+
       fibers = StaticArray(Fiber::ExecutionContext::FiberCounter, 763).new do |i|
         Fiber::ExecutionContext::FiberCounter.new(new_fake_fiber("f#{i}"))
       end
 
       n = 7
       increments = 15
+
       queue = Fiber::ExecutionContext::GlobalQueue.new(Thread::Mutex.new)
-      ready = Thread::WaitGroup.new(n)
-
-      threads = Array.new(n) do |i|
-        new_thread("ONE-#{i}") do
-          slept = 0
-          ready.done
-
-          loop do
-            if fiber = queue.pop?
-              fc = fibers.find! { |x| x.@fiber == fiber }
-              queue.push(fiber) if fc.increment < increments
-              slept = 0
-            elsif slept < 100
-              slept += 1
-              Thread.sleep(1.nanosecond) # don't burn CPU
-            else
-              break
-            end
-          end
-        end
-      end
-      ready.wait
 
-      fibers.each_with_index do |fc, i|
-        queue.push(fc.@fiber)
-        Thread.sleep(10.nanoseconds) if i % 10 == 9
-      end
+      Fiber::ExecutionContext.stress_test(
+        n,
+        iteration: ->(i : Int32) {
+          if fiber = queue.pop?
+            fc = fibers.find! { |x| x.@fiber == fiber }
+            queue.push(fiber) if fc.increment < increments
+            return :next
+          end
 
-      threads.each(&.join)
+          # done?
+          if fibers.all? { |fc| fc.counter >= increments }
+            return :break
+          end
+        },
+        publish: -> {
+          fibers.each_with_index do |fc, i|
+            queue.push(fc.@fiber)
+            Thread.sleep(10.nanoseconds) if i % 10 == 9
+          end
+        },
+      )
 
       # must have dequeued each fiber exactly X times
       fibers.each { |fc| fc.counter.should eq(increments) }
     end
 
-    {% if flag?(:darwin) %}
-      # FIXME: the spec regularly fails on macOS with "expected 15 got 0"
-      pending "bulk operations"
-    {% else %}
-      it "bulk operations" do
-        n = 7
-        increments = 15
+    it "bulk operations" do
+      n = 7
+      increments = 15
 
-        fibers = StaticArray(Fiber::ExecutionContext::FiberCounter, 765).new do |i| # 765 can be divided by 3 and 5
-          Fiber::ExecutionContext::FiberCounter.new(new_fake_fiber("f#{i}"))
-        end
+      fibers = StaticArray(Fiber::ExecutionContext::FiberCounter, 765).new do |i| # 765 can be divided by 3 and 5
+        Fiber::ExecutionContext::FiberCounter.new(new_fake_fiber("f#{i}"))
+      end
 
-        queue = Fiber::ExecutionContext::GlobalQueue.new(Thread::Mutex.new)
-        ready = Thread::WaitGroup.new(n)
-
-        threads = Array.new(n) do |i|
-          new_thread("BULK-#{i}") do
-            slept = 0
-
-            r = Fiber::ExecutionContext::Runnables(3).new(queue)
-
-            batch = Fiber::List.new
-            size = 0
-
-            reenqueue = -> {
-              if size > 0
-                queue.bulk_push(pointerof(batch))
-                names = [] of String?
-                batch.each { |f| names << f.name }
-                batch.clear
-                size = 0
-              end
-            }
-
-            execute = ->(fiber : Fiber) {
-              fc = fibers.find! { |x| x.@fiber == fiber }
-
-              if fc.increment < increments
-                batch.push(fc.@fiber)
-                size += 1
-              end
-            }
-
-            ready.done
-
-            loop do
-              if fiber = r.shift?
-                execute.call(fiber)
-                slept = 0
-                next
-              end
-
-              if fiber = queue.grab?(r, 1)
-                reenqueue.call
-                execute.call(fiber)
-                slept = 0
-                next
-              end
-
-              if slept >= 100
-                break
-              end
-
-              reenqueue.call
-              slept += 1
-              Thread.sleep(1.nanosecond) # don't burn CPU
-            end
-          end
+      queue = Fiber::ExecutionContext::GlobalQueue.new(Thread::Mutex.new)
+
+      runnables = Array.new(n) { Fiber::ExecutionContext::Runnables(3).new(queue) }
+      batches = Array.new(n) { Fiber::List.new }
+
+      reenqueue = ->(batch : Pointer(Fiber::List)) {
+        if batch.value.size > 0
+          queue.bulk_push(batch)
+          names = [] of String?
+          batch.value.each { |f| names << f.name }
+          batch.value.clear
         end
-        ready.wait
-
-        # enqueue in batches of 5
-        0.step(to: fibers.size - 1, by: 5) do |i|
-          list = Fiber::List.new
-          5.times { |j| list.push(fibers[i + j].@fiber) }
-          queue.bulk_push(pointerof(list))
-          Thread.sleep(10.nanoseconds) if i % 4 == 3
+      }
+
+      execute = ->(fiber : Fiber, batch : Pointer(Fiber::List)) {
+        fc = fibers.find! { |x| x.@fiber == fiber }
+
+        if fc.increment < increments
+          batch.value.push(fc.@fiber)
         end
+      }
 
-        threads.each(&.join)
+      Fiber::ExecutionContext.stress_test(
+        n,
+        iteration: ->(i : Int32) {
+          r = runnables[i]
+          batch = batches.to_unsafe + i
 
-        # must have dequeued each fiber exactly X times (no less, no more)
-        fibers.each { |fc| fc.counter.should eq(increments) }
-      end
-    {% end %}
+          if fiber = r.shift?
+            execute.call(fiber, batch)
+            return :next
+          end
+
+          if fiber = queue.grab?(r, 1)
+            reenqueue.call(batch)
+            execute.call(fiber, batch)
+            return :next
+          end
+
+          # done?
+          if fibers.all? { |fc| fc.counter >= increments }
+            return :break
+          end
+
+          reenqueue.call(batch)
+        },
+        publish: -> {
+          # enqueue in batches of 5
+          0.step(to: fibers.size - 1, by: 5) do |i|
+            list = Fiber::List.new
+            5.times { |j| list.push(fibers[i + j].@fiber) }
+            queue.bulk_push(pointerof(list))
+            Thread.sleep(10.nanoseconds) if i % 4 == 3
+          end
+        }
+      )
+
+      # must have dequeued each fiber exactly X times (no less, no more)
+      fibers.each { |fc| fc.counter.should eq(increments) }
+    end
   end
 end

spec/std/fiber/execution_context/runnables_spec.cr

@@ -216,68 +216,62 @@ describe Fiber::ExecutionContext::Runnables do
       end
 
       global_queue = Fiber::ExecutionContext::GlobalQueue.new(Thread::Mutex.new)
-      ready = Thread::WaitGroup.new(n)
 
       all_runnables = Array(Fiber::ExecutionContext::Runnables(16)).new(n) do
         Fiber::ExecutionContext::Runnables(16).new(global_queue)
       end
 
-      threads = Array.new(n) do |i|
-        new_thread("RUN-#{i}") do
+      all_randoms = Array.new(n) { Random.split }
+
+      execute = ->(fiber : Fiber, runnables : Fiber::ExecutionContext::Runnables(16)) {
+        fc = fibers.find! { |x| x.@fiber == fiber }
+        runnables.push(fiber) if fc.increment < increments
+      }
+
+      Fiber::ExecutionContext.stress_test(
+        n,
+        iteration: ->(i : Int32) {
           runnables = all_runnables[i]
-          slept = 0
-
-          execute = ->(fiber : Fiber) {
-            fc = fibers.find! { |x| x.@fiber == fiber }
-            runnables.push(fiber) if fc.increment < increments
-          }
-
-          ready.done
-
-          loop do
-            # dequeue from local queue
-            if fiber = runnables.shift?
-              execute.call(fiber)
-              slept = 0
-              next
-            end
-
-            # steal from another queue
-            while (r = all_runnables.sample) == runnables
-            end
-            if fiber = runnables.steal_from(r)
-              execute.call(fiber)
-              slept = 0
-              next
-            end
-
-            # dequeue from global queue
-            if fiber = global_queue.grab?(runnables, n)
-              execute.call(fiber)
-              slept = 0
-              next
-            end
-
-            if slept >= 100
-              break
-            end
-
-            slept += 1
-            Thread.sleep(1.nanosecond) # don't burn CPU
+          random = all_randoms[i]
+
+          # dequeue from local queue
+          if fiber = runnables.shift?
+            execute.call(fiber, runnables)
+            return :next
+          end
+
+          # steal from another queue
+          j = 0
+          while (r = all_runnables.sample(random)) == runnables
+            next if (j += 1) < 1000
+            raise "FATAL: all_runnables.sample returned the local queue 1000 times!"
+          end
+          if fiber = runnables.steal_from(r)
+            execute.call(fiber, runnables)
+            return :next
+          end
+
+          # dequeue from global queue
+          if fiber = global_queue.grab?(runnables, n)
+            execute.call(fiber, runnables)
+            return :next
           end
-        end
-      end
-      ready.wait
-
-      # enqueue in batches
-      0.step(to: fibers.size - 1, by: 9) do |i|
-        list = Fiber::List.new
-        9.times { |j| list.push(fibers[i + j].@fiber) }
-        global_queue.bulk_push(pointerof(list))
-        Thread.sleep(10.nanoseconds) if i % 2 == 1
-      end
 
-      threads.each(&.join)
+          # done?
+          if fibers.all? { |fc| fc.counter >= increments }
+            return :break
+          end
+        },
+        publish: -> {
+          # enqueue in batches of 9
+          0.step(to: fibers.size - 1, by: 9) do |i|
+            list = Fiber::List.new
+            9.times { |j| list.push(fibers[i + j].@fiber) }
+            global_queue.bulk_push(pointerof(list))
+            Thread.sleep(10.nanoseconds) if i % 2 == 1
+          end
+        },
+      )
 
       # must have dequeued each fiber exactly X times (no less, no more)
       fibers.each { |fc| fc.counter.should eq(increments) }

spec/std/fiber/execution_context/spec_helper.cr

@@ -19,4 +19,52 @@ module Fiber::ExecutionContext
       @counter.get(:relaxed)
     end
   end
+
+  # Runs a multithreaded test by starting *n* threads, waiting for all the
+  # threads to have been started, then runs the *publish* proc.
+  #
+  # Each thread calls *iteration* until the timeout is reached or the proc
+  # returns `:break`; if the proc returns `:next` the thread goes immediately to
+  # the next iteration, other it will ease the CPU before the next iteration.
+  #
+  # Returns after every thread has been joined.
+  def self.stress_test(n, *, iteration, publish, name = "STRESS", timeout = 1.second)
+    ready = Thread::WaitGroup.new(n)
+
+    threads = Array.new(n) do |i|
+      new_thread("#{name}-#{i}") do
+        ready.done
+
+        started = Time.monotonic
+        attempts = 0
+
+        iter = 0
+        while iter += 1
+          if iter % 100 == 99 && (Time.monotonic - started) >= timeout
+            # reached timeout: abort
+            break
+          end
+
+          case iteration.call(i)
+          when :next
+            attempts = 0
+            next
+          when :break
+            break
+          else
+            # don't burn CPU
+            attempts = Thread.delay(attempts)
+          end
+        end
+      end
+    end
+
+    ready.wait(timeout * 2) do
+      raise "timeout while waiting for threads to be ready"
+    end
+
+    publish.call
+
+    threads.each(&.join)
+  end
 end