use pipeline to retry failed moved cmds

increase the redict numbers and make sure to try slow mode on the final failed commands

Author: kefu-reddit

rediscluster/client.py

@@ -973,17 +973,17 @@ def scan_iter(self, match=None, count=None, _type=None):
                 for r in resp:
                     yield r
 
-    # def mget(self, keys, *args):
-    #     """
-    #     Returns a list of values ordered identically to ``keys``
-    #
-    #     Cluster impl:
-    #         Itterate all keys and send GET for each key.
-    #         This will go alot slower than a normal mget call in Redis.
-    #
-    #         Operation is no longer atomic.
-    #     """
-    #     return [self.get(arg) for arg in list_or_args(keys, args)]
+    def mget(self, keys, *args):
+        """
+        Returns a list of values ordered identically to ``keys``
+
+        Cluster impl:
+            Itterate all keys and send GET for each key.
+            This will go alot slower than a normal mget call in Redis.
+
+            Operation is no longer atomic.
+        """
+        return [self.get(arg) for arg in list_or_args(keys, args)]
 
     def mset(self, *args, **kwargs):
         """

rediscluster/nodemanager.py

@@ -420,6 +420,9 @@ def get_node(self, host, port, server_type=None):
         return self.nodes[node_name]
 
     def move_slot_to_node(self, slot, node):
+        """
+        When moved response received, we should move all replicas with the master to the new slot.
+        """
         node_name = node['name']
         self.slots[slot] = [node]
         slave_nodes = self.slave_nodes_by_master.get(node_name)

rediscluster/pipeline.py

@@ -196,9 +196,12 @@ def _get_commands_by_node(self, cmds):
 
             master_node = self.connection_pool.get_node_by_slot(slot)
 
+            # for the same master_node, it should always get the same proxy_node to group
+            # as many commands as possible per node
             if master_node['name'] in proxy_node_by_master:
                 node = proxy_node_by_master[master_node['name']]
             else:
+                # TODO: should determine if using replicas by if command is read only
                 node = self.connection_pool.get_node_by_slot(slot, self.read_from_replicas)
                 proxy_node_by_master[master_node['name']] = node
 
@@ -216,7 +219,7 @@ def _get_commands_by_node(self, cmds):
 
             nodes[node_name].append(c)
 
-        return nodes
+        return nodes, connection_by_node
 
     def _execute_single_command(self, cmd):
         try:
@@ -234,43 +237,69 @@ def _send_cluster_commands(self, stack, raise_on_error=True, allow_redirections=
         """
         # the first time sending the commands we send all of the commands that were queued up.
         # if we have to run through it again, we only retry the commands that failed.
-        attempt = sorted(stack, key=lambda x: x.position)
-
-        # build a list of node objects based on node names we need to
-        nodes = self._get_commands_by_node(attempt)
-
-        # send the commands in sequence.
-        # we  write to all the open sockets for each node first, before reading anything
-        # this allows us to flush all the requests out across the network essentially in parallel
-        # so that we can read them all in parallel as they come back.
-        # we dont' multiplex on the sockets as they come available, but that shouldn't make too much difference.
-        node_commands = nodes.values()
-        events = []
-        for n in node_commands:
-            events.append(gevent.spawn(self._execute_node_commands, n))
-
-        gevent.joinall(events)
-
-        # release all of the redis connections we allocated earlier back into the connection pool.
-        # we used to do this step as part of a try/finally block, but it is really dangerous to
-        # release connections back into the pool if for some reason the socket has data still left in it
-        # from a previous operation. The write and read operations already have try/catch around them for
-        # all known types of errors including connection and socket level errors.
-        # So if we hit an exception, something really bad happened and putting any of
-        # these connections back into the pool is a very bad idea.
-        # the socket might have unread buffer still sitting in it, and then the
-        # next time we read from it we pass the buffered result back from a previous
-        # command and every single request after to that connection will always get
-        # a mismatched result. (not just theoretical, I saw this happen on production x.x).
-        for conn in connection_by_node.values():
-            self.connection_pool.release(conn)
+        cmds = sorted(stack, key=lambda x: x.position)
+
+        max_redirects = 5
+        cur_attempt = 0
+
+        while cur_attempt < max_redirects:
+
+            # build a list of node objects based on node names we need to
+            nodes, connection_by_node = self._get_commands_by_node(cmds)
+
+            # send the commands in sequence.
+            # we  write to all the open sockets for each node first, before reading anything
+            # this allows us to flush all the requests out across the network essentially in parallel
+            # so that we can read them all in parallel as they come back.
+            # we dont' multiplex on the sockets as they come available, but that shouldn't make too much difference.
+
+            # duke-cliff: I think it would still be faster if we use gevent to make the command in parallel
+            # the io is non-blocking, but serialization/deserialization will still be blocking previously
+            node_commands = nodes.values()
+            events = []
+            for n in node_commands:
+                events.append(gevent.spawn(self._execute_node_commands, n))
+
+            gevent.joinall(events)
+
+            # release all of the redis connections we allocated earlier back into the connection pool.
+            # we used to do this step as part of a try/finally block, but it is really dangerous to
+            # release connections back into the pool if for some reason the socket has data still left in it
+            # from a previous operation. The write and read operations already have try/catch around them for
+            # all known types of errors including connection and socket level errors.
+            # So if we hit an exception, something really bad happened and putting any of
+            # these connections back into the pool is a very bad idea.
+            # the socket might have unread buffer still sitting in it, and then the
+            # next time we read from it we pass the buffered result back from a previous
+            # command and every single request after to that connection will always get
+            # a mismatched result. (not just theoretical, I saw this happen on production x.x).
+            for conn in connection_by_node.values():
+                self.connection_pool.release(conn)
+
+            # will regroup moved commands and retry using pipeline(stacked commands)
+            # this would increase the pipeline performance a lot
+            moved_cmds = []
+            for c in cmds:
+                if isinstance(c.result, MovedError):
+                    e = c.result
+                    node = self.connection_pool.nodes.get_node(e.host, e.port, server_type='master')
+                    self.connection_pool.nodes.move_slot_to_node(e.slot_id, node)
+
+                    moved_cmds.append(c)
+
+            if moved_cmds:
+                cur_attempt += 1
+                cmds = sorted(moved_cmds, key=lambda x: x.position)
+                continue
+
+            break
 
         # if the response isn't an exception it is a valid response from the node
         # we're all done with that command, YAY!
         # if we have more commands to attempt, we've run into problems.
         # collect all the commands we are allowed to retry.
         # (MOVED, ASK, or connection errors or timeout errors)
-        attempt = sorted([c for c in attempt if isinstance(c.result, ERRORS_ALLOW_RETRY)], key=lambda x: x.position)
+        attempt = sorted([c for c in stack if isinstance(c.result, ERRORS_ALLOW_RETRY)], key=lambda x: x.position)
         if attempt and allow_redirections:
             # RETRY MAGIC HAPPENS HERE!
             # send these remaing comamnds one at a time using `execute_command`
@@ -288,14 +317,19 @@ def _send_cluster_commands(self, stack, raise_on_error=True, allow_redirections=
             # flag to rebuild the slots table from scratch. So MOVED errors should
             # correct themselves fairly quickly.
 
-            log.debug("pipeline has failed commands: {}".format([c.result for c in attempt]))
+            # with the previous redirect retries, I could barely see the slow mode happening again
+            log.info("pipeline in slow mode to execute failed commands: {}".format([c.result for c in attempt]))
 
             self.connection_pool.nodes.increment_reinitialize_counter(len(attempt))
+
+            # even in the slow mode, we could use gevent to make things faster
             events = []
             for c in attempt:
                 events.append(gevent.spawn(self._execute_single_command, c))
+
             gevent.joinall(events)
 
+
         # turn the response back into a simple flat array that corresponds
         # to the sequence of commands issued in the stack in pipeline.execute()
         response = [c.result for c in sorted(stack, key=lambda x: x.position)]

requirements.txt

@@ -1,3 +1,3 @@
 redis>=3.0.0,<4.0.0
-gevent==1.5.0
-greenlet==0.4.16
+gevent
+greenlet

setup.py

@@ -19,8 +19,8 @@
     history = f.read()
 
 setup(
-    name="redis-py-cluster-patched",
-    version="2.1.0.999.10",
+    name="redis-py-cluster",
+    version="2.1.1",
     description="Library for communicating with Redis Clusters. Built on top of redis-py lib",
     long_description=readme + '\n\n' + history,
     long_description_content_type="text/markdown",
@@ -30,12 +30,11 @@
     maintainer_email='Grokzen@gmail.com',
     packages=["rediscluster"],
     url='http://github.com/grokzen/redis-py-cluster',
-    download_url="https://github.com/duke-cliff/redis-py-cluster/archive/2.1.0.999.10.tar.gz",
     license='MIT',
     install_requires=[
         'redis>=3.0.0,<4.0.0',
-        'gevent==1.5.0',
-        'greenlet==0.4.16',
+        'gevent',
+        'greenlet',
     ],
     python_requires=">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4",
     extras_require={