use gevent for pipeline execution

Author: kefu-reddit

rediscluster/pipeline.py

@@ -2,6 +2,7 @@
 
 # python std lib
 import sys
+import logging
 
 # rediscluster imports
 from .client import RedisCluster
@@ -15,6 +16,10 @@
 from redis.exceptions import ConnectionError, RedisError, TimeoutError
 from redis._compat import imap, unicode
 
+from gevent import monkey; monkey.patch_all()
+import gevent
+
+log = logging.getLogger(__name__)
 
 ERRORS_ALLOW_RETRY = (ConnectionError, TimeoutError, MovedError, AskError, TryAgainError)
 
@@ -174,6 +179,11 @@ def send_cluster_commands(self, stack, raise_on_error=True, allow_redirections=T
         # If it fails the configured number of times then raise exception back to caller of this method
         raise ClusterDownError("CLUSTERDOWN error. Unable to rebuild the cluster")
 
+    def _execute_node_commands(self, n):
+        n.write()
+
+        n.read()
+
     def _send_cluster_commands(self, stack, raise_on_error=True, allow_redirections=True):
         """
         Send a bunch of cluster commands to the redis cluster.
@@ -227,11 +237,11 @@ def _send_cluster_commands(self, stack, raise_on_error=True, allow_redirections=
         # 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:
-            n.write()
+            events.append(gevent.spawn(self._execute_node_commands, n))
 
-        for n in node_commands:
-            n.read()
+        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
@@ -269,6 +279,8 @@ def _send_cluster_commands(self, stack, raise_on_error=True, allow_redirections=
             # If a lot of commands have failed, we'll be setting the
             # flag to rebuild the slots table from scratch. So MOVED errors should
             # correct themselves fairly quickly.
+
+            log.debug("pipeline has failed commands: {}".format(attempt))
             self.connection_pool.nodes.increment_reinitialize_counter(len(attempt))
             for c in attempt:
                 try:

requirements.txt

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

setup.py

@@ -20,7 +20,7 @@
 
 setup(
     name="redis-py-cluster-patched",
-    version="2.1.0.999.8",
+    version="2.1.0.999.9",
     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,10 +30,12 @@
     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.8.tar.gz",
+    download_url="https://github.com/duke-cliff/redis-py-cluster/archive/2.1.0.999.9.tar.gz",
     license='MIT',
     install_requires=[
-        'redis>=3.0.0,<4.0.0'
+        'redis>=3.0.0,<4.0.0',
+        'gevent==1.5.0',
+        'greenlet==0.4.16',
     ],
     python_requires=">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4",
     extras_require={