1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
|
##
# The TimedStack manages a pool of homogeneous connections (or any resource
# you wish to manage). Connections are created lazily up to a given maximum
# number.
# Examples:
#
# ts = TimedStack.new(1) { MyConnection.new }
#
# # fetch a connection
# conn = ts.pop
#
# # return a connection
# ts.push conn
#
# conn = ts.pop
# ts.pop timeout: 5
# #=> raises Bundler::ConnectionPool::TimeoutError after 5 seconds
class Bundler::ConnectionPool::TimedStack
attr_reader :max
##
# Creates a new pool with +size+ connections that are created from the given
# +block+.
def initialize(size = 0, &block)
@create_block = block
@created = 0
@que = []
@max = size
@mutex = Thread::Mutex.new
@resource = Thread::ConditionVariable.new
@shutdown_block = nil
end
##
# Returns +obj+ to the stack. +options+ is ignored in TimedStack but may be
# used by subclasses that extend TimedStack.
def push(obj, options = {})
@mutex.synchronize do
if @shutdown_block
@shutdown_block.call(obj)
else
store_connection obj, options
end
@resource.broadcast
end
end
alias << push
##
# Retrieves a connection from the stack. If a connection is available it is
# immediately returned. If no connection is available within the given
# timeout a Bundler::ConnectionPool::TimeoutError is raised.
#
# +:timeout+ is the only checked entry in +options+ and is preferred over
# the +timeout+ argument (which will be removed in a future release). Other
# options may be used by subclasses that extend TimedStack.
def pop(timeout = 0.5, options = {})
options, timeout = timeout, 0.5 if Hash === timeout
timeout = options.fetch :timeout, timeout
deadline = current_time + timeout
@mutex.synchronize do
loop do
raise Bundler::ConnectionPool::PoolShuttingDownError if @shutdown_block
return fetch_connection(options) if connection_stored?(options)
connection = try_create(options)
return connection if connection
to_wait = deadline - current_time
raise Bundler::ConnectionPool::TimeoutError, "Waited #{timeout} sec" if to_wait <= 0
@resource.wait(@mutex, to_wait)
end
end
end
##
# Shuts down the TimedStack by passing each connection to +block+ and then
# removing it from the pool. Attempting to checkout a connection after
# shutdown will raise +Bundler::ConnectionPool::PoolShuttingDownError+ unless
# +:reload+ is +true+.
def shutdown(reload: false, &block)
raise ArgumentError, "shutdown must receive a block" unless block_given?
@mutex.synchronize do
@shutdown_block = block
@resource.broadcast
shutdown_connections
@shutdown_block = nil if reload
end
end
##
# Returns +true+ if there are no available connections.
def empty?
(@created - @que.length) >= @max
end
##
# The number of connections available on the stack.
def length
@max - @created + @que.length
end
private
def current_time
Process.clock_gettime(Process::CLOCK_MONOTONIC)
end
##
# This is an extension point for TimedStack and is called with a mutex.
#
# This method must returns true if a connection is available on the stack.
def connection_stored?(options = nil)
!@que.empty?
end
##
# This is an extension point for TimedStack and is called with a mutex.
#
# This method must return a connection from the stack.
def fetch_connection(options = nil)
@que.pop
end
##
# This is an extension point for TimedStack and is called with a mutex.
#
# This method must shut down all connections on the stack.
def shutdown_connections(options = nil)
while connection_stored?(options)
conn = fetch_connection(options)
@shutdown_block.call(conn)
end
@created = 0
end
##
# This is an extension point for TimedStack and is called with a mutex.
#
# This method must return +obj+ to the stack.
def store_connection(obj, options = nil)
@que.push obj
end
##
# This is an extension point for TimedStack and is called with a mutex.
#
# This method must create a connection if and only if the total number of
# connections allowed has not been met.
def try_create(options = nil)
unless @created == @max
object = @create_block.call
@created += 1
object
end
end
end
|
