The Threads Drill

The Threads Drill: fibers are polite; threads are not. Everything in this gem that claims to be shared-safe gets hammered by real Ruby threads - the kind that run truly parallel on JRuby, where there is no GVL to be your accidental bodyguard. The journal and registry hold. The windowed limiter's bookkeeping is the one to watch, and the drill says so out loud.

source on github

a real captured run

  drill 1 - journal, 8 threads x 150 events:
    1200/1200 lines written, 0 torn - mutex + flock + fsync held

  drill 2 - registry, 8 threads x 50 register+execute:
    0 registrations lost - registry held

  drill 3 - windowed try_acquire, 8 threads x 200 attempts (ceiling 50):
    admitted 50/50 - the stamp bookkeeping holds a real Mutex now

  all three structures now hold under real threads for the right
  reason: real locks (two Mutexes, flock, fsync), not scheduling
  luck. this drill went from characterization to CERTIFICATION
  when the round-12 release paid the limiter's lock debt - the
  answer is now the same on every Ruby, which is the only kind
  of thread-safety worth writing in a README.

source

# frozen_string_literal: true

# The Threads Drill: fibers are polite; threads are not. Everything
# in this gem that claims to be shared-safe gets hammered by real
# Ruby threads - the kind that run truly parallel on JRuby, where
# there is no GVL to be your accidental bodyguard. The journal and
# registry hold. The windowed limiter's bookkeeping is the one to
# watch, and the drill says so out loud.
#
#   bundle exec ruby examples/threads_drill.rb
#
# Runs offline; exits 1 if a guaranteed-safe structure corrupts.

require class="s">"bundler/setup"
require class="s">"agentic"
require class="s">"tmpdir"
require class="s">"json"

Agentic.logger.level = class="y">:fatal

THREADS = 8
EVENTS = 150
failures = 0

# --- drill 1: the journal under parallel writers --------------------------------
path = File.join(Dir.tmpdir, class="s">"agentic_threads.journal.jsonl")
File.delete(path) if File.exist?(path)
journal = Agentic:class="y">:ExecutionJournal.new(path: path)

THREADS.times.map { |t|
  Thread.new do
    EVENTS.times { |i| journal.record(class="y">:task_succeeded, task_id: class="s">"t#{t}-#{i}", description: class="s">"t#{t}-#{i}", duration: 0.001, output: class="s">"x" * 64) }
  end
}.each(&class="y">:join)

lines = File.readlines(path)
parseable = lines.count do |line|
  JSON.parse(line)
  true
rescue JSON:class="y">:ParserError
  false
end
torn = lines.size - parseable
expected = THREADS * EVENTS
failures += 1 if lines.size != expected || torn.positive?
puts class="s">"  drill 1 - journal, #{THREADS} threads x #{EVENTS} events:"
puts format(class="s">"    %d/%d lines written, %d torn - %s", parseable, expected, torn,
  (torn.zero? && lines.size == expected) ? class="s">"mutex + flock + fsync held" : class="s">"CORRUPTED")
puts

# --- drill 2: the registry under concurrent registration ------------------------
registry = Agentic:class="y">:AgentCapabilityRegistry.instance
THREADS.times.map { |t|
  Thread.new do
    50.times do |i|
      spec = Agentic:class="y">:CapabilitySpecification.new(
        name: class="s">"cap-#{t}-#{i}", description: class="s">"x", version: class="s">"1.0.0",
        inputs: {a: {type: class="s">"number", required: true}}
      )
      Agentic.register_capability(spec, Agentic:class="y">:CapabilityProvider.new(capability: spec, implementation: ->(inputs) { inputs }))
      registry.get_provider(class="s">"cap-#{t}-#{i}")&.execute(a: 1)
    end
  end
}.each(&class="y">:join)

missing = THREADS.times.sum { |t| 50.times.count { |i| registry.get_provider(class="s">"cap-#{t}-#{i}").nil? } }
failures += 1 if missing.positive?
puts class="s">"  drill 2 - registry, #{THREADS} threads x 50 register+execute:"
puts format(class="s">"    %d registrations lost - %s", missing, missing.zero? ? class="s">"registry held" : class="s">"RACE")
puts

# --- drill 3: the windowed limiter's check-then-act ------------------------------
# In round 11 this bookkeeping had no mutex and the drill called it
# "luck wearing a lab coat". The round-12 release put a real Mutex
# around the stamp dance, so the drill now ASSERTS what it could
# previously only observe.
limit = Agentic:class="y">:RateLimit.new(50, per: 60)
admitted = THREADS.times.map {
  Thread.new { 200.times.count { limit.try_acquire } }
}.map(&class="y">:value).sum

failures += 1 if admitted != 50
puts class="s">"  drill 3 - windowed try_acquire, #{THREADS} threads x 200 attempts (ceiling 50):"
puts format(class="s">"    admitted %d/50 - %s", admitted,
  (admitted == 50) ? class="s">"the stamp bookkeeping holds a real Mutex now" : class="s">"OVER-ADMISSION - the lock is gone")
puts
puts class="s">"  all three structures now hold under real threads for the right"
puts class="s">"  reason: real locks (two Mutexes, flock, fsync), not scheduling"
puts class="s">"  luck. this drill went from characterization to CERTIFICATION"
puts class="s">"  when the round-12 release paid the limiter's lock debt - the"
puts class="s">"  answer is now the same on every Ruby, which is the only kind"
puts class="s">"  of thread-safety worth writing in a README."

exit(failures.zero? ? 0 : 1)