Fix issue #300

Former-commit-id: e9551c9e8d196f37e3742dfc7df824e164181d60

src/networking.cpp

@@ -2315,6 +2315,8 @@ void processInputBuffer(client *c, int callFlags) {
         /* Immediately abort if the client is in the middle of something. */
         if (c->flags & CLIENT_BLOCKED) break;
 
+        if (c->flags & CLIENT_EXECUTING_COMMAND) break;
+
         /* Don't process input from the master while there is a busy script
          * condition on the replica. We want just to accumulate the replication
          * stream (instead of replying -BUSY like we do with other clients) and
@@ -2349,13 +2351,16 @@ void processInputBuffer(client *c, int callFlags) {
         if (c->argc == 0) {
             resetClient(c);
         } else {
+            c->flags |= CLIENT_EXECUTING_COMMAND;
             /* We are finally ready to execute the command. */
             if (processCommandAndResetClient(c, callFlags) == C_ERR) {
                 /* If the client is no longer valid, we avoid exiting this
                  * loop and trimming the client buffer later. So we return
                  * ASAP in that case. */
+                c->flags &= ~CLIENT_EXECUTING_COMMAND;
                 return;
             }
+            c->flags &= ~CLIENT_EXECUTING_COMMAND;
         }
     }
 

src/server.h

@@ -441,10 +441,8 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
 #define CLIENT_PENDING_READ (1<<29) /* The client has pending reads and was put
                                        in the list of clients we can read
                                        from. */
-#define CLIENT_PENDING_COMMAND (1<<30) /* Used in threaded I/O to signal after
+#define CLIENT_EXECUTING_COMMAND (1<<30) /* Used to handle reentrency cases in processCommandWhileBlocked 
-                                          we return single threaded that the
+                                            to ensure we don't process a client already executing */
-                                          client has already pending commands
-                                          to be executed. */
 #define CLIENT_TRACKING (1ULL<<31) /* Client enabled keys tracking in order to
                                    perform client side caching. */
 #define CLIENT_TRACKING_BROKEN_REDIR (1ULL<<32) /* Target client is invalid. */

tests/unit/scripting.tcl

@@ -430,6 +430,21 @@ start_server {tags {"scripting"}} {
         set res
     } {102}
 
+    test {EVAL with pipelined command (No crash)} {
+        r flushall
+        r config set lua-time-limit 1
+        set rd [redis_deferring_client]
+        $rd eval {for i=1,1000000 do redis.call('set', i, 'sdfdsfd') end} 0
+        $rd set testkey foo
+        $rd get testkey
+        after 1200
+        catch {r echo "foo"} err
+        assert_match {BUSY*} $err
+        $rd read
+        $rd close
+    }
+
+
     test {EVAL timeout from AOF} {
         # generate a long running script that is propagated to the AOF as script
         # make sure that the script times out during loading