Add LMPOP/BLMPOP commands. (#9373)

We want to add COUNT option for BLPOP.
But we can't do it without breaking compatibility due to the command arguments syntax.
So this commit introduce two new commands.

Syntax for the new LMPOP command:
`LMPOP numkeys [<key> ...] LEFT|RIGHT [COUNT count]`

Syntax for the new BLMPOP command:
`BLMPOP timeout numkeys [<key> ...] LEFT|RIGHT [COUNT count]`

Some background:
- LPOP takes one key, and can return multiple elements.
- BLPOP takes multiple keys, but returns one element from just one key.
- LMPOP can take multiple keys and return multiple elements from just one key.

Note that LMPOP/BLMPOP  can take multiple keys, it eventually operates on just one key.
And it will propagate as LPOP or RPOP with the COUNT option.

As a new command, it still return NIL if we can't pop any elements.
For the normal response is nested arrays in RESP2 and RESP3, like:
```
LMPOP/BLMPOP 
1) keyname
2) 1) element1
   2) element2
```
I.e. unlike BLPOP that returns a key name and one element so it uses a flat array,
and LPOP that returns multiple elements with no key name, and again uses a flat array,
this one has to return a nested array, and it does for for both RESP2 and RESP3 (like SCAN does)

Some discuss can see: #766 #8824

1 files changed, 337 insertions, 59 deletions

diff --git a/tests/unit/type/list.tcl b/tests/unit/type/list.tcl
index 20fd0e49b..735c5eb18 100644
--- a/tests/unit/type/list.tcl
+++ b/tests/unit/type/list.tcl
@@ -6,6 +6,28 @@ start_server {
 } {
     source "tests/unit/type/list-common.tcl"
 
+    # A helper function for BPOP/BLMPOP with one input key.
+    proc bpop_command {rd pop key timeout} {
+        if {$pop == "BLMPOP_LEFT"} {
+            $rd blmpop $timeout 1 $key left count 1
+        } elseif {$pop == "BLMPOP_RIGHT"} {
+            $rd blmpop $timeout 1 $key right count 1
+        } else {
+            $rd $pop $key $timeout
+        }
+    }
+
+    # A helper function for BPOP/BLMPOP with two input keys.
+    proc bpop_command_two_key {rd pop key key2 timeout} {
+        if {$pop == "BLMPOP_LEFT"} {
+            $rd blmpop $timeout 2 $key $key2 left count 1
+        } elseif {$pop == "BLMPOP_RIGHT"} {
+            $rd blmpop $timeout 2 $key $key2 right count 1
+        } else {
+            $rd $pop $key $key2 $timeout
+        }
+    }
+
     test {LPOS basic usage} {
         r DEL mylist
         r RPUSH mylist a b c 1 2 3 c c
@@ -111,10 +133,6 @@ start_server {
         assert_equal $largevalue(linkedlist) [r rpop mylist2]
         assert_equal c [r lpop mylist2]
     }
-
-    test {R/LPOP against empty list} {
-        r lpop non-existing-list
-    } {}
     
     test {R/LPOP with the optional count argument} {
         assert_equal 7 [r lpush listcount aa bb cc dd ee ff gg]
@@ -147,53 +165,77 @@ start_server {
     }
 
     foreach {type large} [array get largevalue] {
-        test "BLPOP, BRPOP: single existing list - $type" {
+    foreach {pop} {BLPOP BLMPOP_LEFT} {
+        test "$pop: single existing list - $type" {
             set rd [redis_deferring_client]
             create_list blist "a b $large c d"
 
-            $rd blpop blist 1
+            bpop_command $rd $pop blist 1
             assert_equal {blist a} [$rd read]
-            $rd brpop blist 1
+            if {$pop == "BLPOP"} {
+                bpop_command $rd BRPOP blist 1
+            } else {
+                bpop_command $rd BLMPOP_RIGHT blist 1
+            }
             assert_equal {blist d} [$rd read]
 
-            $rd blpop blist 1
+            bpop_command $rd $pop blist 1
             assert_equal {blist b} [$rd read]
-            $rd brpop blist 1
+            if {$pop == "BLPOP"} {
+                bpop_command $rd BRPOP blist 1
+            } else {
+                bpop_command $rd BLMPOP_RIGHT blist 1
+            }
             assert_equal {blist c} [$rd read]
+
+            assert_equal 1 [r llen blist]
         }
 
-        test "BLPOP, BRPOP: multiple existing lists - $type" {
+        test "$pop: multiple existing lists - $type" {
             set rd [redis_deferring_client]
             create_list blist1{t} "a $large c"
             create_list blist2{t} "d $large f"
 
-            $rd blpop blist1{t} blist2{t} 1
+            bpop_command_two_key $rd $pop blist1{t} blist2{t} 1
             assert_equal {blist1{t} a} [$rd read]
-            $rd brpop blist1{t} blist2{t} 1
+            if {$pop == "BLPOP"} {
+                bpop_command_two_key $rd BRPOP blist1{t} blist2{t} 1
+            } else {
+                bpop_command_two_key $rd BLMPOP_RIGHT blist1{t} blist2{t} 1
+            }
             assert_equal {blist1{t} c} [$rd read]
             assert_equal 1 [r llen blist1{t}]
             assert_equal 3 [r llen blist2{t}]
 
-            $rd blpop blist2{t} blist1{t} 1
+            bpop_command_two_key $rd $pop blist2{t} blist1{t} 1
             assert_equal {blist2{t} d} [$rd read]
-            $rd brpop blist2{t} blist1{t} 1
+            if {$pop == "BLPOP"} {
+                bpop_command_two_key $rd BRPOP blist2{t} blist1{t} 1
+            } else {
+                bpop_command_two_key $rd BLMPOP_RIGHT blist2{t} blist1{t} 1
+            }
             assert_equal {blist2{t} f} [$rd read]
             assert_equal 1 [r llen blist1{t}]
             assert_equal 1 [r llen blist2{t}]
         }
 
-        test "BLPOP, BRPOP: second list has an entry - $type" {
+        test "$pop: second list has an entry - $type" {
             set rd [redis_deferring_client]
             r del blist1{t}
             create_list blist2{t} "d $large f"
 
-            $rd blpop blist1{t} blist2{t} 1
+            bpop_command_two_key $rd $pop blist1{t} blist2{t} 1
             assert_equal {blist2{t} d} [$rd read]
-            $rd brpop blist1{t} blist2{t} 1
+            if {$pop == "BLPOP"} {
+                bpop_command_two_key $rd BRPOP blist1{t} blist2{t} 1
+            } else {
+                bpop_command_two_key $rd BLMPOP_RIGHT blist1{t} blist2{t} 1
+            }
             assert_equal {blist2{t} f} [$rd read]
             assert_equal 0 [r llen blist1{t}]
             assert_equal 1 [r llen blist2{t}]
         }
+    }
 
         test "BRPOPLPUSH - $type" {
             r del target{t}
@@ -239,26 +281,31 @@ start_server {
         }
     }
 
-    test "BLPOP, LPUSH + DEL should not awake blocked client" {
+foreach {pop} {BLPOP BLMPOP_LEFT} {
+    test "$pop, LPUSH + DEL should not awake blocked client" {
         set rd [redis_deferring_client]
         r del list
 
-        $rd blpop list 0
+        bpop_command $rd $pop list 0
+        after 100 ;# Make sure rd is blocked before MULTI
+        wait_for_blocked_client
+
         r multi
         r lpush list a
         r del list
         r exec
         r del list
         r lpush list b
-        $rd read
-    } {list b}
+        assert_equal {list b} [$rd read]
+    }
 
-    test "BLPOP, LPUSH + DEL + SET should not awake blocked client" {
+    test "$pop, LPUSH + DEL + SET should not awake blocked client" {
         set rd [redis_deferring_client]
         r del list
 
-        $rd blpop list 0
+        bpop_command $rd $pop list 0
         after 100 ;# Make sure rd is blocked before MULTI
+        wait_for_blocked_client
 
         r multi
         r lpush list a
@@ -267,8 +314,9 @@ start_server {
         r exec
         r del list
         r lpush list b
-        $rd read
-    } {list b}
+        assert_equal {list b} [$rd read]
+    }
+}
 
     test "BLPOP with same key multiple times should work (issue #801)" {
         set rd [redis_deferring_client]
@@ -291,29 +339,36 @@ start_server {
         assert_equal [$rd read] {list2{t} b}
     }
 
-    test "MULTI/EXEC is isolated from the point of view of BLPOP" {
+foreach {pop} {BLPOP BLMPOP_LEFT} {
+    test "MULTI/EXEC is isolated from the point of view of $pop" {
         set rd [redis_deferring_client]
         r del list
-        $rd blpop list 0
+
+        bpop_command $rd $pop list 0
+        after 100 ;# Make sure rd is blocked before MULTI
+        wait_for_blocked_client
+
         r multi
         r lpush list a
         r lpush list b
         r lpush list c
         r exec
-        $rd read
-    } {list c}
+        assert_equal {list c} [$rd read]
+    }
 
-    test "BLPOP with variadic LPUSH" {
+    test "$pop with variadic LPUSH" {
         set rd [redis_deferring_client]
         r del blist
         if {$::valgrind} {after 100}
-        $rd blpop blist 0
+        bpop_command $rd $pop blist 0
         if {$::valgrind} {after 100}
+        wait_for_blocked_client
         assert_equal 2 [r lpush blist foo bar]
         if {$::valgrind} {after 100}
         assert_equal {blist bar} [$rd read]
         assert_equal foo [lindex [r lrange blist 0 -1] 0]
     }
+}
 
     test "BRPOPLPUSH with zero timeout should block indefinitely" {
         set rd [redis_deferring_client]
@@ -412,6 +467,32 @@ start_server {
         assert_equal {foo} [r lrange target2{t} 0 -1]
     }
 
+    test "BLMPOP with multiple blocked clients" {
+        set rd1 [redis_deferring_client]
+        set rd2 [redis_deferring_client]
+        set rd3 [redis_deferring_client]
+        set rd4 [redis_deferring_client]
+        r del blist{t} blist2{t}
+
+        $rd1 blmpop 0 2 blist{t} blist2{t} left count 1
+        $rd2 blmpop 0 2 blist{t} blist2{t} right count 10
+        $rd3 blmpop 0 2 blist{t} blist2{t} left count 10
+        $rd4 blmpop 0 2 blist{t} blist2{t} right count 1
+        wait_for_blocked_clients_count 4
+
+        r multi
+        r lpush blist{t} a b c d e
+        r lpush blist2{t} 1 2 3 4 5
+        r exec
+
+        assert_equal {blist{t} e} [$rd1 read]
+        assert_equal {blist{t} {a b c d}} [$rd2 read]
+        assert_equal {blist2{t} {5 4 3 2 1}} [$rd3 read]
+
+        r lpush blist2{t} 1 2 3
+        assert_equal {blist2{t} 1} [$rd4 read]
+    }
+
     test "Linked LMOVEs" {
       set rd1 [redis_deferring_client]
       set rd2 [redis_deferring_client]
@@ -420,6 +501,7 @@ start_server {
 
       $rd1 blmove list1{t} list2{t} right left 0
       $rd2 blmove list2{t} list3{t} left right 0
+      wait_for_blocked_clients_count 2
 
       r rpush list1{t} foo
 
@@ -436,6 +518,7 @@ start_server {
 
       $rd1 brpoplpush list1{t} list2{t} 0
       $rd2 brpoplpush list2{t} list1{t} 0
+      wait_for_blocked_clients_count 2
 
       r rpush list1{t} foo
 
@@ -449,6 +532,7 @@ start_server {
       r del blist{t}
 
       $rd brpoplpush blist{t} blist{t} 0
+      wait_for_blocked_client
 
       r rpush blist{t} foo
 
@@ -475,6 +559,7 @@ start_server {
         r del srclist{t} dstlist{t} somekey{t}
         r set somekey{t} somevalue
         $blocked_client brpoplpush srclist{t} dstlist{t} 0
+        wait_for_blocked_client
         $watching_client watch dstlist{t}
         $watching_client read
         $watching_client multi
@@ -492,6 +577,7 @@ start_server {
         r del srclist{t} dstlist{t} somekey{t}
         r set somekey{t} somevalue
         $blocked_client brpoplpush srclist{t} dstlist{t} 0
+        wait_for_blocked_client
         $watching_client watch dstlist{t}
         $watching_client read
         $watching_client multi
@@ -513,16 +599,19 @@ start_server {
       $rd read
     } {}
 
-    test "BLPOP when new key is moved into place" {
+foreach {pop} {BLPOP BLMPOP_LEFT} {
+    test "$pop when new key is moved into place" {
         set rd [redis_deferring_client]
+        r del foo{t}
 
-        $rd blpop foo{t} 5
+        bpop_command $rd $pop foo{t} 0
+        wait_for_blocked_client
         r lpush bob{t} abc def hij
         r rename bob{t} foo{t}
         $rd read
     } {foo{t} hij}
 
-    test "BLPOP when result key is created by SORT..STORE" {
+    test "$pop when result key is created by SORT..STORE" {
         set rd [redis_deferring_client]
 
         # zero out list from previous test without explicit delete
@@ -530,17 +619,20 @@ start_server {
         r lpop foo{t}
         r lpop foo{t}
 
-        $rd blpop foo{t} 5
+        bpop_command $rd $pop foo{t} 5
+        wait_for_blocked_client
         r lpush notfoo{t} hello hola aguacate konichiwa zanzibar
         r sort notfoo{t} ALPHA store foo{t}
         $rd read
     } {foo{t} aguacate}
+}
 
-    foreach {pop} {BLPOP BRPOP} {
+    foreach {pop} {BLPOP BRPOP BLMPOP_LEFT BLMPOP_RIGHT} {
         test "$pop: with single empty list argument" {
             set rd [redis_deferring_client]
             r del blist1
-            $rd $pop blist1 1
+            bpop_command $rd $pop blist1 1
+            wait_for_blocked_client
             r rpush blist1 foo
             assert_equal {blist1 foo} [$rd read]
             assert_equal 0 [r exists blist1]
@@ -548,14 +640,14 @@ start_server {
 
         test "$pop: with negative timeout" {
             set rd [redis_deferring_client]
-            $rd $pop blist1 -1
+            bpop_command $rd $pop blist1 -1
             assert_error "ERR*is negative*" {$rd read}
         }
 
         test "$pop: with non-integer timeout" {
             set rd [redis_deferring_client]
             r del blist1
-            $rd $pop blist1 0.1
+            bpop_command $rd $pop blist1 0.1
             r rpush blist1 foo
             assert_equal {blist1 foo} [$rd read]
             assert_equal 0 [r exists blist1]
@@ -565,7 +657,8 @@ start_server {
             # To test this, use a timeout of 0 and wait a second.
             # The blocking pop should still be waiting for a push.
             set rd [redis_deferring_client]
-            $rd $pop blist1 0
+            bpop_command $rd $pop blist1 0
+            wait_for_blocked_client
             after 1000
             r rpush blist1 foo
             assert_equal {blist1 foo} [$rd read]
@@ -575,6 +668,7 @@ start_server {
             set rd [redis_deferring_client]
             r del blist1{t} blist2{t}
             r set blist2{t} nolist{t}
+            bpop_command_two_key $rd $pop blist1{t} blist2{t} 1
             $rd $pop blist1{t} blist2{t} 1
             assert_error "WRONGTYPE*" {$rd read}
         }
@@ -582,7 +676,8 @@ start_server {
         test "$pop: timeout" {
             set rd [redis_deferring_client]
             r del blist1{t} blist2{t}
-            $rd $pop blist1{t} blist2{t} 1
+            bpop_command_two_key $rd $pop blist1{t} blist2{t} 1
+            wait_for_blocked_client
             assert_equal {} [$rd read]
         }
 
@@ -590,13 +685,15 @@ start_server {
             set rd [redis_deferring_client]
             r del blist1{t} blist2{t}
 
-            $rd $pop blist1{t} blist2{t} 1
+            bpop_command_two_key $rd $pop blist1{t} blist2{t} 1
+            wait_for_blocked_client
             r rpush blist1{t} foo
             assert_equal {blist1{t} foo} [$rd read]
             assert_equal 0 [r exists blist1{t}]
             assert_equal 0 [r exists blist2{t}]
 
-            $rd $pop blist1{t} blist2{t} 1
+            bpop_command_two_key $rd $pop blist1{t} blist2{t} 1
+            wait_for_blocked_client
             r rpush blist2{t} foo
             assert_equal {blist2{t} foo} [$rd read]
             assert_equal 0 [r exists blist1{t}]
@@ -604,16 +701,57 @@ start_server {
         }
     }
 
-    test {BLPOP inside a transaction} {
+foreach {pop} {BLPOP BLMPOP_LEFT} {
+    test "$pop inside a transaction" {
         r del xlist
         r lpush xlist foo
         r lpush xlist bar
         r multi
-        r blpop xlist 0
-        r blpop xlist 0
-        r blpop xlist 0
+
+        bpop_command r $pop xlist 0
+        bpop_command r $pop xlist 0
+        bpop_command r $pop xlist 0
         r exec
     } {{xlist bar} {xlist foo} {}}
+}
+
+    test {BLMPOP propagate as pop with count command to replica} {
+        set rd [redis_deferring_client]
+        set repl [attach_to_replication_stream]
+
+        # BLMPOP without block.
+        r lpush mylist{t} a b c
+        r rpush mylist2{t} 1 2 3
+        r blmpop 0 1 mylist{t} left count 1
+        r blmpop 0 2 mylist{t} mylist2{t} right count 10
+        r blmpop 0 2 mylist{t} mylist2{t} right count 10
+
+        # BLMPOP with block.
+        $rd blmpop 0 1 mylist{t} left count 1
+        wait_for_blocked_client
+        r lpush mylist{t} a
+        $rd blmpop 0 2 mylist{t} mylist2{t} left count 5
+        wait_for_blocked_client
+        r lpush mylist{t} a b c
+        $rd blmpop 0 2 mylist{t} mylist2{t} right count 10
+        wait_for_blocked_client
+        r rpush mylist2{t} a b c
+
+        assert_replication_stream $repl {
+            {select *}
+            {lpush mylist{t} a b c}
+            {rpush mylist2{t} 1 2 3}
+            {lpop mylist{t} 1}
+            {rpop mylist{t} 2}
+            {rpop mylist2{t} 3}
+            {lpush mylist{t} a}
+            {lpop mylist{t} 1}
+            {lpush mylist{t} a b c}
+            {lpop mylist{t} 3}
+            {rpush mylist2{t} a b c}
+            {rpop mylist2{t} 3}
+        }
+    } {} {needs:repl}
 
     test {LPUSHX, RPUSHX - generic} {
         r del xlist
@@ -860,23 +998,46 @@ start_server {
     } {}
 
     foreach {type large} [array get largevalue] {
-        test "Basic LPOP/RPOP - $type" {
+        test "Basic LPOP/RPOP/LMPOP - $type" {
             create_list mylist "$large 1 2"
             assert_equal $large [r lpop mylist]
             assert_equal 2 [r rpop mylist]
             assert_equal 1 [r lpop mylist]
             assert_equal 0 [r llen mylist]
 
-            # pop on empty list
-            assert_equal {} [r lpop mylist]
-            assert_equal {} [r rpop mylist]
+            create_list mylist "$large 1 2"
+            assert_equal "mylist $large" [r lmpop 1 mylist left count 1]
+            assert_equal {mylist {2 1}} [r lmpop 2 mylist mylist right count 2]
         }
     }
 
-    test {LPOP/RPOP against non list value} {
-        r set notalist foo
-        assert_error WRONGTYPE* {r lpop notalist}
-        assert_error WRONGTYPE* {r rpop notalist}
+    test {LPOP/RPOP/LMPOP against empty list} {
+        r del non-existing-list{t} non-existing-list2{t}
+
+        assert_equal {} [r lpop non-existing-list{t}]
+        assert_equal {} [r rpop non-existing-list2{t}]
+
+        assert_equal {} [r lmpop 1 non-existing-list{t} left count 1]
+        assert_equal {} [r lmpop 1 non-existing-list{t} left count 10]
+        assert_equal {} [r lmpop 2 non-existing-list{t} non-existing-list2{t} right count 1]
+        assert_equal {} [r lmpop 2 non-existing-list{t} non-existing-list2{t} right count 10]
+    }
+
+    test {LPOP/RPOP/LMPOP NON-BLOCK or BLOCK against non list value} {
+        r set notalist{t} foo
+        assert_error WRONGTYPE* {r lpop notalist{t}}
+        assert_error WRONGTYPE* {r blpop notalist{t} 0}
+        assert_error WRONGTYPE* {r rpop notalist{t}}
+        assert_error WRONGTYPE* {r brpop notalist{t} 0}
+
+        r del notalist2{t}
+        assert_error "WRONGTYPE*" {r lmpop 2 notalist{t} notalist2{t} left count 1}
+        assert_error "WRONGTYPE*" {r blmpop 0 2 notalist{t} notalist2{t} left count 1}
+
+        r del notalist{t}
+        r set notalist2{t} nolist
+        assert_error "WRONGTYPE*" {r lmpop 2 notalist{t} notalist2{t} right count 10}
+        assert_error "WRONGTYPE*" {r blmpop 0 2 notalist{t} notalist2{t} left count 1}
     }
 
     foreach {type num} {quicklist 250 quicklist 500} {
@@ -897,6 +1058,121 @@ start_server {
         }
     }
 
+    test {LMPOP with illegal argument} {
+        assert_error "ERR wrong number of arguments*" {r lmpop}
+        assert_error "ERR wrong number of arguments*" {r lmpop 1}
+        assert_error "ERR wrong number of arguments*" {r lmpop 1 mylist{t}}
+
+        assert_error "ERR numkeys*" {r lmpop 0 mylist{t} LEFT}
+        assert_error "ERR numkeys*" {r lmpop a mylist{t} LEFT}
+        assert_error "ERR numkeys*" {r lmpop -1 mylist{t} RIGHT}
+
+        assert_error "ERR syntax error*" {r lmpop 1 mylist{t} bad_where}
+        assert_error "ERR syntax error*" {r lmpop 1 mylist{t} LEFT bar_arg}
+        assert_error "ERR syntax error*" {r lmpop 1 mylist{t} RIGHT LEFT}
+        assert_error "ERR syntax error*" {r lmpop 1 mylist{t} COUNT}
+        assert_error "ERR syntax error*" {r lmpop 2 mylist{t} mylist2{t} bad_arg}
+
+        assert_error "ERR count*" {r lmpop 1 mylist{t} LEFT COUNT 0}
+        assert_error "ERR count*" {r lmpop 1 mylist{t} RIGHT COUNT a}
+        assert_error "ERR count*" {r lmpop 1 mylist{t} LEFT COUNT -1}
+        assert_error "ERR count*" {r lmpop 2 mylist{t} mylist2{t} RIGHT COUNT -1}
+    }
+
+    test {LMPOP single existing list} {
+        # Same key multiple times.
+        create_list mylist{t} "a b c d e f"
+        assert_equal {mylist{t} {a b}} [r lmpop 2 mylist{t} mylist{t} left count 2]
+        assert_equal {mylist{t} {f e}} [r lmpop 2 mylist{t} mylist{t} right count 2]
+        assert_equal 2 [r llen mylist{t}]
+
+        # First one exists, second one does not exist.
+        create_list mylist{t} "a b c d e"
+        r del mylist2{t}
+        assert_equal {mylist{t} a} [r lmpop 2 mylist{t} mylist2{t} left count 1]
+        assert_equal 4 [r llen mylist{t}]
+        assert_equal {mylist{t} {e d c b}} [r lmpop 2 mylist{t} mylist2{t} right count 10]
+        assert_equal {} [r lmpop 2 mylist{t} mylist2{t} right count 1]
+
+        # First one does not exist, second one exists.
+        r del mylist{t}
+        create_list mylist2{t} "1 2 3 4 5"
+        assert_equal {mylist2{t} 5} [r lmpop 2 mylist{t} mylist2{t} right count 1]
+        assert_equal 4 [r llen mylist2{t}]
+        assert_equal {mylist2{t} {1 2 3 4}} [r lmpop 2 mylist{t} mylist2{t} left count 10]
+
+        assert_equal 0 [r exists mylist{t} mylist2{t}]
+    }
+
+    test {LMPOP multiple existing lists} {
+        create_list mylist{t} "a b c d e"
+        create_list mylist2{t} "1 2 3 4 5"
+
+        # Pop up from the first key.
+        assert_equal {mylist{t} {a b}} [r lmpop 2 mylist{t} mylist2{t} left count 2]
+        assert_equal 3 [r llen mylist{t}]
+        assert_equal {mylist{t} {e d c}} [r lmpop 2 mylist{t} mylist2{t} right count 3]
+        assert_equal 0 [r exists mylist{t}]
+
+        # Pop up from the second key.
+        assert_equal {mylist2{t} {1 2 3}} [r lmpop 2 mylist{t} mylist2{t} left count 3]
+        assert_equal 2 [r llen mylist2{t}]
+        assert_equal {mylist2{t} {5 4}} [r lmpop 2 mylist{t} mylist2{t} right count 2]
+        assert_equal 0 [r exists mylist{t}]
+
+        # Pop up all elements.
+        create_list mylist{t} "a b c"
+        create_list mylist2{t} "1 2 3"
+        assert_equal {mylist{t} {a b c}} [r lmpop 2 mylist{t} mylist2{t} left count 10]
+        assert_equal 0 [r llen mylist{t}]
+        assert_equal {mylist2{t} {3 2 1}} [r lmpop 2 mylist{t} mylist2{t} right count 10]
+        assert_equal 0 [r llen mylist2{t}]
+        assert_equal 0 [r exists mylist{t} mylist2{t}]
+    }
+
+    test {LMPOP propagate as pop with count command to replica} {
+        set repl [attach_to_replication_stream]
+
+        # left/right propagate as lpop/rpop with count
+        r lpush mylist{t} a b c
+
+        # Pop elements from one list.
+        r lmpop 1 mylist{t} left count 1
+        r lmpop 1 mylist{t} right count 1
+
+        # Now the list have only one element
+        r lmpop 2 mylist{t} mylist2{t} left count 10
+
+        # No elements so we don't propagate.
+        r lmpop 2 mylist{t} mylist2{t} left count 10
+
+        # Pop elements from the second list.
+        r rpush mylist2{t} 1 2 3
+        r lmpop 2 mylist{t} mylist2{t} left count 2
+        r lmpop 2 mylist{t} mylist2{t} right count 1
+
+        # Pop all elements.
+        r rpush mylist{t} a b c
+        r rpush mylist2{t} 1 2 3
+        r lmpop 2 mylist{t} mylist2{t} left count 10
+        r lmpop 2 mylist{t} mylist2{t} right count 10
+
+        assert_replication_stream $repl {
+            {select *}
+            {lpush mylist{t} a b c}
+            {lpop mylist{t} 1}
+            {rpop mylist{t} 1}
+            {lpop mylist{t} 1}
+            {rpush mylist2{t} 1 2 3}
+            {lpop mylist2{t} 2}
+            {rpop mylist2{t} 1}
+            {rpush mylist{t} a b c}
+            {rpush mylist2{t} 1 2 3}
+            {lpop mylist{t} 3}
+            {rpop mylist2{t} 3}
+        }
+    } {} {needs:repl}
+
     foreach {type large} [array get largevalue] {
         test "LRANGE basics - $type" {
             create_list mylist "$large 1 2 3 4 5 6 7 8 9"
@@ -1034,6 +1310,7 @@ start_server {
         r ping
     } {PONG}
 
+foreach {pop} {BLPOP BLMPOP_RIGHT} {
     test "client unblock tests" {
         r del l
         set rd [redis_deferring_client]
@@ -1041,19 +1318,19 @@ start_server {
         set id [$rd read]
 
         # test default args
-        $rd blpop l 0
+        bpop_command $rd $pop l 0
         wait_for_blocked_client
         r client unblock $id
         assert_equal {} [$rd read]
 
         # test with timeout
-        $rd blpop l 0
+        bpop_command $rd $pop l 0
         wait_for_blocked_client
         r client unblock $id TIMEOUT
         assert_equal {} [$rd read]
 
         # test with error
-        $rd blpop l 0
+        bpop_command $rd $pop l 0
         wait_for_blocked_client
         r client unblock $id ERROR
         catch {[$rd read]} e
@@ -1069,11 +1346,12 @@ start_server {
         assert_equal $e {invalid command name "0"}
 
         # finally, see the this client and list are still functional
-        $rd blpop l 0
+        bpop_command $rd $pop l 0
         wait_for_blocked_client
         r lpush l foo
         assert_equal {l foo} [$rd read]
     } {}
+}
 
     test {List ziplist of various encodings} {
         r del k