Merge branch 'unstable'

Author: Grokzen

docs/pipelines.rst

@@ -5,11 +5,13 @@ Pipelines
 How pipelining works
 --------------------
 
+In redis-py-cluster, pipelining is all about trying to achieve greater network efficiency. Transaction support is disabled in redis-py-cluster. Use pipelines to avoid extra network round-trips, not to ensure atomicity.
+
 Just like in `redis-py`, `redis-py-cluster` queues up all the commands inside the client until execute is called. But, once execute is called, `redis-py-cluster` internals work slightly differently. It still packs the commands to efficiently transmit multiple commands across the network. But since different keys may be mapped to different nodes, redis-py-cluster must first map each key to the expected node. It then packs all the commands destined for each node in the cluster into its own packed sequence of commands. It uses the redis-py library to communicate with each node in the cluster.
 
-Ideally all the commands should be sent to each node in the cluster in parallel so that all the commands can be processed as fast as possible. The naive approach is to iterate through each node and send each batch of commands sequentially to each node. If redis-py supported some sort of non-blocking i/o we could send the network requests first and multiplex the socket responses from each node. Instead, we use threads to send the requests in parallel so that the total execution time only equals the amount of time for the slowest round trip to and from the given set of nodes in the cluster needed to process the commands.
+Ideally all the commands should be sent to each node in the cluster in parallel so that all the commands can be processed as fast as possible. We do this by first writing all of the commands to the sockets sequentially before reading any of the responses. This allows us to parallelize the network i/o without the overhead of managing python threads.
 
-In previous versions of the library there were some bugs associated with threaded operations and pipelining. We were freeing connections back into the connection pool prior to reading the responses from each thread and it caused all kinds of problems. Those issues were fixed but there was a special flag to allow you to turn off threading in case you were worried about it. Since we no longer have to use threads at all to get the performance we want, that flag was removed from the client.
+In previous versions of the library there were some bugs associated with pipelining operations. In an effort to simplify the logic and lessen the likelihood of bugs, if we get back connection errors, MOVED errors, ASK errors or any other error that can safely be retried, we fall back to sending these remaining commands sequentially to each individual node just as we would in a normal redis call. We still buffer the results inside the pipeline response so there will be no change in client behavior. During normal cluster operations, pipelined commands should work nearly efficiently as pipelined commands to a single instance redis. When there is a disruption to the cluster topography, like when keys are being resharded, or when a slave takes over for a master, there will be a slight loss of network efficiency. Commands that are rejected by the server are tried one at a time as we rebuild the slot mappings. Once the slots table is rebuilt correctly (usally in a second or so), the client resumes efficient networking behavior. We felt it was more important to prioritize correctness of behavior and reliable error handling over networking efficiency for the rare cases where the cluster topography is in flux.
 
 
 

docs/release-notes.rst

@@ -2,6 +2,14 @@ Release Notes
 =============
 
 
+1.3.1 (Oct 13, 2016)
+--------------------
+
+    * Rebuilt broken method scan_iter. Previous tests was to small to detect the problem but is not corrected to work on a bigger dataset during the test of that method. (korvus81, Grokzen, RedWhiteMiko)
+    * Errors in pipeline that should be retried, like connection errors, moved, errors and ask errors now fall back to single operation logic in StrictRedisCluster.execute_command. (72squared).
+    * Moved reinitialize_steps and counter into nodemanager so it can be correctly counted across pipeline operations (72squared).
+
+
 1.3.0 (Sep 11, 2016)
 --------------------
 

docs/upgrading.rst

@@ -3,6 +3,14 @@ Upgrading redis-py-cluster
 
 This document describes what must be done when upgrading between different versions to ensure that code still works.
 
+1.3.0 --> 1.3.1
+---------------
+
+Method `scan_iter` was rebuilt becuase it was broken and did not perform as expected. If you are using this method you should be carefull with this new implementation and test it through before using it. The expanded testing for that method indicates it should work without problems. If you find any issues with the new method please open a issue on github.
+
+A major refactoring was performed in the pipeline system that improved error handling and reliability of execution. It also simplified the code alot to make it easier to understand and continue to develop in the future. Becuase of this major refactoring you should really test throuhg your pipeline code to ensure that none of your code is broken because of this refactoring.
+
+
 
 1.2.0 --> Next release
 ----------------------

rediscluster/__init__.py

@@ -16,7 +16,7 @@
 setattr(redis, "StrictClusterPipeline", StrictClusterPipeline)
 
 # Major, Minor, Fix version
-__version__ = (1, 3, 0)
+__version__ = (1, 3, 1)
 
 if sys.version_info[0:3] == (3, 4, 0):
     raise RuntimeError("CRITICAL: rediscluster do not work with python 3.4.0. Please use 3.4.1 or higher.")

rediscluster/client.py

@@ -28,7 +28,7 @@
 from redis import StrictRedis
 from redis.client import list_or_args, parse_info
 from redis.connection import Token
-from redis._compat import iteritems, basestring, b, izip, nativestr
+from redis._compat import iteritems, basestring, b, izip, nativestr, long
 from redis.exceptions import RedisError, ResponseError, TimeoutError, DataError, ConnectionError, BusyLoadingError
 
 
@@ -154,6 +154,7 @@ def __init__(self, host=None, port=None, startup_nodes=None, max_connections=32,
                 startup_nodes=startup_nodes,
                 init_slot_cache=init_slot_cache,
                 max_connections=max_connections,
+                reinitialize_steps=reinitialize_steps,
                 max_connections_per_node=max_connections_per_node,
                 **kwargs
             )
@@ -164,8 +165,6 @@ def __init__(self, host=None, port=None, startup_nodes=None, max_connections=32,
         self.nodes_flags = self.__class__.NODES_FLAGS.copy()
         self.result_callbacks = self.__class__.RESULT_CALLBACKS.copy()
         self.response_callbacks = self.__class__.RESPONSE_CALLBACKS.copy()
-        self.reinitialize_counter = 0
-        self.reinitialize_steps = reinitialize_steps or 25
         self.response_callbacks = dict_merge(self.response_callbacks, self.CLUSTER_COMMANDS_RESPONSE_CALLBACKS)
 
     def __repr__(self):
@@ -196,10 +195,8 @@ def pipeline(self, transaction=None, shard_hint=None):
         return StrictClusterPipeline(
             connection_pool=self.connection_pool,
             startup_nodes=self.connection_pool.nodes.startup_nodes,
-            refresh_table_asap=self.refresh_table_asap,
             result_callbacks=self.result_callbacks,
             response_callbacks=self.response_callbacks,
-            reinitialize_steps=self.reinitialize_steps
         )
 
     def transaction(self, *args, **kwargs):
@@ -326,14 +323,13 @@ def execute_command(self, *args, **kwargs):
                 # This counter will increase faster when the same client object
                 # is shared between multiple threads. To reduce the frequency you
                 # can set the variable 'reinitialize_steps' in the constructor.
-                self.reinitialize_counter += 1
-                if self.reinitialize_counter % self.reinitialize_steps == 0:
-                    self.refresh_table_asap = True
+                self.refresh_table_asap = True
+                self.connection_pool.nodes.increment_reinitialize_counter()
 
                 node = self.connection_pool.nodes.set_node(e.host, e.port, server_type='master')
                 self.connection_pool.nodes.slots[e.slot_id][0] = node
             except TryAgainError as e:
-                if ttl < self.COMMAND_TTL / 2:
+                if ttl < self.RedisClusterRequestTTL / 2:
                     time.sleep(0.05)
             except AskError as e:
                 redirect_addr, asking = "{0}:{1}".format(e.host, e.port), True
@@ -551,6 +547,13 @@ def cluster_slots(self):
     ##########
     # All methods that must have custom implementation
 
+    def _parse_scan(self, response, **options):
+        """
+        Borrowed from redis-py::client.py
+        """
+        cursor, r = response
+        return long(cursor), r
+
     def scan_iter(self, match=None, count=None):
         """
         Make an iterator using the SCAN command so that the client doesn't
@@ -562,13 +565,36 @@ def scan_iter(self, match=None, count=None):
         Cluster impl:
             Result from SCAN is different in cluster mode.
         """
-        cursor = '0'
-        while cursor != 0:
-            for _, node_data in self.scan(cursor=cursor, match=match, count=count).items():
-                cursor, data = node_data
+        cursors = {}
+        nodeData = {}
+        for master_node in self.connection_pool.nodes.all_masters():
+            cursors[master_node["name"]] = "0"
+            nodeData[master_node["name"]] = master_node
+
+        while not all(cursors[node] == 0 for node in cursors):
+            for node in cursors:
+                if cursors[node] == 0:
+                    continue
+
+                conn = self.connection_pool.get_connection_by_node(nodeData[node])
+
+                pieces = ['SCAN', cursors[node]]
+                if match is not None:
+                    pieces.extend([Token('MATCH'), match])
+                if count is not None:
+                    pieces.extend([Token('COUNT'), count])
+
+                conn.send_command(*pieces)
+
+                raw_resp = conn.read_response()
+
+                # if you don't release the connection, the driver will make another, and you will hate your life
+                self.connection_pool.release(conn)
+                cur, resp = self._parse_scan(raw_resp)
+                cursors[node] = cur
 
-                for item in data:
-                    yield item
+                for r in resp:
+                    yield r
 
     def mget(self, keys, *args):
         """
@@ -1095,7 +1121,6 @@ def pipeline(self, transaction=True, shard_hint=None):
         return StrictClusterPipeline(
             connection_pool=self.connection_pool,
             startup_nodes=self.connection_pool.nodes.startup_nodes,
-            refresh_table_asap=self.refresh_table_asap,
             response_callbacks=self.response_callbacks
         )
 

rediscluster/connection.py

@@ -70,15 +70,15 @@ class ClusterConnectionPool(ConnectionPool):
     RedisClusterDefaultTimeout = None
 
     def __init__(self, startup_nodes=None, init_slot_cache=True, connection_class=ClusterConnection,
-                 max_connections=None, max_connections_per_node=False, **connection_kwargs):
+                 max_connections=None, max_connections_per_node=False, reinitialize_steps=None, **connection_kwargs):
         """
         """
         super(ClusterConnectionPool, self).__init__(connection_class=connection_class, max_connections=max_connections)
 
         self.max_connections = max_connections or 2 ** 31
         self.max_connections_per_node = max_connections_per_node
 
-        self.nodes = NodeManager(startup_nodes, **connection_kwargs)
+        self.nodes = NodeManager(startup_nodes, reinitialize_steps=reinitialize_steps, **connection_kwargs)
         if init_slot_cache:
             self.nodes.initialize()
 

rediscluster/nodemanager.py

@@ -19,14 +19,16 @@ class NodeManager(object):
     """
     RedisClusterHashSlots = 16384
 
-    def __init__(self, startup_nodes=None, **connection_kwargs):
+    def __init__(self, startup_nodes=None, reinitialize_steps=None, **connection_kwargs):
         """
         """
         self.connection_kwargs = connection_kwargs
         self.nodes = {}
         self.slots = {}
         self.startup_nodes = [] if startup_nodes is None else startup_nodes
         self.orig_startup_nodes = [node for node in self.startup_nodes]
+        self.reinitialize_counter = 0
+        self.reinitialize_steps = reinitialize_steps or 25
 
         if not self.startup_nodes:
             raise RedisClusterException("No startup nodes provided")
@@ -237,6 +239,13 @@ def initialize(self):
         # Set the tmp variables to the real variables
         self.slots = tmp_slots
         self.nodes = nodes_cache
+        self.reinitialize_counter = 0
+
+    def increment_reinitialize_counter(self, ct=1):
+        for i in range(1, ct):
+            self.reinitialize_counter += 1
+            if self.reinitialize_counter % self.reinitialize_steps == 0:
+                self.initialize()
 
     def cluster_require_full_coverage(self, nodes_cache):
         """