Adding new feature to pipeline.execute() so it can optionally add

multi-exec around consecutive redis commands that are for the same
slot.  Redis cluster does not support multi-exec for different slots,
even if the slots are on the same shard.

This feature provides limited support (*1) of atomic transaction using
multi-exec on redis cluster.

Step1: add optional arg use_multi=False to pipeline.execute(), so the
default behavior is what it used to be.  ie. this feature is off by
default.

*1: limited support: only for consecutive commands in pipeline for the
same slot.  eg. for commands
    incr a{1}   #1
    decr b{1}   #2
    incr a{2}   #3
    decr b{2}   #4
    incr a{3}   Grokzen#5
Since #1,#2 are for the same slot, we will add multi-exec around them
(multi before #1, and exec after #2).  #3,#4 are in their own one
slot, but different from that of #1,#2, so another set of multi-exec.
Grokzen#5 is in its own slot, no multi-exec added.
Thus the commands sent to the cluster are:
    multi
    incr a{1}   #1
    decr b{1}   #2
    exec
    multi
    incr a{2}   #3
    decr b{2}   #4
    exec
    incr a{3}   Grokzen#5
The result of the added multi-exec are stripped from the response, so
the client will see no change in response format.

Author: timliim

rediscluster/pipeline.py

@@ -90,7 +90,7 @@ def annotate_exception(self, exception, number, command):
             number, cmd, unicode(exception.args[0]))
         exception.args = (msg,) + exception.args[1:]
 
-    def execute(self, raise_on_error=True):
+    def execute(self, raise_on_error=True, use_multi = False):
         """
         """
         stack = self.command_stack
@@ -99,7 +99,7 @@ def execute(self, raise_on_error=True):
             return []
 
         try:
-            return self.send_cluster_commands(stack, raise_on_error)
+            return self.send_cluster_commands(stack, raise_on_error, use_multi = use_multi )
         finally:
             self.reset()
 
@@ -136,7 +136,7 @@ def reset(self):
         #     self.connection = None
 
     @clusterdown_wrapper
-    def send_cluster_commands(self, stack, raise_on_error=True, allow_redirections=True):
+    def send_cluster_commands(self, stack, raise_on_error=True, allow_redirections=True, use_multi = False ):
         """
         Send a bunch of cluster commands to the redis cluster.
 
@@ -165,7 +165,10 @@ def send_cluster_commands(self, stack, raise_on_error=True, allow_redirections=T
             # we can build a list of commands for each node.
             node_name = node['name']
             if node_name not in nodes:
-                nodes[node_name] = NodeCommands(self.parse_response, self.connection_pool.get_connection_by_node(node))
+                nodes[node_name] = NodeCommands(
+                                        self.parse_response,
+                                        self.connection_pool.get_connection_by_node(node),
+                                        use_multi           = use_multi )
 
             nodes[node_name].append(c)
 
@@ -377,12 +380,14 @@ class NodeCommands(object):
     """
     """
 
-    def __init__(self, parse_response, connection):
+    def __init__(self, parse_response, connection,
+                 use_multi = False ):
         """
         """
         self.parse_response = parse_response
         self.connection = connection
-        self.commands = []
+        self.commands   = []
+        self.useMulti   = use_multi     # if true, use multi-exec where possible.
 
     def append(self, c):
         """
@@ -404,12 +409,26 @@ def write(self):
         # build up all commands into a single request to increase network perf
         # send all the commands and catch connection and timeout errors.
         try:
-            connection.send_packed_command(connection.pack_commands([c.args for c in commands]))
+            if  self.useMulti :
+                cmds = self.addTransaction( commands )
+            else :
+                cmds = [ c.args for c in commands ]
+            connection.send_packed_command( connection.pack_commands( cmds ) )
         except (ConnectionError, TimeoutError) as e:
             for c in commands:
                 c.result = e
 
+    def addTransaction( self, commands ):
+        cmds    = []
+        return cmds
+
     def read(self):
+        if  self.useMulti :
+            self.readMultiExec()
+        else :
+            self.readPlain()
+
+    def readPlain(self):
         """
         """
         connection = self.connection
@@ -434,3 +453,8 @@ def read(self):
                     return
                 except RedisError:
                     c.result = sys.exc_info()[1]
+
+    def readMultiExec(self):
+        """
+        """
+        pass