on the write side:

- scan thru all the commands for one node.  Add multi-exec around
  consecutive commands for the same slot.
- add NodeCommands.iRsp[] to record the indices of the expected
  responses.  eg. for [1], iRsp will be
    [ None, "0", "1", [ 0, 1 ], None, "2", "3", [ 2, 3 ], 4 ]
      multi #1   #2    exec     multi #3   #4    exec     Grokzen#5
  - the rsp to MULTI is "OK", so we want to discard that rsp.
    If the index is None, we will discard the rsp.
  - the rsp for queued commands is "QUEUED" (good case) or error.
    If good case, we want to discard the placeholder rsp "QUEUED".
    An index with type str indicates we should discard good case rsp.
    In case of error, we use that index (str->int) to record error rsp.
  - the rsp for exec is a list of rsp for queued commands (good
    case) or an error.
    If good case, we have the indices of the commands corresponding to
    the rsp.
    If error, we recorded the errors already, so we can discard this
    error (error of exec says "see previous errors" anyway).
  - A plain int index is for commands outside multi-exec, eg. Grokzen#5.

[1]
        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

Author: timliim

rediscluster/pipeline.py

@@ -389,6 +389,7 @@ def __init__(self, parse_response, connection,
         self.parse_response = parse_response
         self.connection = connection
         self.commands   = []
+        self.iRsp       = []
         self.useMulti   = use_multi     # if true, use multi-exec where possible.
 
     def append(self, c):
@@ -422,8 +423,51 @@ def write(self):
 
     def addTransaction( self, commands ):
         cmds    = []
+        iRsp    = []                    # list of indices where the response should go
+
+        nCmds   = len( commands )
+        if nCmds        < 2:
+            cmds        = [ c.args for    c in            commands  ]
+            iRsp        = [ i      for i, c in enumerate( commands )]
+            self.iRsp   = iRsp
+            return cmds
+
+        iRun    = []                    # list of indices in the run
+        inArun  = False                 # in a run of commands for same slot
+        for i  in range( nCmds - 1 ) :
+            c1  = commands[ i      ]
+            c2  = commands[ i  + 1 ]
+          # starting a new run?
+            if  not inArun and c1.slot == c2.slot :
+                inArun  = True
+                cmds.append( ( 'MULTI', ) )
+                iRsp.append( None )     # should discard this rsp (will be "OK")
+          # add the actual cmd
+            inArun  = self.addCmd( cmds, c1, i,     iRsp, iRun, inArun, c1.slot == c2.slot )
+
+      # remember to add the last commands
+        inArun      = self.addCmd( cmds, c2, i + 1, iRsp, iRun, inArun, False )
+
+        self.iRsp   = iRsp
         return cmds
 
+    def addCmd( self, cmds, cmd, i, iRsp, iRun, inArun, sameSlot ):
+        cmds.append( cmd.args )
+        if  inArun:
+            index = str( i     )        # in good case we should discard this rsp.
+            iRsp.append( index )        # should discard this rsp (will be "QUEUED")
+            iRun.append( i     )
+        else:
+            iRsp.append( i    )         # this response is for cmds[ i ]
+
+      # ending a run?
+        if  inArun  and not sameSlot :
+            inArun  = False
+            cmds.append( ( 'EXEC', ) )
+            iRsp.append( iRun[ : ] )    # need [:] to create a shallow copy, because we clear iRun next.
+            iRun.clear()                # must use clear(), since the caller owns iRun.
+        return inArun
+
     def read(self):
         if  self.useMulti :
             self.readMultiExec()