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
|
# Copyright 2013 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import abc
import collections
import sys
import threading
from oslo_log import log as logging
from oslo_utils import timeutils
from oslo_messaging._drivers import common
LOG = logging.getLogger(__name__)
# TODO(harlowja): remove this when we no longer have to support 2.7
if sys.version_info[0:2] < (3, 2):
def wait_condition(cond):
# FIXME(markmc): timeout needed to allow keyboard interrupt
# http://bugs.python.org/issue8844
cond.wait(timeout=1)
else:
def wait_condition(cond):
cond.wait()
class Pool(object, metaclass=abc.ABCMeta):
"""A thread-safe object pool.
Modelled after the eventlet.pools.Pool interface, but designed to be safe
when using native threads without the GIL.
Resizing is not supported.
"""
def __init__(self, max_size=4, min_size=2, ttl=1200, on_expire=None):
super(Pool, self).__init__()
self._min_size = min_size
self._max_size = max_size
self._item_ttl = ttl
self._current_size = 0
self._cond = threading.Condition()
self._items = collections.deque()
self._on_expire = on_expire
def expire(self):
"""Remove expired items from left (the oldest item) to
right (the newest item).
"""
with self._cond:
while len(self._items) > self._min_size:
try:
ttl_watch, item = self._items.popleft()
if ttl_watch.expired():
self._on_expire and self._on_expire(item)
self._current_size -= 1
else:
self._items.appendleft((ttl_watch, item))
return
except IndexError:
break
def put(self, item):
"""Return an item to the pool."""
with self._cond:
ttl_watch = timeutils.StopWatch(duration=self._item_ttl)
ttl_watch.start()
self._items.append((ttl_watch, item))
self._cond.notify()
def get(self, retry=None):
"""Return an item from the pool, when one is available.
This may cause the calling thread to block.
"""
with self._cond:
while True:
try:
ttl_watch, item = self._items.pop()
self.expire()
return item
except IndexError:
pass
if self._current_size < self._max_size:
self._current_size += 1
break
LOG.warning("Connection pool limit exceeded: "
"current size %s surpasses max "
"configured rpc_conn_pool_size %s",
self._current_size, self._max_size)
wait_condition(self._cond)
# We've grabbed a slot and dropped the lock, now do the creation
try:
return self.create(retry=retry)
except Exception:
with self._cond:
self._current_size -= 1
raise
def iter_free(self):
"""Iterate over free items."""
while True:
try:
_, item = self._items.pop()
yield item
except IndexError:
return
@abc.abstractmethod
def create(self, retry=None):
"""Construct a new item."""
class ConnectionPool(Pool):
"""Class that implements a Pool of Connections."""
def __init__(self, conf, max_size, min_size, ttl, url, connection_cls):
self.connection_cls = connection_cls
self.conf = conf
self.url = url
super(ConnectionPool, self).__init__(max_size, min_size, ttl,
self._on_expire)
def _on_expire(self, connection):
connection.close()
LOG.debug("Idle connection has expired and been closed."
" Pool size: %d" % len(self._items))
def create(self, purpose=common.PURPOSE_SEND, retry=None):
LOG.debug('Pool creating new connection')
return self.connection_cls(self.conf, self.url, purpose, retry=retry)
def empty(self):
for item in self.iter_free():
item.close()
|
