Commit ca3fd367 authored by antirez's avatar antirez
Browse files

TCC: refactoring + detection of scripts locked keys access.

parent 555adf72
...@@ -1837,15 +1837,55 @@ int lockKey(client *c, robj *key, int locktype, robj **optr) { ...@@ -1837,15 +1837,55 @@ int lockKey(client *c, robj *key, int locktype, robj **optr) {
} }
/* If the key this client is trying to access is locked, put it into /* 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. */ * its waiting list and return 1, otherwise return 0.
int queueClientIfKeyIsLocked(client *c, robj *key) { * 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); lockedKey *lk = dictFetchValue(c->db->locked_keys,key);
if (lk == NULL) return C_ERR; if (lk == NULL) return 0;
if (queue) {
lockedKeyClient *lkc = zmalloc(sizeof(*lkc)); lockedKeyClient *lkc = zmalloc(sizeof(*lkc));
lkc->id = c->id; lkc->id = c->id;
listAddNodeTail(lk->waiting,lkc); listAddNodeTail(lk->waiting,lkc);
return C_OK; }
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)
{
return 0;
}
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]],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. /* Unlock the specified key in the context of the specified client.
......
...@@ -622,7 +622,10 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) { ...@@ -622,7 +622,10 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
/* Write commands are forbidden against read-only slaves, or if a /* Write commands are forbidden against read-only slaves, or if a
* command marked as non-deterministic was already called in the context * 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) { if (cmd->flags & CMD_WRITE) {
int deny_write_type = writeCommandsDeniedByDiskError(); int deny_write_type = writeCommandsDeniedByDiskError();
if (server.lua_random_dirty && !server.lua_replicate_commands) { if (server.lua_random_dirty && !server.lua_replicate_commands) {
...@@ -646,6 +649,11 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) { ...@@ -646,6 +649,11 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
sdsfree(aof_write_err); sdsfree(aof_write_err);
} }
goto cleanup; 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;
} }
} }
......
...@@ -3502,27 +3502,11 @@ int processCommand(client *c) { ...@@ -3502,27 +3502,11 @@ 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, and the operation the client is going to perform * are locked, and the operation the client is going to perform
* is not read-only. For now we only block clients in the case of * is not compatible with the key lock type. */
* write operations since we are just supporting read locks so far. if (clientShouldWaitOnLockedKeys(c,1)) {
* 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); blockClient(c,BLOCKED_LOCK);
return C_OK; return C_OK;
} }
}
/* Handle the maxmemory directive. /* Handle the maxmemory directive.
* *
......
...@@ -1629,7 +1629,7 @@ typedef void (*coreThreadedCommandCallback)(client *c, void *options); ...@@ -1629,7 +1629,7 @@ 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, robj *key, int locktype, robj **optr); int lockKey(client *c, robj *key, int locktype, robj **optr);
int queueClientIfKeyIsLocked(client *c, robj *key); int clientShouldWaitOnLockedKeys(client *c, int queue);
void unlockAllKeys(client *c); void unlockAllKeys(client *c);
/* Utils */ /* Utils */
...@@ -2006,6 +2006,7 @@ size_t freeMemoryGetNotCountedMemory(); ...@@ -2006,6 +2006,7 @@ size_t freeMemoryGetNotCountedMemory();
int freeMemoryIfNeeded(void); int freeMemoryIfNeeded(void);
int freeMemoryIfNeededAndSafe(void); int freeMemoryIfNeededAndSafe(void);
int processCommand(client *c); int processCommand(client *c);
int processCommandAndResetClient(client *c);
void setupSignalHandlers(void); void setupSignalHandlers(void);
struct redisCommand *lookupCommand(sds name); struct redisCommand *lookupCommand(sds name);
struct redisCommand *lookupCommandByCString(char *s); struct redisCommand *lookupCommandByCString(char *s);
......
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