discard dodgy thrift connections that went into a bad state, made PoolingConnection thread-safe

Author: schiermike

elasticsearch/connection/pooling.py

@@ -1,22 +1,14 @@
+from Queue import Queue
 from .base import Connection
 
+
 class PoolingConnection(Connection):
  def __init__(self, *args, **kwargs):
- self._max_pool_size = kwargs.pop('max_connection_pool_size', 50)
- self._free_connections = []
- self._in_use_connections = set()
+ self._free_connections = Queue()
  super(PoolingConnection, self).__init__(*args, **kwargs)
 
  def _get_connection(self):
- try:
- con = self._free_connections.pop()
- except IndexError:
- con = self._make_connection()
-
- self._in_use_connections.add(con)
- return con
+ return self._make_connection() if self._free_connections.empty() else self._free_connections.get()
 
  def _release_connection(self, con):
- self._in_use_connections.remove(con)
- self._free_connections.append(con)
-
+ self._free_connections.put(con)

elasticsearch/connection/thrift.py

@@ -65,9 +65,9 @@ def perform_request(self, method, url, params=None, body=None, timeout=None, ign
  except (TException, SocketTimeout) as e:
  self.log_request_fail(method, url, body, time.time() - start, exception=e)
  raise ConnectionError('N/A', str(e), e)
- finally:
-  if tclient:
-  self._release_connection(tclient)
+
+ if tclient:
+ self._release_connection(tclient)
 
  if not (200 <= response.status < 300) and response.status not in ignore:
  self.log_request_fail(method, url, body, duration, response.status)