The Knee Finder

The Knee Finder: runs the same plan at increasing concurrency limits, measures wall time and total queue-wait via the task_slot_acquired hook, and recommends the limit where adding lanes stops paying. Guessing concurrency limits is a superstition; this is a measurement.

source on github

a real captured run

KNEE FINDER: 12 calls, 1200ms of total IO

  limit   wall     total queue-wait
      1   1211ms     6821ms  ################################################
      2    657ms     2904ms  ##########################
      3    474ms     1668ms  ###################
      4    354ms      981ms  ##############
      6    300ms      478ms  ############
      8    300ms      230ms  ############
     12    301ms        0ms  ############

  recommendation: concurrency_limit 6
  (smallest limit within 15% of the best wall time - beyond it you
   hold more connections open to save less time than the jitter)

source

# frozen_string_literal: true

# The Knee Finder: runs the same plan at increasing concurrency limits,
# measures wall time and total queue-wait via the task_slot_acquired
# hook, and recommends the limit where adding lanes stops paying.
# Guessing concurrency limits is a superstition; this is a measurement.
#
#   bundle exec ruby examples/knee_finder.rb
#
# Runs offline; the workload is 12 simulated API calls of mixed latency.

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

# One slow call dominates, as it always does in production
LATENCIES = [0.08, 0.05, 0.12, 0.06, 0.30, 0.05, 0.15, 0.07, 0.05, 0.09, 0.06, 0.12].freeze

def run_at(limit)
  queue_wait = 0.0
  orchestrator = Agentic:class="y">:PlanOrchestrator.new(
    concurrency_limit: limit,
    lifecycle_hooks: {
      task_slot_acquired: ->(task_id:, task:, waited:) { queue_wait += waited }
    }
  )

  LATENCIES.each_with_index do |latency, i|
    orchestrator.add_task(Agentic:class="y">:Task.new(
      description: class="s">"call-#{i}",
      agent_spec: {class="s">"name" => class="s">"api", class="s">"instructions" => class="s">"wait"},
      payload: latency
    ), agent: ->(t) { sleep(t.payload) || class="y">:ok })
  end

  result = orchestrator.execute_plan
  {wall: result.execution_time, queue_wait: queue_wait}
end

puts class="s">"KNEE FINDER: #{LATENCIES.size} calls, #{(LATENCIES.sum * 1000).round}ms of total IO"
puts
puts class="s">"  limit   wall     total queue-wait"

measurements = [1, 2, 3, 4, 6, 8, 12].to_h do |limit|
  m = run_at(limit)
  bar = class="s">"#" * (m[class="y">:wall] * 40).round
  puts format(class="s">"  %5d   %4dms   %6dms  %s", limit, m[class="y">:wall] * 1000, m[class="y">:queue_wait] * 1000, bar)
  [limit, m]
end

# The knee: smallest limit whose wall time is within 15% of the best
best = measurements.values.map { |m| m[class="y">:wall] }.min
knee = measurements.find { |_, m| m[class="y">:wall] <= best * 1.15 }.first

puts
puts class="s">"  recommendation: concurrency_limit #{knee}"
puts class="s">"  (smallest limit within 15% of the best wall time - beyond it you"
puts class="s">"   hold more connections open to save less time than the jitter)"