diff --git a/elasticsearch/connection_pool.py b/elasticsearch/connection_pool.py index c68a7a3b..c443d665 100644 --- a/elasticsearch/connection_pool.py +++ b/elasticsearch/connection_pool.py @@ -83,12 +83,15 @@ class ConnectionPool(object): live pool. A connection that has been peviously marked as dead and succeedes will be marked as live (it's fail count will be deleted). """ - def __init__(self, connections, dead_timeout=60, selector_class=RoundRobinSelector, randomize_hosts=True, **kwargs): + def __init__(self, connections, dead_timeout=60, timeout_cutoff=5, + selector_class=RoundRobinSelector, randomize_hosts=True, **kwargs): """ :arg connections: list of tuples containing the :class:`~elasticsearch.Connection` instance and it's options :arg dead_timeout: number of seconds a connection should be retired for - after a failure + after a failure, increases on consecutive failures + :arg timeout_cutoff: number of consecutive failures after which the + timeout doesn't increase :arg selector_class: :class:`~elasticsearch.ConnectionSelector` subclass to use :arg randomize_hosts: shuffle the list of connections upon arrival to @@ -107,6 +110,7 @@ class ConnectionPool(object): # default timeout after which to try resurrecting a connection self.dead_timeout = dead_timeout + self.timeout_cutoff = timeout_cutoff self.selector = selector_class(dict(connections)) @@ -127,7 +131,8 @@ class ConnectionPool(object): else: dead_count = self.dead_count.get(connection, 0) + 1 self.dead_count[connection] = dead_count - self.dead.put((now + self.dead_timeout * 2 ** (dead_count - 1), connection)) + timeout = self.dead_timeout * 2 ** min(dead_count - 1, self.timeout_cutoff) + self.dead.put((now + timeout, connection)) def mark_live(self, connection): """ diff --git a/test_elasticsearch/test_connection_pool.py b/test_elasticsearch/test_connection_pool.py index 51b98667..0d03cf90 100644 --- a/test_elasticsearch/test_connection_pool.py +++ b/test_elasticsearch/test_connection_pool.py @@ -74,6 +74,15 @@ class TestConnectionPool(TestCase): self.assertEquals(3, pool.dead_count[42]) self.assertEquals((now + 4*60, 42), pool.dead.get()) + def test_timeout_for_failed_connections_is_limitted(self): + pool = ConnectionPool([(x, {}) for x in range(100)]) + now = time.time() + pool.dead_count[42] = 245 + pool.mark_dead(42, now=now) + + self.assertEquals(246, pool.dead_count[42]) + self.assertEquals((now + 32*60, 42), pool.dead.get()) + def test_dead_count_is_wiped_clean_for_connection_if_marked_live(self): pool = ConnectionPool([(x, {}) for x in range(100)]) now = time.time()