Class: ThrottleMachines::AsyncLimiter

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

Overview

Async-aware rate limiter for fiber-safe operations

Like quantum entanglement communications - allows multiple spacecraft to communicate simultaneously without interference, each in their own quantum state (fiber).

Example:

limiter = ThrottleMachines::AsyncLimiter.new("quantum_comms",
limit: 100,
period: 60,
algorithm: :gcra
)

Async do
if limiter.allowed_async?
  # Non-blocking operation
end
end

Instance Attribute Summary

Attributes inherited from Limiter

#algorithm, #key, #limit, #period, #storage

Instance Method Summary collapse

Methods inherited from Limiter

#allow?, #remaining, #reset!, #retry_after, #throttle!, #to_h

Constructor Details

#initialize(key, limit:, period:, algorithm: :fixed_window, storage: nil) ⇒ AsyncLimiter

Returns a new instance of AsyncLimiter.



25
26
27
28
 
# File 'lib/throttle_machines/async_limiter.rb', line 25

def initialize(key, limit:, period:, algorithm: :fixed_window, storage: nil)
  super
  @fiber_storage = Concurrent::Map.new # Thread-safe fiber storage
end

Instance Method Details

#allowed_async? ⇒ Boolean

Async version of allowed? that's fiber-safe

Returns:

  • (Boolean) 


31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/throttle_machines/async_limiter.rb', line 31

def allowed_async?
  allowed = if defined?(Async::Task) && Async::Task.current?
              # In async context, use fiber-local checking
              fiber_allowed?
            else
              # Fall back to regular synchronous check
              allowed?
            end

  # Don't double-instrument when calling parent allowed?
  return allowed unless defined?(Async::Task) && Async::Task.current?

  # Instrument the async check
  Instrumentation.rate_limit_checked(self, allowed: allowed, remaining: nil)
  allowed
end

#check_async ⇒ Object

Non-blocking check with async support



49
50
51
52
53
54
55
56
 
# File 'lib/throttle_machines/async_limiter.rb', line 49

def check_async
  if allowed_async?
    yield if block_given?
    true
  else
    false
  end
end

#cleanup_fiber_storage ⇒ Object

Clean up fiber storage periodically



97
98
99
100
101
102
 
# File 'lib/throttle_machines/async_limiter.rb', line 97

def cleanup_fiber_storage
  current_fibers = ObjectSpace.each_object(Fiber).map(&:object_id)
  @fiber_storage.each_key do |fiber_id|
    @fiber_storage.delete(fiber_id) unless current_fibers.include?(fiber_id)
  end
end

#fiber_state ⇒ Object

Get current fiber's state



86
87
88
89
90
91
92
93
94
 
# File 'lib/throttle_machines/async_limiter.rb', line 86

def fiber_state
  fiber_id = Fiber.current.object_id
  @fiber_storage.compute_if_absent(fiber_id) do
    {
      last_check: 0,
      tokens: @limit.to_f
    }
  end
end

#throttle_async(max_wait: nil) ⇒ Object

Async throttle with automatic retry



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
 
# File 'lib/throttle_machines/async_limiter.rb', line 59

def throttle_async(max_wait: nil)
  start_time = current_time

  loop do
    if allowed_async?
      return yield if block_given?

      return true
    end

    wait_time = retry_after

    # Check if we've exceeded max wait time
    if max_wait && (current_time - start_time + wait_time) > max_wait
      raise ThrottleError, 'Maximum wait time exceeded'
    end

    # Non-blocking sleep in async context
    if defined?(Async::Task) && Async::Task.current?
      Async::Task.current.sleep(wait_time)
    else
      sleep(wait_time)
    end
  end
end