TCC: protect dictionaries from changing while accessed.

Certain Redis objects may change upon read only access. This is the
case, for instance, of hash tables, that may continue to incrementally
rehash after a rehashing operation. A similar problem also happens with
the PFCOUNT operation and other operations that may write as a side
effect of reading. In the case of PFCOUNT probably the right approach
would be to flag the command in a special way in the command table, so
that the operation is blocked as it if was a write operation.

src/blocked.c

@@ -144,26 +144,34 @@ void queueClientForReprocessing(client *c) {
 /* Unblock a client calling the right function depending on the kind
  * of operation the client is blocking for. */
 void unblockClient(client *c) {
-    if (c->btype == BLOCKED_LIST ||
-        c->btype == BLOCKED_ZSET ||
-        c->btype == BLOCKED_STREAM) {
+    int btype = c->btype;
+
+    if (btype == BLOCKED_LIST ||
+        btype == BLOCKED_ZSET ||
+        btype == BLOCKED_STREAM) {
         unblockClientWaitingData(c);
-    } else if (c->btype == BLOCKED_WAIT) {
+    } else if (btype == BLOCKED_WAIT) {
         unblockClientWaitingReplicas(c);
-    } else if (c->btype == BLOCKED_MODULE) {
+    } else if (btype == BLOCKED_MODULE) {
         if (moduleClientIsBlockedOnKeys(c)) unblockClientWaitingData(c);
         unblockClientFromModule(c);
+    } else if (btype == BLOCKED_LOCK) {
+        /* We'll handle it at the end of the function. */
     } else {
         serverPanic("Unknown btype in unblockClient().");
     }
     /* Clear the flags, and put the client in the unblocked list so that
      * we'll process new commands in its query buffer ASAP. */
     server.blocked_clients--;
-    server.blocked_clients_by_type[c->btype]--;
+    server.blocked_clients_by_type[btype]--;
     c->flags &= ~CLIENT_BLOCKED;
     c->btype = BLOCKED_NONE;
     removeClientFromTimeoutTable(c);
     queueClientForReprocessing(c);
+
+    /* Re-execute the command if it was not executed at all since the
+     * client was blocked because of key locks. */
+    if (btype == BLOCKED_LOCK) processCommandAndResetClient(c);
 }
 
 /* This function gets called when a blocked client timed out in order to
@@ -172,7 +180,8 @@ void unblockClient(client *c) {
 void replyToBlockedClientTimedOut(client *c) {
     if (c->btype == BLOCKED_LIST ||
         c->btype == BLOCKED_ZSET ||
-        c->btype == BLOCKED_STREAM) {
+        c->btype == BLOCKED_STREAM ||
+        c->btype == BLOCKED_LOCK) { /* type LOCK never timeouts actually. */
         addReplyNullArray(c);
     } else if (c->btype == BLOCKED_WAIT) {
         addReplyLongLong(c,replicationCountAcksByOffset(c->bpop.reploffset));

src/db.c

@@ -1749,3 +1749,249 @@ unsigned int delKeysInSlot(unsigned int hashslot) {
 unsigned int countKeysInSlot(unsigned int hashslot) {
     return server.cluster->slots_keys_count[hashslot];
 }
+
+/*-----------------------------------------------------------------------------
+ * Locked keys API
+ *----------------------------------------------------------------------------*/
+
+int lockedClientListMatch(void *a, void *b) {
+    lockedKeyClient *ca = a, *cb = b;
+    return ca->id == cb->id;
+}
+
+/* Lock the specified key, in the specified DB and in the context of the
+ * specified client, and return the locked object by reference.
+ * The type of the lock can be KEYLOCK_READ or KEYLOCK_WRITE (XXX: not supported
+ * right now).
+ *
+ * Locking the key may fail because the key is already locked or because
+ * Redis is not able to lock keys in a given moment (because it is waiting
+ * to fork or for other reasons). When this happens, the command should
+ * signal to the caller that it requires to be rescheduled, or alternatively
+ * may implement the operation synchronously without locking any key.
+ *
+ * When locking fails C_ERR is returned, otherwise C_OK is returned.
+ * Note that 'optr' may be populated with NULL if the key that is going
+ * to be locked is empty. If 'optr' is NULL, the object is not returned.
+ * This is often useful since we lookup the key and already have the object
+ * and later lock the key if needed (for instance if the value is large).
+ *
+ * Signaling of the need to lock keys is performed using the
+ * wouldLockKey() API (XXX: yet to be implemented). Check the function
+ * documentation for more information. */
+int lockKey(client *c, robj *key, int locktype, robj **optr) {
+    dictEntry *de;
+
+    de = dictFind(c->db->locked_keys,key);
+    lockedKey *lk = de ? dictGetVal(de) : NULL;
+
+    if (lk == NULL) {
+        lk = zmalloc(sizeof(*lk));
+        lk->lock_type = locktype;
+        lk->waiting = listCreate();
+        lk->owners = listCreate();
+        listSetMatchMethod(lk->waiting,lockedClientListMatch);
+        listSetMatchMethod(lk->owners,lockedClientListMatch);
+        listSetFreeMethod(lk->waiting,zfree);
+        listSetFreeMethod(lk->owners,zfree);
+        lk->obj = lookupKeyReadWithFlags(c->db,key,LOOKUP_NOTOUCH);
+        dictAdd(c->db->locked_keys,key,lk);
+        incrRefCount(key);
+
+        /* Make the object immutable. In the trivial case of hash tables
+         * we just increment the iterators count, to prevent rehashing
+         * when the object is accessed in read-only. */
+        if (lk->obj) {
+            if (lk->obj->encoding == OBJ_ENCODING_HT) {
+                ((dict*)lk->obj->ptr)->iterators++;
+            } else if (lk->obj->encoding == OBJ_ENCODING_SKIPLIST) {
+                ((zset*)lk->obj->ptr)->dict->iterators++;
+            }
+        }
+    } else {
+        /* If there is already a lock, it is incompatible with a new lock
+         * both in the case the lock is of write type, or we want to lock
+         * for write. */
+        if (lk->lock_type == LOCKEDKEY_WRITE || locktype == LOCKEDKEY_WRITE)
+            return C_ERR;
+    }
+
+    /* We need to remember that this client locked this key: for
+     * now we set this information in the real client locking the key,
+     * however later this information is moved into the blocked client
+     * handle that we use for threaded execution. The reason is that we
+     * want to unlock the keys back when the thread has finished, not just
+     * when this client disconnects. */
+    if (c->locked == NULL)
+        c->locked = dictCreate(&objectKeyPointerValueDictType,NULL);
+
+    de = dictAddRaw(c->locked,key,NULL);
+    if (de == NULL) {
+        /* The key is already locked for this client? Odd but let's allow
+         * that without crashing. So that command implementations don't
+         * have to do complex checks in the command line in case of repeating
+         * keys. */
+        if (optr) *optr = lk->obj;
+        return C_OK;
+    }
+    incrRefCount(key);
+
+    /* Lock allowed: put the client in the list of owners. */
+    lockedKeyClient *lkc = zmalloc(sizeof(*lkc));
+    lkc->id = c->id;
+    listAddNodeTail(lk->owners,lkc);
+
+   /* Remember the client ID of the lock, so if we move this in the
+     * blocked handle, we can still track the original ID even if the
+     * client is gone. */
+    dictSetUnsignedIntegerVal(de,c->id);
+    if (optr) *optr = lk->obj;
+    return C_OK;
+}
+
+/* If the key this client is trying to access is locked, put it into
+ * its waiting list and return 1, otherwise return 0.
+ * If queue is zero, the lock is just tested without really putting the
+ * client it the waiting list of the locked key. */
+int queueClientIfKeyIsLocked(client *c, robj *key, int queue) {
+    lockedKey *lk = dictFetchValue(c->db->locked_keys,key);
+    if (lk == NULL) return 0;
+
+    if (queue) {
+        lockedKeyClient *lkc = zmalloc(sizeof(*lkc));
+        lkc->id = c->id;
+        listAddNodeTail(lk->waiting,lkc);
+    }
+    return 1;
+}
+
+/* Test if the client is executing a command that should be blocked because
+ * the command access mode is incompatible with some lock of the keys it
+ * is going to access. If the client should be blocked the functionm returns
+ * 1, otherwise 0 is returned.
+ *
+ * If 'queue' is true, when the funciton returns 1 the client is put on the
+ * waiting list of the locked keys it is trying to access, otherwise
+ * the function will just test the condition. */
+int clientShouldWaitOnLockedKeys(client *c, int queue) {
+    /* Don't use any CPU time if we have no locked keys at all. */
+    if (dictSize(c->db->locked_keys) == 0) return 0;
+
+    /* For now we just have read locks, if the client wants to execute
+     * a command that does not write to the key, it should be allowed to do
+     * so: our threads are just reading from the locked objects. */
+    if (!(c->cmd->flags & CMD_WRITE) &&
+        c->cmd->proc != evalCommand &&
+        c->cmd->proc != evalShaCommand &&
+        c->cmd->proc != execCommand)
+    {
+        return 0;
+    }
+
+    int locked = 0, numkeys, *keyidx;
+    if (c->flags & CLIENT_MULTI) {
+        for (int j = 0; j < c->mstate.count; j++) {
+            struct redisCommand *cmd = c->mstate.commands[j].cmd;
+            robj **argv = c->mstate.commands[j].argv;
+            int argc = c->mstate.commands[j].argc;
+
+            /* Even in the case of transactions, read only commands
+             * do not require we to sleep for the locked keys. */
+            if (!(cmd->flags & CMD_WRITE) &&
+                cmd->proc != evalCommand &&
+                cmd->proc != evalShaCommand) continue;
+
+            keyidx = getKeysFromCommand(cmd,argv,argc,&numkeys);
+            for (int j = 0; j < numkeys; j++) {
+                if (queueClientIfKeyIsLocked(c,argv[keyidx[j]],queue))
+                    locked++;
+            }
+            getKeysFreeResult(keyidx);
+        }
+    } else {
+        keyidx = getKeysFromCommand(c->cmd,c->argv,c->argc,&numkeys);
+        for (int j = 0; j < numkeys; j++) {
+            if (queueClientIfKeyIsLocked(c,c->argv[keyidx[j]],queue))
+                locked++;
+        }
+        getKeysFreeResult(keyidx);
+    }
+
+    /* Lock the client and return if we are waiting for at least one
+     * key. */
+    return locked;
+}
+
+/* Unlock the specified key in the context of the specified client.
+ * This API should not be called by the command implementation itself,
+ * instead unlockAllKeys() should be used. */
+void unlockKey(client *c, robj *key, uint64_t owner_id) {
+    lockedKey *lk = dictFetchValue(c->db->locked_keys,key);
+    if (lk == NULL) return;
+    struct lockedKeyClient lkc = {owner_id};
+    listNode *ln = listSearchKey(lk->owners,&lkc);
+    if (ln == NULL) return;
+
+    /* Remove the lock for thsi owner, and if this key dropped to a locked
+     * count of zero, we need to resume the clients in the waiting list. */
+    listDelNode(lk->owners,ln);
+    if (listLength(lk->owners) != 0) return;
+
+    /* There are no longer owners for this lock: unlock the key
+     * before resuming the clients. */
+    dictDelete(c->db->locked_keys,key);
+
+    listIter li;
+    listRewind(lk->waiting,&li);
+    while((ln = listNext(&li))) {
+        lockedKeyClient *lkc = listNodeValue(ln);
+        client *blocked = lookupClientByID(lkc->id);
+        if (blocked != NULL &&
+            blocked->flags & CLIENT_BLOCKED &&
+            blocked->btype == BLOCKED_LOCK)
+        {
+            /* XXX: Check if the client is yet locked in some other key
+             * before unblocking it? Right now we just unblock it, and
+             * it will get blocked again if there are yet keys that are
+             * locked for it. */
+            unblockClient(blocked);
+        }
+        listDelNode(lk->waiting,ln);
+    }
+
+    /* If we modified the object in order to make it immutable during
+     * read operations, restore it in its normal state. */
+    if (lk->obj) {
+        if (lk->obj->encoding == OBJ_ENCODING_HT) {
+            ((dict*)lk->obj->ptr)->iterators--;
+        } else if (lk->obj->encoding == OBJ_ENCODING_SKIPLIST) {
+            ((zset*)lk->obj->ptr)->dict->iterators--;
+        }
+    }
+
+    /* Free the lock state. */
+    listRelease(lk->owners);
+    listRelease(lk->waiting);
+    zfree(lk);
+}
+
+/* Unlock all the keys locked in the specified client. We need to pass
+ * the client ID of the original client that locked the key, since the
+ * list of locked keys may be transfered to the blocked client handle.
+ * As a side effect the function also will release the dictionary of
+ * locked keys. */
+void unlockAllKeys(client *c) {
+    if (c->locked == NULL) return;
+
+    dictIterator *di;
+    dictEntry *de;
+    di = dictGetSafeIterator(c->locked);
+    while((de = dictNext(di)) != NULL) {
+        uint64_t owner_id = dictGetUnsignedIntegerVal(de);
+        robj *key = dictGetKey(de);
+        unlockKey(c,key,owner_id);
+    }
+    dictReleaseIterator(di);
+    dictRelease(c->locked);
+    c->locked = NULL;
+}

src/debug.c

@@ -367,6 +367,14 @@ void mallctl_string(client *c, robj **argv, int argc) {
 }
 #endif
 
+/* Threaded half of the DEBUG LOCK command. */
+void debugLockThreadedHalf(client *c, void *opt) {
+    mstime_t sleeptime = (unsigned long)opt;
+    usleep(sleeptime*1000);
+    addReplyStatusFormat(c,"Slept for %lu milliseconds",
+        (unsigned long)sleeptime);
+}
+
 void debugCommand(client *c) {
     if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
         const char *help[] = {
@@ -396,6 +404,7 @@ void debugCommand(client *c) {
 "STRUCTSIZE -- Return the size of different Redis core C structures.",
 "ZIPLIST <key> -- Show low level info about the ziplist encoding.",
 "STRINGMATCH-TEST -- Run a fuzz tester against the stringmatchlen() function.",
+"LOCK <milliseconds> <type> <key1> ... <keyN>  -- Lock the specified keys for the specified amount of milliseconds. Lock type should be read or write.",
 #ifdef USE_JEMALLOC
 "MALLCTL <key> [<val>] -- Get or set a malloc tunning integer.",
 "MALLCTL-STR <key> [<val>] -- Get or set a malloc tunning string.",
@@ -791,11 +800,33 @@ NULL
     {
         stringmatchlen_fuzz_test();
         addReplyStatus(c,"Apparently Redis did not crash: test passed");
+    } else if (!strcasecmp(c->argv[1]->ptr,"lock") && c->argc >= 5) {
+        /* DEBUG LOCK <ms> <type> ... keys ... */
+        long ms;
+        if (getLongFromObjectOrReply(c,c->argv[2],&ms,NULL) != C_OK) return;
+        if (!strcasecmp(c->argv[3]->ptr,"read")) {
+            /* The only one supported so far... */
+        } else {
+            addReplyError(c,"DEBUG LOCK only supports read locks so far.");
+        }
+        /* Try locking all the keys. */
+        for (int j = 4; j < c->argc; j++) {
+            robj *myobj;
+            if (lockKey(c,c->argv[j],LOCKEDKEY_READ,&myobj) == C_ERR) {
+                unlockAllKeys(c);
+                addReplyErrorFormat(c,"Unable to lock %s. Already locked?",
+                                    c->argv[j]->ptr);
+                return;
+            }
+        }
+        /* Pass control to the threaded part. */
+        executeThreadedCommand(c,debugLockThreadedHalf,
+                               (void*)(unsigned long)ms);
 #ifdef USE_JEMALLOC
-    } else if(!strcasecmp(c->argv[1]->ptr,"mallctl") && c->argc >= 3) {
+    } else if (!strcasecmp(c->argv[1]->ptr,"mallctl") && c->argc >= 3) {
         mallctl_int(c, c->argv+2, c->argc-2);
         return;
-    } else if(!strcasecmp(c->argv[1]->ptr,"mallctl-str") && c->argc >= 3) {
+    } else if (!strcasecmp(c->argv[1]->ptr,"mallctl-str") && c->argc >= 3) {
         mallctl_string(c, c->argv+2, c->argc-2);
         return;
 #endif

src/module.c

@@ -369,6 +369,15 @@ typedef struct RedisModuleUser {
     user *user; /* Reference to the real redis user */
 } RedisModuleUser;
 
+/* The Redis core itself registeres a module in order to use module services
+ * for things like blocking clients for threaded execution of slow commands. */
+RedisModule *coremodule;
+
+/* The CoreModuleBlockedClients counter is the number of clients currently
+ * blocked on a threaded commad: this allows us to limit the maximum number
+ * of threaded executions at a given time. */
+_Atomic unsigned long CoreModuleBlockedClients = 0;
+unsigned long CoreModuleThreadsMax = 50;
 
 /* --------------------------------------------------------------------------
  * Prototypes
@@ -4373,7 +4382,7 @@ RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdF
     bc->unblocked = 0;
     c->bpop.timeout = timeout;
 
-    if (islua || ismulti) {
+    if (ctx->module != coremodule && (islua || ismulti)) {
         c->bpop.module_blocked_handle = NULL;
         addReplyError(c, islua ?
             "Blocking module command called from Lua script" :
@@ -4570,12 +4579,23 @@ int RM_UnblockClient(RedisModuleBlockedClient *bc, void *privdata) {
     return REDISMODULE_OK;
 }
 
+/* Low level API to abort a blocked client operation. After we call
+ * moduleBlockClient(), if later there aren't the condition for really
+ * going forward with the plan of blocking (for instance the pthread
+ * API returns a failure creating a thread), we may "undo" the operation
+ * and avoid any callback to be called, with the exception of the
+ * free_privdata callback (in case we specify a privdata that is
+ * not NULL). */
+int moduleAbortBlock(RedisModuleBlockedClient *bc, void *privdata) {
+    bc->reply_callback = NULL;
+    bc->disconnect_callback = NULL;
+    return RM_UnblockClient(bc,privdata);
+}
+
 /* Abort a blocked client blocking operation: the client will be unblocked
  * without firing any callback. */
 int RM_AbortBlock(RedisModuleBlockedClient *bc) {
-    bc->reply_callback = NULL;
-    bc->disconnect_callback = NULL;
-    return RM_UnblockClient(bc,NULL);
+    return moduleAbortBlock(bc,NULL);
 }
 
 /* Set a callback that will be called if a blocked client disconnects
@@ -4607,6 +4627,7 @@ void RM_SetDisconnectCallback(RedisModuleBlockedClient *bc, RedisModuleDisconnec
  * When this happens the RedisModuleBlockedClient structure in the queue
  * will have the 'client' field set to NULL. */
 void moduleHandleBlockedClients(void) {
+    if (moduleCount() == 0 && CoreModuleBlockedClients == 0) return;
     listNode *ln;
     RedisModuleBlockedClient *bc;
 
@@ -7197,6 +7218,133 @@ void processModuleLoadingProgressEvent(int is_aof) {
     }
 }
 
+/* --------------------------------------------------------------------------
+ * Module support for Redis core operations
+ *
+ * Here we define the interfaces that the core itself can use when
+ * using abstractions implemented by the modules interface, like the
+ * ability to run threaded operations.
+ * -------------------------------------------------------------------------- */
+typedef void (*coreThreadedCommandCallback)(client *c, void *options);
+
+/* This sturcture is used in order to pass state from the main thread to
+ * the thread that will execute the command, and then to the function
+ * that is responsible of freeing the private data. */
+typedef struct {
+    RedisModuleBlockedClient *bc;           /* Blocked client handle. */
+    coreThreadedCommandCallback callback;   /* Threaded command callback. */
+    void *options;                          /* A pointer with details about
+                                               the command to execute. Passed
+                                               to the callback. */
+} tcprivdata;
+
+/* Free the private data allocated for a client blocked because of a
+ * threaded Redis core command execution. */
+void threadedCoreCommandFreePrivdata(RedisModuleCtx *ctx, void *privdata) {
+    UNUSED(ctx);
+    tcprivdata *tcpd = privdata;
+
+    unlockAllKeys(tcpd->bc->reply_client);
+    zfree(privdata);
+    CoreModuleBlockedClients--;
+}
+
+/* The entry point of the threaded execution. */
+void *threadedCoreCommandEnty(void *argptr) {
+    tcprivdata *tcpd = argptr;
+    /* We use the blocked command handle reply client as client to pass to the
+     * command implementation: this is a client that is not referenced
+     * at all in the Redis main thread, it just holds a copy of the
+     * original command execution argv/argc pair. This way the command can
+     * use Redis low level API to accumulate the reply there, and later when
+     * the client is unblocked, the reply will be concatenated to the
+     * real client. */
+    tcpd->callback(tcpd->bc->reply_client,tcpd->options);
+    moduleUnblockClientByHandle(tcpd->bc,tcpd);
+    return NULL;
+}
+
+/* This function is called by the Redis core when we want to execute the
+ * actual implementation of a long running command in a thread. The callback
+ * will have many limits: it cannot touch the keyspace, but can only work
+ * with the passed arguments (that are stored by copying them).
+ *
+ * This strategy works well when the input is small (fast to copy) compared
+ * to the total execution time of the command. In the future the ability to
+ * lock specific keys in the main thread will also be implemented, so that
+ * the threaded part of a command can also operate on key values without
+ * any race condition.
+ *
+ * 'options' is some private information passed to the callback that
+ * may contain parameters that the "main thread half" if the command
+ * parsed from the command line of the command. Normally it will pass
+ * things like the options in order to execute the command, the objects
+ * the command should work with (that may be copies or made inaccessible
+ * by the main thread via other means), and so forth.
+ *
+ * The info pointed by 'options' must be allocated by the callback implementing
+ * the threaded half of the command once it's done. */
+void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, void *options) {
+    /* We have a fast path in case we are not in a situation where
+     * threaded execution is allowed: we can just call the callback
+     * synchronously and return. A note about server.loading: this is
+     * in case a threaded command is loaded via the AOF file. */
+    int islua = c->flags & CLIENT_LUA;
+    int ismulti = c->flags & CLIENT_MULTI;
+    if (islua || ismulti || server.loading ||
+        CoreModuleBlockedClients >= CoreModuleThreadsMax)
+    {
+        callback(c,options);
+        return;
+    }
+
+    RedisModuleCtx ctx = REDISMODULE_CTX_INIT;
+    ctx.module = coremodule;
+    ctx.client = c;
+
+    /* We pass the arguments to the thread entry point using an array
+     * of void pointers. */
+    tcprivdata *tcpd = zmalloc(sizeof(*tcpd));
+
+    /* Block the client. */
+    CoreModuleBlockedClients++;
+    RedisModuleBlockedClient *bc = RM_BlockClient(&ctx,NULL,NULL,threadedCoreCommandFreePrivdata,0);
+    tcpd->bc = bc;
+    tcpd->callback = callback;
+    tcpd->options = options;
+    bc->privdata = tcpd;
+
+    /* Move the list of locked keys in the reply client of the
+     * blocked handle: this way if the real client is freed because of
+     * disconnection, we still unlock the keys in the right moment, that
+     * is once the thread returns. */
+    selectDb(bc->reply_client,c->db->id);
+    bc->reply_client->locked = c->locked;
+    c->locked = NULL;
+
+    /* We need the thread to work on a copy of the client argument
+     * vector. */
+    client *copy = bc->reply_client;
+    copy->argc = c->argc;
+    copy->argv = zmalloc(sizeof(robj*) * c->argc);
+    for (int j = 0; j < c->argc; j++)
+        copy->argv[j] = createStringObject(c->argv[j]->ptr,
+                                           sdslen(c->argv[j]->ptr));
+
+    /* Try to spawn the thread that will actually execute the command. */
+    pthread_t tid;
+    if (pthread_create(&tid,NULL,threadedCoreCommandEnty,tcpd) != 0) {
+        /* Execute the command synchronously if we can't spawn a thread.. */
+        callback(c,tcpd->options);
+        moduleAbortBlock(bc,tcpd);
+    }
+    moduleFreeContext(&ctx);
+}
+
+unsigned long runningThreadedCommandsCount(void) {
+    return CoreModuleBlockedClients;
+}
+
 /* --------------------------------------------------------------------------
  * Modules API internals
  * -------------------------------------------------------------------------- */
@@ -7267,6 +7415,12 @@ void moduleInitModulesSystem(void) {
     /* Our thread-safe contexts GIL must start with already locked:
      * it is just unlocked when it's safe. */
     pthread_mutex_lock(&moduleGIL);
+
+    /* Create the Redis core module. */
+    RedisModuleCtx ctx = REDISMODULE_CTX_INIT;
+    RM_SetModuleAttribs(&ctx,"rediscore",1,1);
+    coremodule = ctx.module;
+    moduleFreeContext(&ctx);
 }
 
 /* Load all the modules in the server.loadmodule_queue list, which is

src/networking.c

@@ -153,6 +153,7 @@ client *createClient(connection *conn) {
     c->watched_keys = listCreate();
     c->pubsub_channels = dictCreate(&objectKeyPointerValueDictType,NULL);
     c->pubsub_patterns = listCreate();
+    c->locked = NULL;
     c->peerid = NULL;
     c->client_list_node = NULL;
     c->client_tracking_redirection = 0;
@@ -1155,6 +1156,17 @@ void freeClient(client *c) {
     unwatchAllKeys(c);
     listRelease(c->watched_keys);
 
+    /* Unlock the keys and free the dictionary of locked keys.
+     * Note that if the client disconnects while there is a thread executing
+     * a threaded command for it, actually the locked keys were passed to
+     * the blocked client handle, so c->locked will be NULL, and the keys
+     * will really be unlocked when the thread returns instead.
+     *
+     * However we may get here with locked keys in case for some reason
+     * the client didn't call executeThreadedCommand() after locking the
+     * key. */
+    unlockAllKeys(c);
+
     /* Unsubscribe from all the pubsub channels */
     pubsubUnsubscribeAllChannels(c,0);
     pubsubUnsubscribeAllPatterns(c,0);
@@ -1748,9 +1760,12 @@ void commandProcessed(client *c) {
     /* Don't reset the client structure for clients blocked in a
      * module blocking command, so that the reply callback will
      * still be able to access the client argv and argc field.
-     * The client will be reset in unblockClientFromModule(). */
+     * The client will be reset in unblockClientFromModule().
+     * Don't reset the client for clients waiting to be re-executed
+     * with the current command line because they are waiting for
+     * locked keys. */
     if (!(c->flags & CLIENT_BLOCKED) ||
-        c->btype != BLOCKED_MODULE)
+        (c->btype != BLOCKED_MODULE && c->btype != BLOCKED_LOCK))
     {
         resetClient(c);
     }

src/scripting.c

@@ -622,7 +622,10 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
 
     /* Write commands are forbidden against read-only slaves, or if a
      * command marked as non-deterministic was already called in the context
-     * of this script. */
+     * of this script. Finally writing to keys that are locked is not correct:
+     * normally the script would be blocked before being executed if the
+     * keys are all declared by EVAL/EVALSHA, but scripts can easily escape
+     * this protocol: we need to return an error in such case. */
     if (cmd->flags & CMD_WRITE) {
         int deny_write_type = writeCommandsDeniedByDiskError();
         if (server.lua_random_dirty && !server.lua_replicate_commands) {
@@ -646,6 +649,11 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
                 sdsfree(aof_write_err);
             }
             goto cleanup;
+        } else if (clientShouldWaitOnLockedKeys(c,0)) {
+            luaPushError(lua,
+                "-LOCKED Lua script is trying to access locked keys "
+                "that were not specified in the EVAL keys list\r\n");
+            goto cleanup;
         }
     }
 

src/server.c

@@ -2151,7 +2151,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
 
     /* Check if there are clients unblocked by modules that implement
      * blocking commands. */
-    if (moduleCount()) moduleHandleBlockedClients();
+    moduleHandleBlockedClients();
 
     /* Try to process pending commands for clients that were just unblocked. */
     if (listLength(server.unblocked_clients))
@@ -2820,6 +2820,7 @@ void initServer(void) {
         server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
         server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL);
         server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);
+        server.db[j].locked_keys = dictCreate(&objectKeyPointerValueDictType,NULL);
         server.db[j].id = j;
         server.db[j].avg_ttl = 0;
         server.db[j].defrag_later = listCreate();
@@ -3499,6 +3500,14 @@ int processCommand(client *c) {
         }
     }
 
+    /* We want to block this client if the keys it is going to access
+     * are locked, and the operation the client is going to perform
+     * is not compatible with the key lock type. */
+    if (clientShouldWaitOnLockedKeys(c,1)) {
+        blockClient(c,BLOCKED_LOCK);
+        return C_OK;
+    }
+
     /* Handle the maxmemory directive.
      *
      * Note that we do not want to reclaim memory if we are here re-entering
@@ -4068,11 +4077,13 @@ sds genRedisInfoString(const char *section) {
             "client_recent_max_output_buffer:%zu\r\n"
             "blocked_clients:%d\r\n"
             "tracking_clients:%d\r\n"
+            "blocked_for_core_thread:%lu\r\n"
             "clients_in_timeout_table:%llu\r\n",
             listLength(server.clients)-listLength(server.slaves),
             maxin, maxout,
             server.blocked_clients,
             server.tracking_clients,
+            runningThreadedCommandsCount(),
             (unsigned long long) raxSize(server.clients_timeout_table));
     }
 

src/server.h

@@ -265,7 +265,8 @@ typedef long long ustime_t; /* microsecond time type. */
 #define BLOCKED_MODULE 3  /* Blocked by a loadable module. */
 #define BLOCKED_STREAM 4  /* XREAD. */
 #define BLOCKED_ZSET 5    /* BZPOP et al. */
-#define BLOCKED_NUM 6     /* Number of blocked states. */
+#define BLOCKED_LOCK 6    /* Waiting for locked keys. */
+#define BLOCKED_NUM 7     /* Number of blocked states. */
 
 /* Client request types */
 #define PROTO_REQ_INLINE 1
@@ -645,14 +646,55 @@ typedef struct redisDb {
     dict *dict;                 /* The keyspace for this DB */
     dict *expires;              /* Timeout of keys with a timeout set */
     dict *blocking_keys;        /* Keys with clients waiting for data (BLPOP)*/
-    dict *ready_keys;           /* Blocked keys that received a PUSH */
+    dict *ready_keys;           /* Blocked keys that received a write */
     dict *watched_keys;         /* WATCHED keys for MULTI/EXEC CAS */
+    dict *locked_keys;          /* Keys locked for threaded operations */
     int id;                     /* Database ID */
     long long avg_ttl;          /* Average TTL, just for stats */
     unsigned long expires_cursor; /* Cursor of the active expire cycle. */
     list *defrag_later;         /* List of key names to attempt to defrag one by one, gradually. */
 } redisDb;
 
+#define LOCKEDKEY_READ  0       /* Key is locked in read mode. The list of
+                                   current_owners may contain multiple
+                                   clients. */
+#define LOCKEDKEY_WRITE  1      /* Key is locked in write mode. There is only
+                                   one client in the list of owners. */
+#define LOCKEDKEY_WAIT   2      /* Key is not locked, but we need to fork or
+                                   to perform a full scan operation such as
+                                   DEBUG DIGEST, so we are not allowing keys
+                                   to be locked. The owner list is empty while
+                                   the wait queue is populated by clients that
+                                   want to lock the key. */
+
+/* This structure is stored at db->locked_keys for each key that has some
+ * lock or waiting list active. */
+typedef struct lockedKey {
+    int lock_type;              /* LOCKEDKEY_* defines above. */
+    int deleted;                /* True if the key was deleted in the
+                                   main thread while locked: in this case
+                                   it is up to us to free the object once
+                                   all the owners will return. Moreover when
+                                   this happens, clients in the wait queue can
+                                   be rescheduled ASAP. */
+    robj *obj;                  /* The locked object itself. May be NULL
+                                   if the key was empty at the time we
+                                   locked the key. */
+    list *owners;               /* List of clients that locked this key
+                                   and are doing threaded operations. See
+                                   the LOCKEDKEY defines comments for more
+                                   information. */
+    list *waiting;              /* Clients that are waiting for the key
+                                   to return available in order to be
+                                   rescheduled. */
+} lockedKey;
+
+/* This is the structure we put in the onwers and wait lists of the
+ * locked keys. */
+typedef struct lockedKeyClient {
+    uint64_t id;
+} lockedKeyClient;
+
 /* Client MULTI/EXEC state */
 typedef struct multiCmd {
     robj **argv;
@@ -820,6 +862,7 @@ typedef struct client {
     list *watched_keys;     /* Keys WATCHED for MULTI/EXEC CAS */
     dict *pubsub_channels;  /* channels a client is interested in (SUBSCRIBE) */
     list *pubsub_patterns;  /* patterns a client is interested in (SUBSCRIBE) */
+    dict *locked;           /* Dictionary of keys locked by the client. */
     sds peerid;             /* Cached peer ID. */
     listNode *client_list_node; /* list node in client list */
     RedisModuleUserChangedFunc auth_callback; /* Module callback to execute
@@ -1581,6 +1624,14 @@ void moduleUnblockClient(client *c);
 int moduleClientIsBlockedOnKeys(client *c);
 void moduleNotifyUserChanged(client *c);
 
+/* Modules functionalities exported to core commands. */
+typedef void (*coreThreadedCommandCallback)(client *c, void *options);
+void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, void *options);
+unsigned long runningThreadedCommandsCount(void);
+int lockKey(client *c, robj *key, int locktype, robj **optr);
+int clientShouldWaitOnLockedKeys(client *c, int queue);
+void unlockAllKeys(client *c);
+
 /* Utils */
 long long ustime(void);
 long long mstime(void);
@@ -1955,6 +2006,7 @@ size_t freeMemoryGetNotCountedMemory();
 int freeMemoryIfNeeded(void);
 int freeMemoryIfNeededAndSafe(void);
 int processCommand(client *c);
+int processCommandAndResetClient(client *c);
 void setupSignalHandlers(void);
 struct redisCommand *lookupCommand(sds name);
 struct redisCommand *lookupCommandByCString(char *s);

src/t_list.c

@@ -395,6 +395,38 @@ void rpopCommand(client *c) {
     popGenericCommand(c,LIST_TAIL);
 }
 
+struct lrangeThreadOptions {
+    long start, rangelen;
+    robj *o;
+};
+
+void lrangeThreadedHalf(client *c, void *options) {
+    struct lrangeThreadOptions *opt = options;
+
+    /* Return the result in form of a multi-bulk reply */
+    addReplyArrayLen(c,opt->rangelen);
+    if (opt->o->encoding == OBJ_ENCODING_QUICKLIST) {
+        listTypeIterator *iter = listTypeInitIterator(opt->o, opt->start,
+                                                      LIST_TAIL);
+
+        while(opt->rangelen--) {
+            listTypeEntry entry;
+            listTypeNext(iter, &entry);
+            quicklistEntry *qe = &entry.entry;
+            if (qe->value) {
+                addReplyBulkCBuffer(c,qe->value,qe->sz);
+            } else {
+                addReplyBulkLongLong(c,qe->longval);
+            }
+        }
+        listTypeReleaseIterator(iter);
+    } else {
+        serverPanic("List encoding is not QUICKLIST!");
+    }
+    zfree(options);
+}
+
+#define LRANGE_THREADED_THRESHOLD 1000
 void lrangeCommand(client *c) {
     robj *o;
     long start, end, llen, rangelen;
@@ -420,24 +452,18 @@ void lrangeCommand(client *c) {
     if (end >= llen) end = llen-1;
     rangelen = (end-start)+1;
 
-    /* Return the result in form of a multi-bulk reply */
-    addReplyArrayLen(c,rangelen);
-    if (o->encoding == OBJ_ENCODING_QUICKLIST) {
-        listTypeIterator *iter = listTypeInitIterator(o, start, LIST_TAIL);
-
-        while(rangelen--) {
-            listTypeEntry entry;
-            listTypeNext(iter, &entry);
-            quicklistEntry *qe = &entry.entry;
-            if (qe->value) {
-                addReplyBulkCBuffer(c,qe->value,qe->sz);
-            } else {
-                addReplyBulkLongLong(c,qe->longval);
-            }
-        }
-        listTypeReleaseIterator(iter);
+    struct lrangeThreadOptions *opt = zmalloc(sizeof(*opt));
+    opt->start = start;
+    opt->rangelen = rangelen;
+    if (llen <= LRANGE_THREADED_THRESHOLD ||
+        lockKey(c,c->argv[1],LOCKEDKEY_READ,&opt->o) == C_ERR)
+    {
+        /* In the case of LRANGE, we execute the command synchronously
+         * if we are unable to get a lock. */
+        opt->o = o;
+        lrangeThreadedHalf(c,opt);
     } else {
-        serverPanic("List encoding is not QUICKLIST!");
+        executeThreadedCommand(c, lrangeThreadedHalf, opt);
     }
 }
 

src/t_set.c

@@ -29,12 +29,16 @@
 
 #include "server.h"
 
+/* Use threaded execution, when available, if at least one of the sets
+ * involved is greater or equal to the threshold. */
+#define THREADED_SET_SIZE_THRESHOLD 5000
+
 /*-----------------------------------------------------------------------------
  * Set Commands
  *----------------------------------------------------------------------------*/
 
 void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
-                              robj *dstkey, int op);
+                              robj *dstkey, int op, int sync);
 
 /* Factory method to return a set that *can* hold "value". When the object has
  * an integer-encodable value, an intset will be returned. Otherwise a regular
@@ -436,7 +440,7 @@ void spopWithCountCommand(client *c) {
      * the number of elements inside the set: simply return the whole set. */
     if (count >= size) {
         /* We just return the entire set */
-        sunionDiffGenericCommand(c,c->argv+1,1,NULL,SET_OP_UNION);
+        sunionDiffGenericCommand(c,c->argv+1,1,NULL,SET_OP_UNION,1);
 
         /* Delete the set as it is now empty */
         dbDelete(c->db,c->argv[1]);
@@ -663,7 +667,7 @@ void srandmemberWithCountCommand(client *c) {
      * The number of requested elements is greater than the number of
      * elements inside the set: simply return the whole set. */
     if (count >= size) {
-        sunionDiffGenericCommand(c,c->argv+1,1,NULL,SET_OP_UNION);
+        sunionDiffGenericCommand(c,c->argv+1,1,NULL,SET_OP_UNION,0);
         return;
     }
 
@@ -789,9 +793,19 @@ int qsortCompareSetsByRevCardinality(const void *s1, const void *s2) {
     return 0;
 }
 
-void sinterGenericCommand(client *c, robj **setkeys,
-                          unsigned long setnum, robj *dstkey) {
-    robj **sets = zmalloc(sizeof(robj*)*setnum);
+struct sinterThreadOptions {
+    unsigned long setnum;
+    int threaded;
+    robj **sets;
+    robj *dstkey; /* Only used if threaded is false. */
+};
+
+void sinterThreadedHalf(client *c, void *options) {
+    struct sinterThreadOptions *opt = options;
+    robj **sets = opt->sets;
+    unsigned long setnum = opt->setnum;
+    robj *dstkey = opt->dstkey;
+
     setTypeIterator *si;
     robj *dstset = NULL;
     sds elesds;
@@ -800,29 +814,6 @@ void sinterGenericCommand(client *c, robj **setkeys,
     unsigned long j, cardinality = 0;
     int encoding;
 
-    for (j = 0; j < setnum; j++) {
-        robj *setobj = dstkey ?
-            lookupKeyWrite(c->db,setkeys[j]) :
-            lookupKeyRead(c->db,setkeys[j]);
-        if (!setobj) {
-            zfree(sets);
-            if (dstkey) {
-                if (dbDelete(c->db,dstkey)) {
-                    signalModifiedKey(c,c->db,dstkey);
-                    server.dirty++;
-                }
-                addReply(c,shared.czero);
-            } else {
-                addReply(c,shared.emptyset[c->resp]);
-            }
-            return;
-        }
-        if (checkType(c,setobj,OBJ_SET)) {
-            zfree(sets);
-            return;
-        }
-        sets[j] = setobj;
-    }
     /* Sort sets from the smallest to largest, this will improve our
      * algorithm's performance */
     qsort(sets,setnum,sizeof(robj*),qsortCompareSetsByCardinality);
@@ -894,7 +885,11 @@ void sinterGenericCommand(client *c, robj **setkeys,
 
     if (dstkey) {
         /* Store the resulting set into the target, if the intersection
-         * is not an empty set. */
+         * is not an empty set.
+         *
+         * Note that this part is never called in the context of a thread
+         * and contains many API calls that are forbidden in the threaded
+         * half: we get here only in the case of synchronous execution. */
         int deleted = dbDelete(c->db,dstkey);
         if (setTypeSize(dstset) > 0) {
             dbAdd(c->db,dstkey,dstset);
@@ -914,6 +909,67 @@ void sinterGenericCommand(client *c, robj **setkeys,
         setDeferredSetLen(c,replylen,cardinality);
     }
     zfree(sets);
+    zfree(options);
+}
+
+void sinterGenericCommand(client *c, robj **setkeys,
+                          unsigned long setnum, robj *dstkey) {
+    robj **sets = zmalloc(sizeof(robj*)*setnum);
+    int threaded_call_allowed = (dstkey == NULL);
+    int threaded = 0;
+    unsigned long j;
+
+    for (j = 0; j < setnum; j++) {
+        robj *setobj = dstkey ?
+            lookupKeyWrite(c->db,setkeys[j]) :
+            lookupKeyRead(c->db,setkeys[j]);
+        if (!setobj) {
+            zfree(sets);
+            if (dstkey) {
+                if (dbDelete(c->db,dstkey)) {
+                    signalModifiedKey(c,c->db,dstkey);
+                    server.dirty++;
+                }
+                addReply(c,shared.czero);
+            } else {
+                addReply(c,shared.emptyset[c->resp]);
+            }
+            return;
+        }
+        if (checkType(c,setobj,OBJ_SET)) {
+            zfree(sets);
+            return;
+        }
+        sets[j] = setobj;
+        if (threaded_call_allowed &&
+            setTypeSize(setobj) >= THREADED_SET_SIZE_THRESHOLD)
+        {
+            threaded = 1;
+        }
+    }
+
+    /* Try to lock the keys only if this is a threaded execution. If we
+     * fail we just revert to non threaded execution. */
+    if (threaded) {
+        for (j = 0; j < setnum; j++) {
+            if (lockKey(c,setkeys[j],LOCKEDKEY_READ,NULL) == C_ERR) {
+                unlockAllKeys(c);
+                threaded = 0;
+            }
+        }
+    }
+
+    struct sinterThreadOptions *opt = zmalloc(sizeof(*opt));
+    opt->setnum = setnum;
+    opt->sets = sets;
+    opt->dstkey = dstkey;
+    opt->threaded = threaded;
+
+    if (threaded) {
+        executeThreadedCommand(c,sinterThreadedHalf,opt);
+    } else {
+        sinterThreadedHalf(c,opt);
+    }
 }
 
 void sinterCommand(client *c) {
@@ -928,30 +984,28 @@ void sinterstoreCommand(client *c) {
 #define SET_OP_DIFF 1
 #define SET_OP_INTER 2
 
-void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
-                              robj *dstkey, int op) {
-    robj **sets = zmalloc(sizeof(robj*)*setnum);
+struct sunionDiffThreadOptions {
+    int op;
+    int setnum;
+    int threaded;
+    robj **sets;
+    robj *dstkey; /* Only used if threaded is false. */
+};
+
+void sunionDiffThreadedHalf(client *c, void *options) {
+    struct sunionDiffThreadOptions *opt = options;
+    robj **sets = opt->sets;
+    int setnum = opt->setnum;
+    int op = opt->op;
+    int threaded = opt->threaded;
+    robj *dstkey = opt->dstkey;
+
     setTypeIterator *si;
     robj *dstset = NULL;
     sds ele;
     int j, cardinality = 0;
     int diff_algo = 1;
 
-    for (j = 0; j < setnum; j++) {
-        robj *setobj = dstkey ?
-            lookupKeyWrite(c->db,setkeys[j]) :
-            lookupKeyRead(c->db,setkeys[j]);
-        if (!setobj) {
-            sets[j] = NULL;
-            continue;
-        }
-        if (checkType(c,setobj,OBJ_SET)) {
-            zfree(sets);
-            return;
-        }
-        sets[j] = setobj;
-    }
-
     /* Select what DIFF algorithm to use.
      *
      * Algorithm 1 is O(N*M) where N is the size of the element first set
@@ -1064,11 +1118,17 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
             sdsfree(ele);
         }
         setTypeReleaseIterator(si);
-        server.lazyfree_lazy_server_del ? freeObjAsync(dstset) :
-                                          decrRefCount(dstset);
+        if (server.lazyfree_lazy_server_del && !threaded)
+            freeObjAsync(dstset);
+        else
+            decrRefCount(dstset);
     } else {
         /* If we have a target key where to store the resulting set
-         * create this key with the result set inside */
+         * create this key with the result set inside.
+         *
+         * Note that this part is never called in the context of a thread
+         * and contains many API calls that are forbidden in the threaded
+         * half: we get here only in the case of synchronous execution. */
         int deleted = dbDelete(c->db,dstkey);
         if (setTypeSize(dstset) > 0) {
             dbAdd(c->db,dstkey,dstset);
@@ -1087,22 +1147,74 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
         server.dirty++;
     }
     zfree(sets);
+    zfree(opt);
+}
+
+void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
+                              robj *dstkey, int op, int sync) {
+    robj **sets = zmalloc(sizeof(robj*)*setnum);
+    int threaded_call_allowed = (dstkey == NULL && !sync);
+    int threaded = 0;
+
+    for (int j = 0; j < setnum; j++) {
+        robj *setobj = dstkey ?
+            lookupKeyWrite(c->db,setkeys[j]) :
+            lookupKeyRead(c->db,setkeys[j]);
+        if (!setobj) {
+            sets[j] = NULL;
+            continue;
+        }
+        if (checkType(c,setobj,OBJ_SET)) {
+            zfree(sets);
+            return;
+        }
+        sets[j] = setobj;
+        if (threaded_call_allowed &&
+            setTypeSize(setobj) >= THREADED_SET_SIZE_THRESHOLD)
+        {
+            threaded = 1;
+        }
+    }
+
+    /* Try to lock the keys only if this is a threaded execution. If we
+     * fail we just revert to non threaded execution. */
+    if (threaded) {
+        for (int j = 0; j < setnum; j++) {
+            if (lockKey(c,setkeys[j],LOCKEDKEY_READ,NULL) == C_ERR) {
+                unlockAllKeys(c);
+                threaded = 0;
+            }
+        }
+    }
+
+    struct sunionDiffThreadOptions *opt = zmalloc(sizeof(*opt));
+    opt->op = op;
+    opt->setnum = setnum;
+    opt->sets = sets;
+    opt->dstkey = dstkey;
+    opt->threaded = threaded;
+
+    if (threaded) {
+        executeThreadedCommand(c,sunionDiffThreadedHalf,opt);
+    } else {
+        sunionDiffThreadedHalf(c,opt);
+    }
 }
 
 void sunionCommand(client *c) {
-    sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,SET_OP_UNION);
+    sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,SET_OP_UNION,0);
 }
 
 void sunionstoreCommand(client *c) {
-    sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],SET_OP_UNION);
+    sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],SET_OP_UNION,1);
 }
 
 void sdiffCommand(client *c) {
-    sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,SET_OP_DIFF);
+    sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,SET_OP_DIFF,0);
 }
 
 void sdiffstoreCommand(client *c) {
-    sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],SET_OP_DIFF);
+    sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],SET_OP_DIFF,1);
 }
 
 void sscanCommand(client *c) {

src/t_string.c

@@ -494,71 +494,25 @@ void stralgoCommand(client *c) {
     }
 }
 
-/* STRALGO <algo> [IDX] [MINMATCHLEN <len>] [WITHMATCHLEN]
- *     STRINGS <string> <string> | KEYS <keya> <keyb>
- */
-void stralgoLCS(client *c) {
+/* Options passed by the main-thread half of STRALGO LCS command, to the
+ * threaded half. */
+struct LCSThreadOptions {
+    robj *obja, *objb;
+    long long minmatchlen;
+    int getlen;
+    int getidx;
+    int withmatchlen;
+};
+
+/* This implements the threaded part of STRALGO LCS. It's the callback
+ * we provide to the background execution engine. */
+void stralgoLCSThreadedHalf(client *c, void *options) {
     uint32_t i, j;
-    long long minmatchlen = 0;
-    sds a = NULL, b = NULL;
-    int getlen = 0, getidx = 0, withmatchlen = 0;
-    robj *obja = NULL, *objb = NULL;
-
-    for (j = 2; j < (uint32_t)c->argc; j++) {
-        char *opt = c->argv[j]->ptr;
-        int moreargs = (c->argc-1) - j;
-
-        if (!strcasecmp(opt,"IDX")) {
-            getidx = 1;
-        } else if (!strcasecmp(opt,"LEN")) {
-            getlen = 1;
-        } else if (!strcasecmp(opt,"WITHMATCHLEN")) {
-            withmatchlen = 1;
-        } else if (!strcasecmp(opt,"MINMATCHLEN") && moreargs) {
-            if (getLongLongFromObjectOrReply(c,c->argv[j+1],&minmatchlen,NULL)
-                != C_OK) return;
-            if (minmatchlen < 0) minmatchlen = 0;
-            j++;
-        } else if (!strcasecmp(opt,"STRINGS") && moreargs > 1) {
-            if (a != NULL) {
-                addReplyError(c,"Either use STRINGS or KEYS");
-                return;
-            }
-            a = c->argv[j+1]->ptr;
-            b = c->argv[j+2]->ptr;
-            j += 2;
-        } else if (!strcasecmp(opt,"KEYS") && moreargs > 1) {
-            if (a != NULL) {
-                addReplyError(c,"Either use STRINGS or KEYS");
-                return;
-            }
-            obja = lookupKeyRead(c->db,c->argv[j+1]);
-            objb = lookupKeyRead(c->db,c->argv[j+2]);
-            obja = obja ? getDecodedObject(obja) : createStringObject("",0);
-            objb = objb ? getDecodedObject(objb) : createStringObject("",0);
-            a = obja->ptr;
-            b = objb->ptr;
-            j += 2;
-        } else {
-            addReply(c,shared.syntaxerr);
-            return;
-        }
-    }
-
-    /* Complain if the user passed ambiguous parameters. */
-    if (a == NULL) {
-        addReplyError(c,"Please specify two strings: "
-                        "STRINGS or KEYS options are mandatory");
-        return;
-    } else if (getlen && getidx) {
-        addReplyError(c,
-            "If you want both the length and indexes, please "
-            "just use IDX.");
-        return;
-    }
 
     /* Compute the LCS using the vanilla dynamic programming technique of
      * building a table of LCS(x,y) substrings. */
+    struct LCSThreadOptions *opt = options;
+    sds a = opt->obja->ptr, b = opt->objb->ptr;
     uint32_t alen = sdslen(a);
     uint32_t blen = sdslen(b);
 
@@ -603,12 +557,12 @@ void stralgoLCS(client *c) {
              brange_end = 0;
 
     /* Do we need to compute the actual LCS string? Allocate it in that case. */
-    int computelcs = getidx || !getlen;
+    int computelcs = opt->getidx || !opt->getlen;
     if (computelcs) result = sdsnewlen(SDS_NOINIT,idx);
 
     /* Start with a deferred array if we have to emit the ranges. */
     uint32_t arraylen = 0;  /* Number of ranges emitted in the array. */
-    if (getidx) {
+    if (opt->getidx) {
         addReplyMapLen(c,2);
         addReplyBulkCString(c,"matches");
         arraylenptr = addReplyDeferredLen(c);
@@ -657,16 +611,16 @@ void stralgoLCS(client *c) {
         /* Emit the current range if needed. */
         uint32_t match_len = arange_end - arange_start + 1;
         if (emit_range) {
-            if (minmatchlen == 0 || match_len >= minmatchlen) {
+            if (opt->minmatchlen == 0 || match_len >= opt->minmatchlen) {
                 if (arraylenptr) {
-                    addReplyArrayLen(c,2+withmatchlen);
+                    addReplyArrayLen(c,2+opt->withmatchlen);
                     addReplyArrayLen(c,2);
                     addReplyLongLong(c,arange_start);
                     addReplyLongLong(c,arange_end);
                     addReplyArrayLen(c,2);
                     addReplyLongLong(c,brange_start);
                     addReplyLongLong(c,brange_end);
-                    if (withmatchlen) addReplyLongLong(c,match_len);
+                    if (opt->withmatchlen) addReplyLongLong(c,match_len);
                     arraylen++;
                 }
             }
@@ -674,14 +628,12 @@ void stralgoLCS(client *c) {
         }
     }
 
-    /* Signal modified key, increment dirty, ... */
-
     /* Reply depending on the given options. */
     if (arraylenptr) {
         addReplyBulkCString(c,"len");
         addReplyLongLong(c,LCS(alen,blen));
         setDeferredArrayLen(c,arraylenptr,arraylen);
-    } else if (getlen) {
+    } else if (opt->getlen) {
         addReplyLongLong(c,LCS(alen,blen));
     } else {
         addReplyBulkSds(c,result);
@@ -689,10 +641,96 @@ void stralgoLCS(client *c) {
     }
 
     /* Cleanup. */
-    if (obja) decrRefCount(obja);
-    if (objb) decrRefCount(objb);
     sdsfree(result);
     zfree(lcs);
+    decrRefCount(opt->obja);
+    decrRefCount(opt->objb);
+    zfree(opt);
     return;
 }
 
+/* STRALGO LCS [IDX] [MINMATCHLEN <len>] [WITHMATCHLEN]
+ *     STRINGS <string> <string> | KEYS <keya> <keyb> */
+void stralgoLCS(client *c) {
+    struct LCSThreadOptions *opt = zmalloc(sizeof(*opt));
+    opt->minmatchlen = 0;
+    opt->getlen = 0;
+    opt->getidx = 0;
+    opt->withmatchlen = 0;
+
+    int data_from_key = 0;
+    robj *obja = NULL, *objb = NULL;
+
+    for (int j = 2; j < c->argc; j++) {
+        char *optname = c->argv[j]->ptr;
+        int moreargs = (c->argc-1) - j;
+
+        if (!strcasecmp(optname,"IDX")) {
+            opt->getidx = 1;
+        } else if (!strcasecmp(optname,"LEN")) {
+            opt->getlen = 1;
+        } else if (!strcasecmp(optname,"WITHMATCHLEN")) {
+            opt->withmatchlen = 1;
+        } else if (!strcasecmp(optname,"MINMATCHLEN") && moreargs) {
+            if (getLongLongFromObjectOrReply(c,c->argv[j+1],
+                &opt->minmatchlen,NULL) != C_OK) return;
+            if (opt->minmatchlen < 0) opt->minmatchlen = 0;
+            j++;
+        } else if (!strcasecmp(optname,"STRINGS") && moreargs > 1) {
+            if (obja != NULL) {
+                addReplyError(c,"Either use STRINGS or KEYS");
+                goto syntaxerr;
+            }
+            obja = c->argv[j+1];
+            objb = c->argv[j+2];
+            /* We increment the objects refcount only to make the cleanup
+             * code path the same as the KEYS option. */
+            incrRefCount(obja);
+            incrRefCount(objb);
+            j += 2;
+        } else if (!strcasecmp(optname,"KEYS") && moreargs > 1) {
+            if (obja != NULL) {
+                addReplyError(c,"Either use STRINGS or KEYS");
+                goto syntaxerr;
+            }
+            obja = lookupKeyRead(c->db,c->argv[j+1]);
+            objb = lookupKeyRead(c->db,c->argv[j+2]);
+            obja = obja ? getDecodedObject(obja) : createStringObject("",0);
+            objb = objb ? getDecodedObject(objb) : createStringObject("",0);
+            j += 2;
+            data_from_key = 1;
+        } else {
+            addReply(c,shared.syntaxerr);
+            goto syntaxerr;
+        }
+    }
+
+    /* Complain if the user passed ambiguous parameters. */
+    if (obja == NULL) {
+        addReplyError(c,"Please specify two strings: "
+                        "STRINGS or KEYS options are mandatory");
+        return;
+    } else if (opt->getlen && opt->getidx) {
+        addReplyError(c,
+            "If you want both the length and indexes, please "
+            "just use IDX.");
+        return;
+    }
+
+    /* Switch to the threaded execution of the command. We always work on
+     * a copy of the data. */
+    robj *a = dupStringObject(obja);
+    robj *b = dupStringObject(objb);
+    decrRefCount(obja);
+    decrRefCount(objb);
+    opt->obja = a;
+    opt->objb = b;
+    executeThreadedCommand(c, stralgoLCSThreadedHalf, opt);
+    return;
+
+syntaxerr:
+    if (obja) decrRefCount(obja);
+    if (objb) decrRefCount(obja);
+    zfree(opt);
+}
+

tests/test_helper.tcl

@@ -66,6 +66,7 @@ set ::all_tests {
     unit/pendingquerybuf
     unit/tls
     unit/tracking
+    unit/corethreads
 }
 # Index to the next test to run in the ::all_tests list.
 set ::next_test 0

tests/unit/corethreads.tcl

@@ -0,0 +1,120 @@
+start_server {tags {"corethreads"}} {
+    test {Write operations are blocked on locked keys} {
+        set rd [redis_deferring_client]
+        set now [clock milliseconds]
+        $rd DEBUG LOCK 1000 read mykey
+        r incr mykey
+        set elapsed [expr {[clock milliseconds]-$now}]
+        assert {$elapsed > 500}
+        $rd read
+        $rd close
+    } 0
+
+    test {Disconnecting a client with an active thread does not crash Redis} {
+        set rd [redis_deferring_client]
+        $rd CLIENT ID
+        set id [$rd read]
+        $rd DEBUG LOCK 5000 read mykey
+        assert {[r client kill id $id] == 1}
+        $rd close
+    }
+
+    test {Disconnecting a client blocked on a key does not crash Redis} {
+        set rd1 [redis_deferring_client]
+        set rd2 [redis_deferring_client]
+        $rd1 CLIENT ID
+        set id [$rd1 read]
+        $rd2 DEBUG LOCK 5000 read mykey
+        $rd1 INCR mykey ; # Now this $rd1 is blocked on mykey
+        assert {[r client kill id $id] == 1}
+        $rd1 close
+        $rd2 close
+    }
+
+    test {Client is resumed only after last key gets unlocked} {
+        set rd1 [redis_deferring_client]
+        set rd2 [redis_deferring_client]
+        set now [clock milliseconds]
+        $rd1 DEBUG LOCK 500 read mykey1
+        $rd2 DEBUG LOCK 2000 read mykey2
+        r mset mykey1 a mykey2 b
+        set elapsed [expr {[clock milliseconds]-$now}]
+        assert {$elapsed > 1500}
+        $rd1 close
+        $rd2 close
+    } 0
+
+    test {Build large data structures for the next tests} {
+        for {set j 0} {$j < 10000} {incr j} {
+            lappend set1 $j
+            lappend set2 [expr {$j*2}]
+        }
+        r sadd set1 {*}$set1
+        r sadd set2 {*}$set2
+        r rpush list1 {*}$set1
+        assert {[r scard set1] == $j}
+        assert {[r scard set2] == $j}
+        assert {[r llen list1] == $j}
+    }
+
+    test {Lua can execute threaded commands synchronously} {
+        set items [r EVAL {return redis.call('smembers','set1')} 0]
+        llength $items
+    } 10000
+
+    test {Lua report an error if we try to access a locked key} {
+        set rd [redis_deferring_client]
+        $rd DEBUG LOCK 5000 read set1
+        catch {
+            r EVAL {return redis.call('sadd','set1','foo')} 0
+        } err
+        $rd close
+        set err
+    } {*LOCKED*}
+
+    test {MULTI can execute threaded commands synchronously} {
+        r MULTI
+        r SMEMBERS set1
+        set items [r EXEC]
+        set items [lindex $items 0]
+        llength $items
+    } 10000
+
+    test {MULTI can block on locked keys} {
+        set rd [redis_deferring_client]
+        set now [clock milliseconds]
+        $rd DEBUG LOCK 1000 read mykey3
+        r multi
+        r incr mykey3
+        r exec
+        set elapsed [expr {[clock milliseconds]-$now}]
+        assert {$elapsed > 500}
+        $rd read
+        $rd close
+    }
+
+    test {Threaded SUNION works} {
+        set res [r SUNION set1 set2]
+        llength $res
+    } {15000}
+
+    test {Threaded SDIFF works} {
+        set res [r SDIFF set1 set2]
+        llength $res
+    } {5000}
+
+    test {Threaded SINTER works} {
+        set res [r SINTER set1 set2]
+        llength $res
+    } {5000}
+
+    test {Threaded SMEMBERS works} {
+        set res [r SMEMBERS set1]
+        llength $res
+    } {10000}
+
+    test {Threaded LRANGE works} {
+        set res [r LRANGE list1 0 -1]
+        llength $res
+    } {10000}
+}