Class: Philiprehberger::Pool::ResourcePool

Inherits:
Object
  • Object
show all
Defined in:
lib/philiprehberger/pool.rb

Instance Method Summary collapse

Constructor Details

#initialize(size:, timeout: 5, idle_timeout: nil, health_check: nil, &factory) ⇒ ResourcePool

Returns a new instance of ResourcePool.

Raises:

  • (ArgumentError) 


16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/philiprehberger/pool.rb', line 16

def initialize(size:, timeout: 5, idle_timeout: nil, health_check: nil, &factory)
  raise ArgumentError, 'size must be positive' unless size.positive?
  raise ArgumentError, 'factory block is required' unless factory

  @size = size
  @timeout = timeout
  @idle_timeout = idle_timeout
  @health_check = health_check
  @factory = factory

  @mutex = Mutex.new
  @condition = ConditionVariable.new
  @available = []
  @created = 0
  @in_use = Set.new
  @waiting = 0
  @shutdown = false
end

Instance Method Details

#checkin(resource) ⇒ Object 



95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/philiprehberger/pool.rb', line 95

def checkin(resource)
  @mutex.synchronize do
    return unless @in_use.delete?(resource)

    if @shutdown
      destroy_resource(resource)
    else
      @available.push(PoolEntry.new(resource: resource, last_used: Time.now))
      @condition.signal
    end
  end
end

#checkout(timeout: nil) ⇒ Object

Raises:



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
 
# File 'lib/philiprehberger/pool.rb', line 46

def checkout(timeout: nil)
  raise ShutdownError, 'pool is shut down' if @shutdown

  effective_timeout = timeout || @timeout
  deadline = Time.now + effective_timeout

  @mutex.synchronize do
    loop do
      raise ShutdownError, 'pool is shut down' if @shutdown

      # Try to get a valid resource from the available pool
      until @available.empty?
        entry = @available.pop

        if idle_expired?(entry)
          destroy_resource(entry.resource)
          next
        end

        if health_check_fails?(entry.resource)
          destroy_resource(entry.resource)
          next
        end

        @in_use.add(entry.resource)
        return entry.resource
      end

      # Create new resource if under capacity
      if @created < @size
        resource = create_resource
        @in_use.add(resource)
        return resource
      end

      # Wait for a resource to become available
      remaining = deadline - Time.now
      raise TimeoutError, "could not obtain resource within #{effective_timeout}s" if remaining <= 0

      @waiting += 1
      begin
        @condition.wait(@mutex, remaining)
      ensure
        @waiting -= 1
      end
    end
  end
end

#clear ⇒ Integer

Destroy every idle (available) resource without changing the pool's configured capacity or shutting it down. Resources currently checked out are left untouched and will be destroyed normally on checkin via the next health-check cycle (or kept if healthy). Useful for credential-rotation flows where every cached connection should be discarded but the pool itself must keep accepting new checkouts.

Returns:

  • (Integer) —

    number of idle resources cleared



207
208
209
210
211
212
213
214
215
216
 
# File 'lib/philiprehberger/pool.rb', line 207

def clear
  @mutex.synchronize do
    raise ShutdownError, 'pool is shut down' if @shutdown

    cleared = @available
    @available = []
    cleared.each { |entry| destroy_resource(entry.resource) }
    cleared.size
  end
end

#drain ⇒ Object 



181
182
183
184
185
186
187
188
189
190
191
192
193
 
# File 'lib/philiprehberger/pool.rb', line 181

def drain
  @mutex.synchronize do
    raise ShutdownError, 'pool is shut down' if @shutdown

    drained = @available.dup
    @available.clear
    @size -= drained.size
    drained.each do |entry|
      entry.resource.close if entry.resource.respond_to?(:close)
    end
    drained.size
  end
end

#prune_idle ⇒ Object 



154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/philiprehberger/pool.rb', line 154

def prune_idle
  @mutex.synchronize do
    raise ShutdownError, 'pool is shut down' if @shutdown
    return 0 unless @idle_timeout

    now = Time.now
    expired, kept = @available.partition { |entry| (now - entry.last_used) > @idle_timeout }
    @available = kept
    expired.each { |entry| destroy_resource(entry.resource) }
    expired.size
  end
end

#shutdown ⇒ Object 



167
168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/philiprehberger/pool.rb', line 167

def shutdown
  @mutex.synchronize do
    @shutdown = true

    @available.each { |entry| destroy_resource(entry.resource) }
    @available.clear

    @in_use.each { |resource| destroy_resource(resource) }
    @in_use.clear

    @condition.broadcast
  end
end

#shutdown? ⇒ Boolean

Returns:

  • (Boolean) 


195
196
197
 
# File 'lib/philiprehberger/pool.rb', line 195

def shutdown?
  @shutdown
end

#size ⇒ Object 



135
136
137
 
# File 'lib/philiprehberger/pool.rb', line 135

def size
  @mutex.synchronize { @size }
end

#stats ⇒ Hash

Snapshot of the pool's internal counters.

Includes the number of threads currently blocked in checkout/with waiting for a resource (:waiting). A non-zero :waiting while in_use == max indicates back-pressure and is a good signal that the pool may need to be sized up.

Returns:

  • (Hash) —

    keys: :size, :available, :in_use, :max, :waiting



116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/philiprehberger/pool.rb', line 116

def stats
  @mutex.synchronize do
    {
      size: @created,
      available: @available.size,
      in_use: @in_use.size,
      max: @size,
      waiting: @waiting
    }
  end
end

#utilization ⇒ Float

Fraction of the pool currently in use.

Returns in_use / max as a Float between 0.0 and 1.0. Useful for dashboards/metrics without needing to unpack #stats. Returns 0.0 when the pool has been shut down or has zero capacity.

Returns:

  • (Float) —

    fraction of capacity in use (0.0..1.0)



146
147
148
149
150
151
152
 
# File 'lib/philiprehberger/pool.rb', line 146

def utilization
  @mutex.synchronize do
    return 0.0 if @size <= 0

    @in_use.size.to_f / @size
  end
end

#waiting ⇒ Integer

Number of threads currently waiting for a resource to become available.

Returns:

  • (Integer) 


131
132
133
 
# File 'lib/philiprehberger/pool.rb', line 131

def waiting
  @mutex.synchronize { @waiting }
end

#with ⇒ Object

Raises:



35
36
37
38
39
40
41
42
43
44
 
# File 'lib/philiprehberger/pool.rb', line 35

def with
  raise ShutdownError, 'pool is shut down' if @shutdown

  resource = checkout
  begin
    yield resource
  ensure
    checkin(resource)
  end
end