Commit 426b8643 authored by antirez's avatar antirez
Browse files

Threaded core commands: block clients on locked keys.

parent dd53453e
...@@ -179,7 +179,7 @@ void replyToBlockedClientTimedOut(client *c) { ...@@ -179,7 +179,7 @@ void replyToBlockedClientTimedOut(client *c) {
if (c->btype == BLOCKED_LIST || if (c->btype == BLOCKED_LIST ||
c->btype == BLOCKED_ZSET || c->btype == BLOCKED_ZSET ||
c->btype == BLOCKED_STREAM || c->btype == BLOCKED_STREAM ||
c->btype == BLOCKED_LOCK) { c->btype == BLOCKED_LOCK) { /* type LOCK never timeouts actually. */
addReplyNullArray(c); addReplyNullArray(c);
} else if (c->btype == BLOCKED_WAIT) { } else if (c->btype == BLOCKED_WAIT) {
addReplyLongLong(c,replicationCountAcksByOffset(c->bpop.reploffset)); addReplyLongLong(c,replicationCountAcksByOffset(c->bpop.reploffset));
......
...@@ -1824,3 +1824,16 @@ int lockKey(client *c, redisDb *db, robj *key, int locktype, robj **optr) { ...@@ -1824,3 +1824,16 @@ int lockKey(client *c, redisDb *db, robj *key, int locktype, robj **optr) {
*optr = lk->obj; *optr = lk->obj;
return C_OK; return C_OK;
} }
/* If the key this client is trying to access is locked, put it into
* its waiting list and return C_OK, otherwise return C_ERR. */
int queueClientIfKeyIsLocked(client *c, robj *key) {
lockedKey *lk = dictFetchValue(c->db->locked_keys,key);
if (lk == NULL) return C_ERR;
lockedKeyClient *lkc = zmalloc(sizeof(*lkc));
lkc->id = c->id;
listAddNodeTail(lk->waiting,lkc);
return C_OK;
}
...@@ -1749,9 +1749,12 @@ void commandProcessed(client *c) { ...@@ -1749,9 +1749,12 @@ void commandProcessed(client *c) {
/* Don't reset the client structure for clients blocked in a /* Don't reset the client structure for clients blocked in a
* module blocking command, so that the reply callback will * module blocking command, so that the reply callback will
* still be able to access the client argv and argc field. * 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) || if (!(c->flags & CLIENT_BLOCKED) ||
c->btype != BLOCKED_MODULE) (c->btype != BLOCKED_MODULE && c->btype != BLOCKED_LOCK))
{ {
resetClient(c); resetClient(c);
} }
......
...@@ -3501,8 +3501,27 @@ int processCommand(client *c) { ...@@ -3501,8 +3501,27 @@ int processCommand(client *c) {
} }
/* We want to block this client if the keys it is going to access /* We want to block this client if the keys it is going to access
* are locked. */ * are locked, and the operation the client is going to perform
if (dictSize(c->db->locked_keys)) { * is not read-only. For now we only block clients in the case of
* write operations since we are just supporting read locks so far.
* When keys will be also locked in write mode, read only operations
* will be queued as well. */
if (c->cmd->flags & CMD_WRITE && dictSize(c->db->locked_keys)) {
int locked = 0;
int numkeys;
int *keyidx = getKeysFromCommand(c->cmd,c->argv,c->argc,&numkeys);
for (int j = 0; j < numkeys; j++) {
if (queueClientIfKeyIsLocked(c,c->argv[keyidx[j]]) == C_OK)
locked++;
}
getKeysFreeResult(keyidx);
/* Lock the client and return if we are waiting for at least one
* key. */
if (locked) {
blockClient(c,BLOCKED_LOCK);
return C_OK;
}
} }
/* Handle the maxmemory directive. /* Handle the maxmemory directive.
......
...@@ -1628,6 +1628,8 @@ void moduleNotifyUserChanged(client *c); ...@@ -1628,6 +1628,8 @@ void moduleNotifyUserChanged(client *c);
typedef void (*coreThreadedCommandCallback)(client *c, void *options); typedef void (*coreThreadedCommandCallback)(client *c, void *options);
void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, void *options); void executeThreadedCommand(client *c, coreThreadedCommandCallback callback, void *options);
unsigned long runningThreadedCommandsCount(void); unsigned long runningThreadedCommandsCount(void);
int lockKey(client *c, redisDb *db, robj *key, int locktype, robj **optr);
int queueClientIfKeyIsLocked(client *c, robj *key);
/* Utils */ /* Utils */
long long ustime(void); long long ustime(void);
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment