Unix Workers

Unix Workers: I like Unix because the operating system already solved process supervision and nobody told the frameworks. A master preforks N plan workers, work arrives on a pipe, SIGTERM means "finish what you hold, then die with dignity", and the master reaps every child by PID and exit status. No supervisor gem, no thread pool config - fork(2), pipe(2), kill(2), wait(2).

source on github

a real captured run

UNIX WORKERS (master 3990, 3 preforked children: 4018, 4021, 4024)

  deploy signal: SIGTERM to all workers (finish what you hold, then exit)

  the reaping (every child accounted for, by pid and exit status):
    pid 4018    exit 0   served 7 job(s)
    pid 4021    exit 0   served 1 job(s)
    pid 4024    exit 0   served 1 job(s)
  total served: 9/9; unserved jobs stay in the pipe for the NEXT fleet

  count what's NOT here: no supervisor gem, no worker heartbeat
  table, no distributed lock. fork gave us isolation (a worker
  segfault kills ONE plan), the shared pipe gave us a work queue
  with kernel-grade load balancing, TERM-then-wait2 gave us
  deploys that finish in-flight work, and each worker's journal
  (flock'd - the process drill proved it) survives its process.
  the operating system is the best framework you already have;
  it's just that its DSL is spelled fork, pipe, kill, and wait.

source

# frozen_string_literal: true

# Unix Workers: I like Unix because the operating system already
# solved process supervision and nobody told the frameworks. A
# master preforks N plan workers, work arrives on a pipe, SIGTERM
# means "finish what you hold, then die with dignity", and the
# master reaps every child by PID and exit status. No supervisor
# gem, no thread pool config - fork(2), pipe(2), kill(2), wait(2).
#
#   bundle exec ruby examples/unix_workers.rb
#
# Runs offline; every process is real, every signal is real.

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

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

WORKERS = 3
JOBS = 9

# Work arrives on a shared pipe: the kernel does the load balancing
# (whichever worker reads first wins - it's a queue because Unix
# says it's a queue)
reader, writer = IO.pipe
results_reader, results_writer = IO.pipe

pids = WORKERS.times.map do |n|
  fork do
    writer.close
    results_reader.close
    draining = false
    trap(class="s">"TERM") { draining = true }

    journal = Agentic:class="y">:ExecutionJournal.new(path: File.join(Dir.tmpdir, class="s">"agentic_worker_#{n}.jsonl"), fsync_every: 10)
    served = 0
    until draining
      line = begin
        reader.read_nonblock(256)
      rescue IO:class="y">:WaitReadable
        sleep(0.005)
        next
      rescue EOFError
        break
      end
      line.split(class="s">"\n").each do |job|
        orchestrator = Agentic:class="y">:PlanOrchestrator.new(lifecycle_hooks: journal.lifecycle_hooks)
        task = Agentic:class="y">:Task.new(description: job, agent_spec: {class="s">"name" => class="s">"w", class="s">"instructions" => class="s">"w"})
        orchestrator.add_task(task, agent: ->(_t) {
          sleep(0.03)
          class="s">"#{job} done"
        })
        orchestrator.execute_plan
        served += 1
      end
    end
    journal.sync
    results_writer.puts JSON.generate({worker: n, pid: Process.pid, served: served})
    exit!(0)
  end
end
reader.close
results_writer.close

puts class="s">"UNIX WORKERS (master #{Process.pid}, #{WORKERS} preforked children: #{pids.join(", class="s">")})"
puts

# Feed the pipe as work actually arrives (paced) - a burst-written
# pipe gets drained by whoever reads first, which is a queue but not
# a fair one; arrival pacing is what lets the whole fleet lift
JOBS.times do |i|
  writer.puts class="s">"job-#{i}"
  sleep(0.025)
end
sleep(0.1) # let the fleet finish chewing

# The deploy: TERM the fleet, then REAP it - by pid, with status
puts class="s">"  deploy signal: SIGTERM to all workers (finish what you hold, then exit)"
pids.each { |pid| Process.kill(class="s">"TERM", pid) }
statuses = pids.map { |pid| Process.wait2(pid) }
writer.close

reports = results_reader.read.lines.map { |l| JSON.parse(l, symbolize_names: true) }.sort_by { |r| r[class="y">:worker] }
puts
puts class="s">"  the reaping (every child accounted for, by pid and exit status):"
statuses.each do |pid, status|
  report = reports.find { |r| r[class="y">:pid] == pid }
  puts format(class="s">"    pid %-7d exit %-3d served %d job(s)", pid, status.exitstatus, report ? report[class="y">:served] : 0)
end
puts format(class="s">"  total served: %d/%d; unserved jobs stay in the pipe for the NEXT fleet", reports.sum { |r| r[class="y">:served] }, JOBS)
puts
puts class="s">"  count what's NOT here: no supervisor gem, no worker heartbeat"
puts class="s">"  table, no distributed lock. fork gave us isolation (a worker"
puts class="s">"  segfault kills ONE plan), the shared pipe gave us a work queue"
puts class="s">"  with kernel-grade load balancing, TERM-then-wait2 gave us"
puts class="s">"  deploys that finish in-flight work, and each worker's journal"
puts class="s">"  (flock'd - the process drill proved it) survives its process."
puts class="s">"  the operating system is the best framework you already have;"
puts class="s">"  it's just that its DSL is spelled fork, pipe, kill, and wait."

clean = statuses.all? { |_, s| s.exitstatus.zero? } && reports.sum { |r| r[class="y">:served] } == JOBS
exit(clean ? 0 : 1)