diff --git a/elasticsearch/connection_pool.py b/elasticsearch/connection_pool.py index c7541e15..477bde39 100644 --- a/elasticsearch/connection_pool.py +++ b/elasticsearch/connection_pool.py @@ -7,10 +7,37 @@ except ImportError: from queue import PriorityQueue, Empty class ConnectionSelector(object): - " Base class for Selectors. " + """ + Simple class used to select a connection from a list of currently live + connection instances. In init time it is passed a dictionary containing all + the connections' options which it can then use during the selection + process. When the `select` method is called it is given a list of + *currently* live connections to choose from. + + The options dictionary is the one that has been passed to + :class:`~elasticsearch.Transport` as `hosts` param and the same that is + used to construct the Connection object itself. When the Connection was + created from information retrieved from the cluster via the sniffing + process it will be the dictionary returned by the `host_info_callback`. + + Example of where this would be useful is a zone-aware selector that would + only select connections from it's own zones and only fall back to other + connections where there would be none in it's zones. + """ def __init__(self, opts): + """ + :arg opts: dictionary of connection instances and theit options + """ self.connection_opts = opts + def select(self, connections): + """ + Select a connection from the given list. + + :arg connections: list of live connections to choose from + """ + pass + class RandomSelector(ConnectionSelector): """ @@ -21,7 +48,9 @@ class RandomSelector(ConnectionSelector): class RoundRobinSelector(ConnectionSelector): - " Default selector using round-robin. " + """ + Selector using round-robin. + """ def __init__(self, opts): super(RoundRobinSelector, self).__init__(opts) self.rr = -1 @@ -33,8 +62,40 @@ class RoundRobinSelector(ConnectionSelector): class ConnectionPool(object): + """ + Container holding the :class:`~elasticsearch.Connection` instances, + managing the selection process (via a + :class:`~elasticsearch.ConnectionSelector`) and dead connections. + + It's only interactions are with the :class:`~elasticsearch.Transport` class + that drives all the actions within `ConnectionPool`. + + Initially connections are stored on the class as a list and, along with the + connection options, get passed to the `ConnectionSelector` instance for + future reference. + + Upon each request the `Transport` will ask for a `Connection` via the + `get_connection` method. If the connection fails (it's `perform_request` + raises a `ConnectionError`) it will be marked as dead (via `mark_dead`) and + put on a timeout (if it fails N times in a row the timeout is exponentially + longer - the formula is `default_timeout * 2 ** (fail_count - 1)`). When + the timeout is over the connection will be resurrected and returned to the + 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): + """ + :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 + :arg selector_class: :class:`~elasticsearch.ConnectionSelector` + subclass to use + :arg randomize_hosts: shuffle the list of connections upon arrival to + avoid dog piling effect across processes + """ self.connections = [c for (c, opts) in connections] + # PriorityQueue for thread safety and ease of timeout management self.dead = PriorityQueue(len(self.connections)) self.dead_count = {} @@ -49,6 +110,13 @@ class ConnectionPool(object): self.selector = selector_class(dict(connections)) def mark_dead(self, connection, dead_count, now=None): + """ + Mark the connection as dead (failed). Remove it from the live pool and + put it on a timeout. + + :arg connection: the failed instance + :arg dead_count: number of consecutive failures + """ # allow inject for testing purposes now = now if now else time.time() try: @@ -61,6 +129,12 @@ class ConnectionPool(object): self.dead.put((now + self.dead_timeout * 2 ** (dead_count - 1), connection)) def mark_live(self, connection): + """ + Mark connection as healthy after a resurrection. Resets the fail + counter for the connection. + + :arg connection: the connection to redeem + """ try: del self.dead_count[connection] except KeyError: @@ -68,6 +142,15 @@ class ConnectionPool(object): pass def resurrect(self, force=False): + """ + Attempt to resurrect a connection from the dead pool. It will try to + locate one (not all) eligible (it's timeout is over) connection to + return to th live pool. + + :arg force: resurrect a connection even if there is none eligible (used + when we have no live connections) + + """ # no dead connections if self.dead.empty(): return @@ -88,6 +171,16 @@ class ConnectionPool(object): self.connections.append(connection) def get_connection(self): + """ + Return a connection from the pool using the `ConnectionSelector` + instance. + + It tries to resurrect eligible connections, forces a resurrection when + no connections are availible and passes the list of live connections to + the selector instance to choose from. + + Returns a connection instance and it's current fail count. + """ self.resurrect() # no live nodes, resurrect one by force