Class: Upfluence::Pool

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

Defined Under Namespace

Classes: NoInstanciationBlock, NullMetrics, PrometheusMetrics, UnknownResource

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(size = 0, options = {}, &block) ⇒ Pool

Returns a new instance of Pool.

Raises:



82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/upfluence/pool.rb', line 82

def initialize(size = 0, options = {}, &block)
  raise NoInstanciationBlock unless block

  @create_block = block
  @redeemed = []
  @que = []
  @max = size
  @mutex = Mutex.new
  @resource = ConditionVariable.new
  @shutdown_block = nil
  @timeout = options.fetch :timeout, 5
  @metrics = options.fetch(:metrics, NullMetrics.new)

  @metrics.set_size(size)
end

Instance Attribute Details

#max ⇒ Object (readonly)

Returns the value of attribute max.



80
81
82
 
# File 'lib/upfluence/pool.rb', line 80

def max
  @max
end

Instance Method Details

#discard(obj) ⇒ Object 



98
99
100
101
102
103
104
105
106
 
# File 'lib/upfluence/pool.rb', line 98

def discard(obj)
  @metrics.observe_discard

  @mutex.synchronize do
    remove_from_redeemed obj
    @metrics.update_checkout(@redeemed.length)
    @resource.broadcast
  end
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


155
156
157
 
# File 'lib/upfluence/pool.rb', line 155

def empty?
  (@redeemed.length - @que.length) >= @max
end

#length ⇒ Object 



159
160
161
 
# File 'lib/upfluence/pool.rb', line 159

def length
  @max - @redeemed.length + @que.length
end

#pop(options = {}) ⇒ Object 



121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/upfluence/pool.rb', line 121

def pop(options = {})
  timeout = options.fetch :timeout, @timeout
  effective = [timeout, Upfluence.context.timeout].compact.min
  deadline = Time.now + effective
  t0 = nil

  @metrics.observe_get

  @mutex.synchronize do
    loop do
      unless @que.empty?
        result = redeem(@que.pop)
        @metrics.update_idle(@que.length)
        @metrics.observe_wait(Time.now - t0) if t0
        return result
      end

      connection = try_create(options)

      if connection
        result = redeem(connection)
        @metrics.observe_wait(Time.now - t0) if t0
        return result
      end

      to_wait = deadline - Time.now
      raise ::Timeout::Error.new("Waited #{timeout} sec") if to_wait <= 0

      t0 ||= Time.now
      @resource.wait(@mutex, to_wait)
    end
  end
end

#push(obj) ⇒ Object 



108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/upfluence/pool.rb', line 108

def push(obj)
  @metrics.observe_put

  @mutex.synchronize do
    remove_from_redeemed obj

    @que.push obj
    @metrics.update_idle(@que.length)
    @metrics.update_checkout(@redeemed.length)
    @resource.broadcast
  end
end