Class: ThrottleMachines::HedgedRequest

Inherits:
Object
  • Object
show all
Defined in:
lib/throttle_machines/hedged_request.rb

Overview

Hedged Request - Multi-path Navigation System

Like sending scout ships on multiple routes to find the fastest path. The first ship to reach the destination wins, others are recalled.

Reduces latency by racing multiple backends/attempts with staggered delays.

Example:

hedged = ThrottleMachines::HedgedRequest.new(
delay: 0.05,     # 50ms between attempts
max_attempts: 3
)

result = hedged.run do |attempt|
case attempt
when 0 then primary_backend.get(key)
when 1 then secondary_backend.get(key)
when 2 then tertiary_backend.get(key)
end
end

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(delay: 0.05, max_attempts: 2, timeout: nil) ⇒ HedgedRequest

Returns a new instance of HedgedRequest.



29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/throttle_machines/hedged_request.rb', line 29

def initialize(delay: 0.05, max_attempts: 2, timeout: nil)
  @delay = delay
  @max_attempts = max_attempts
  @timeout = timeout
  @executor = Concurrent::ThreadPoolExecutor.new(
    min_threads: 1,
    max_threads: max_attempts,
    max_queue: max_attempts,
    fallback_policy: :caller_runs
  )
end

Instance Attribute Details

#delay ⇒ Object (readonly)

Returns the value of attribute delay.



27
28
29
 
# File 'lib/throttle_machines/hedged_request.rb', line 27

def delay
  @delay
end

#max_attempts ⇒ Object (readonly)

Returns the value of attribute max_attempts.



27
28
29
 
# File 'lib/throttle_machines/hedged_request.rb', line 27

def max_attempts
  @max_attempts
end

#timeout ⇒ Object (readonly)

Returns the value of attribute timeout.



27
28
29
 
# File 'lib/throttle_machines/hedged_request.rb', line 27

def timeout
  @timeout
end

Instance Method Details

#run(&block) ⇒ Object

Run hedged request with automatic cancellation of slower attempts



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/throttle_machines/hedged_request.rb', line 42

def run(&block)
  raise ArgumentError, 'Block required' unless block

  # Generate a unique request ID for tracking
  request_id = "#{object_id}-#{Time.now.to_f}"

  # Instrument the start of the hedged request
  Instrumentation.hedged_request_started(request_id, attempts: @max_attempts)

  # Use Concurrent::Promises for better async handling
  futures = []
  first_result = Concurrent::Promises.resolvable_future
  start_time = Time.now.to_f

  @max_attempts.times do |attempt|
    # Schedule with delay
    future = if attempt.zero?
               Concurrent::Promises.future { yield(attempt) }
             else
               Concurrent::Promises.schedule(@delay * attempt) { yield(attempt) }
             end

    # Race to resolve first_result
    future.then do |result|
      if !first_result.resolved? && first_result.fulfill(result)
        # This attempt won the race
        duration = Time.now.to_f - start_time
        Instrumentation.hedged_request_winner(request_id, attempt: attempt, duration: duration)
      end
      result
    end.rescue do |error|
      # Only reject if this was the last attempt and nothing succeeded
      first_result.reject(error) if attempt == @max_attempts - 1 && !first_result.resolved?
    end

    futures << future
  end

  # Wait with optional timeout
  if @timeout
    # Use any_resolved_future with timeout
    timeout_future = Concurrent::Promises.schedule(@timeout) do
      raise TimeoutError, "Hedged request timed out after #{@timeout}s"
    end

    Concurrent::Promises.any_resolved_future(first_result, timeout_future).value!
  else
    first_result.value!
  end
ensure
  # Cancel pending futures
  futures.each { |f| f.cancel if f.pending? }
end

#run_async(&block) ⇒ Object

Run async version



97
98
99
 
# File 'lib/throttle_machines/hedged_request.rb', line 97

def run_async(&block)
  Concurrent::Promises.future { run(&block) }
end

#shutdown ⇒ Object

Shutdown the executor



102
103
104
105
 
# File 'lib/throttle_machines/hedged_request.rb', line 102

def shutdown
  @executor.shutdown
  @executor.wait_for_termination(5)
end