Unverified Commit 77d3c6bf authored by perryitay's avatar perryitay Committed by GitHub
Browse files

fix: lookupKey on SETNX and SETXX only once (#9640)

When using SETNX and SETXX we could end up doing key lookup twice.
This presents a small inefficiency price.
Also once we have statistics of write hit and miss they'll be wrong (recording the same key hit twice) 
parent d25dc089
...@@ -777,7 +777,7 @@ void bitopCommand(client *c) { ...@@ -777,7 +777,7 @@ void bitopCommand(client *c) {
/* Store the computed value into the target key */ /* Store the computed value into the target key */
if (maxlen) { if (maxlen) {
o = createObject(OBJ_STRING,res); o = createObject(OBJ_STRING,res);
setKey(c,c->db,targetkey,o); setKey(c,c->db,targetkey,o,0);
notifyKeyspaceEvent(NOTIFY_STRING,"set",targetkey,c->db->id); notifyKeyspaceEvent(NOTIFY_STRING,"set",targetkey,c->db->id);
decrRefCount(o); decrRefCount(o);
server.dirty++; server.dirty++;
......
...@@ -243,25 +243,29 @@ void dbOverwrite(redisDb *db, robj *key, robj *val) { ...@@ -243,25 +243,29 @@ void dbOverwrite(redisDb *db, robj *key, robj *val) {
* 1) The ref count of the value object is incremented. * 1) The ref count of the value object is incremented.
* 2) clients WATCHing for the destination key notified. * 2) clients WATCHing for the destination key notified.
* 3) The expire time of the key is reset (the key is made persistent), * 3) The expire time of the key is reset (the key is made persistent),
* unless 'keepttl' is true. * unless 'SETKEY_KEEPTTL' is enabled in flags.
* 4) The key lookup can take place outside this interface outcome will be
* delivered with 'SETKEY_ALREADY_EXIST' or 'SETKEY_DOESNT_EXIST'
* *
* All the new keys in the database should be created via this interface. * All the new keys in the database should be created via this interface.
* The client 'c' argument may be set to NULL if the operation is performed * The client 'c' argument may be set to NULL if the operation is performed
* in a context where there is no clear client performing the operation. */ * in a context where there is no clear client performing the operation. */
void genericSetKey(client *c, redisDb *db, robj *key, robj *val, int keepttl, int signal) { void setKey(client *c, redisDb *db, robj *key, robj *val, int flags) {
if (lookupKeyWrite(db,key) == NULL) { int keyfound = 0;
if (flags & SETKEY_ALREADY_EXIST)
keyfound = 1;
else if (!(flags & SETKEY_DOESNT_EXIST))
keyfound = (lookupKeyWrite(db,key) != NULL);
if (!keyfound) {
dbAdd(db,key,val); dbAdd(db,key,val);
} else { } else {
dbOverwrite(db,key,val); dbOverwrite(db,key,val);
} }
incrRefCount(val); incrRefCount(val);
if (!keepttl) removeExpire(db,key); if (!(flags & SETKEY_KEEPTTL)) removeExpire(db,key);
if (signal) signalModifiedKey(c,db,key); if (!(flags & SETKEY_NO_SIGNAL)) signalModifiedKey(c,db,key);
}
/* Common case for genericSetKey() where the TTL is not retained. */
void setKey(client *c, redisDb *db, robj *key, robj *val) {
genericSetKey(c,db,key,val,0,1);
} }
/* Return a random key, in form of a Redis object. /* Return a random key, in form of a Redis object.
......
...@@ -820,7 +820,7 @@ void georadiusGeneric(client *c, int srcKeyIndex, int flags) { ...@@ -820,7 +820,7 @@ void georadiusGeneric(client *c, int srcKeyIndex, int flags) {
if (returned_items) { if (returned_items) {
zsetConvertToListpackIfNeeded(zobj,maxelelen,totelelen); zsetConvertToListpackIfNeeded(zobj,maxelelen,totelelen);
setKey(c,c->db,storekey,zobj); setKey(c,c->db,storekey,zobj,0);
decrRefCount(zobj); decrRefCount(zobj);
notifyKeyspaceEvent(NOTIFY_ZSET,flags & GEOSEARCH ? "geosearchstore" : "georadiusstore",storekey, notifyKeyspaceEvent(NOTIFY_ZSET,flags & GEOSEARCH ? "geosearchstore" : "georadiusstore",storekey,
c->db->id); c->db->id);
......
...@@ -3079,7 +3079,7 @@ int RM_GetToDbIdFromOptCtx(RedisModuleKeyOptCtx *ctx) { ...@@ -3079,7 +3079,7 @@ int RM_GetToDbIdFromOptCtx(RedisModuleKeyOptCtx *ctx) {
int RM_StringSet(RedisModuleKey *key, RedisModuleString *str) { int RM_StringSet(RedisModuleKey *key, RedisModuleString *str) {
if (!(key->mode & REDISMODULE_WRITE) || key->iter) return REDISMODULE_ERR; if (!(key->mode & REDISMODULE_WRITE) || key->iter) return REDISMODULE_ERR;
RM_DeleteKey(key); RM_DeleteKey(key);
genericSetKey(key->ctx->client,key->db,key->key,str,0,0); setKey(key->ctx->client,key->db,key->key,str,SETKEY_NO_SIGNAL);
key->value = str; key->value = str;
return REDISMODULE_OK; return REDISMODULE_OK;
} }
...@@ -3159,7 +3159,7 @@ int RM_StringTruncate(RedisModuleKey *key, size_t newlen) { ...@@ -3159,7 +3159,7 @@ int RM_StringTruncate(RedisModuleKey *key, size_t newlen) {
if (key->value == NULL) { if (key->value == NULL) {
/* Empty key: create it with the new size. */ /* Empty key: create it with the new size. */
robj *o = createObject(OBJ_STRING,sdsnewlen(NULL, newlen)); robj *o = createObject(OBJ_STRING,sdsnewlen(NULL, newlen));
genericSetKey(key->ctx->client,key->db,key->key,o,0,0); setKey(key->ctx->client,key->db,key->key,o,SETKEY_NO_SIGNAL);
key->value = o; key->value = o;
decrRefCount(o); decrRefCount(o);
} else { } else {
...@@ -5437,7 +5437,7 @@ int RM_ModuleTypeSetValue(RedisModuleKey *key, moduleType *mt, void *value) { ...@@ -5437,7 +5437,7 @@ int RM_ModuleTypeSetValue(RedisModuleKey *key, moduleType *mt, void *value) {
if (!(key->mode & REDISMODULE_WRITE) || key->iter) return REDISMODULE_ERR; if (!(key->mode & REDISMODULE_WRITE) || key->iter) return REDISMODULE_ERR;
RM_DeleteKey(key); RM_DeleteKey(key);
robj *o = createModuleObject(mt,value); robj *o = createModuleObject(mt,value);
genericSetKey(key->ctx->client,key->db,key->key,o,0,0); setKey(key->ctx->client,key->db,key->key,o,SETKEY_NO_SIGNAL);
decrRefCount(o); decrRefCount(o);
key->value = o; key->value = o;
return REDISMODULE_OK; return REDISMODULE_OK;
......
...@@ -2645,8 +2645,12 @@ int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle, ...@@ -2645,8 +2645,12 @@ int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
void dbAdd(redisDb *db, robj *key, robj *val); void dbAdd(redisDb *db, robj *key, robj *val);
int dbAddRDBLoad(redisDb *db, sds key, robj *val); int dbAddRDBLoad(redisDb *db, sds key, robj *val);
void dbOverwrite(redisDb *db, robj *key, robj *val); void dbOverwrite(redisDb *db, robj *key, robj *val);
void genericSetKey(client *c, redisDb *db, robj *key, robj *val, int keepttl, int signal);
void setKey(client *c, redisDb *db, robj *key, robj *val); #define SETKEY_KEEPTTL 1
#define SETKEY_NO_SIGNAL 2
#define SETKEY_ALREADY_EXIST 4
#define SETKEY_DOESNT_EXIST 8
void setKey(client *c, redisDb *db, robj *key, robj *val, int flags);
robj *dbRandomKey(redisDb *db); robj *dbRandomKey(redisDb *db);
int dbSyncDelete(redisDb *db, robj *key); int dbSyncDelete(redisDb *db, robj *key);
int dbDelete(redisDb *db, robj *key); int dbDelete(redisDb *db, robj *key);
......
...@@ -570,7 +570,7 @@ void sortCommandGeneric(client *c, int readonly) { ...@@ -570,7 +570,7 @@ void sortCommandGeneric(client *c, int readonly) {
} }
} }
if (outputlen) { if (outputlen) {
setKey(c,c->db,storekey,sobj); setKey(c,c->db,storekey,sobj,0);
notifyKeyspaceEvent(NOTIFY_LIST,"sortstore",storekey, notifyKeyspaceEvent(NOTIFY_LIST,"sortstore",storekey,
c->db->id); c->db->id);
server.dirty += outputlen; server.dirty += outputlen;
......
...@@ -987,7 +987,7 @@ void sinterGenericCommand(client *c, robj **setkeys, ...@@ -987,7 +987,7 @@ void sinterGenericCommand(client *c, robj **setkeys,
/* Store the resulting set into the target, if the intersection /* Store the resulting set into the target, if the intersection
* is not an empty set. */ * is not an empty set. */
if (setTypeSize(dstset) > 0) { if (setTypeSize(dstset) > 0) {
setKey(c,c->db,dstkey,dstset); setKey(c,c->db,dstkey,dstset,0);
addReplyLongLong(c,setTypeSize(dstset)); addReplyLongLong(c,setTypeSize(dstset));
notifyKeyspaceEvent(NOTIFY_SET,"sinterstore", notifyKeyspaceEvent(NOTIFY_SET,"sinterstore",
dstkey,c->db->id); dstkey,c->db->id);
...@@ -1191,7 +1191,7 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum, ...@@ -1191,7 +1191,7 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
/* If we have a target key where to store the resulting set /* 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 */
if (setTypeSize(dstset) > 0) { if (setTypeSize(dstset) > 0) {
setKey(c,c->db,dstkey,dstset); setKey(c,c->db,dstkey,dstset,0);
addReplyLongLong(c,setTypeSize(dstset)); addReplyLongLong(c,setTypeSize(dstset));
notifyKeyspaceEvent(NOTIFY_SET, notifyKeyspaceEvent(NOTIFY_SET,
op == SET_OP_UNION ? "sunionstore" : "sdiffstore", op == SET_OP_UNION ? "sunionstore" : "sdiffstore",
......
...@@ -77,6 +77,9 @@ static int getExpireMillisecondsOrReply(client *c, robj *expire, int flags, int ...@@ -77,6 +77,9 @@ static int getExpireMillisecondsOrReply(client *c, robj *expire, int flags, int
void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) { void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {
long long milliseconds = 0; /* initialized to avoid any harmness warning */ long long milliseconds = 0; /* initialized to avoid any harmness warning */
int found = 0;
int setkey_flags = 0;
if (expire && getExpireMillisecondsOrReply(c, expire, flags, unit, &milliseconds) != C_OK) { if (expire && getExpireMillisecondsOrReply(c, expire, flags, unit, &milliseconds) != C_OK) {
return; return;
} }
...@@ -85,8 +88,10 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, ...@@ -85,8 +88,10 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
if (getGenericCommand(c) == C_ERR) return; if (getGenericCommand(c) == C_ERR) return;
} }
if ((flags & OBJ_SET_NX && lookupKeyWrite(c->db,key) != NULL) || found = (lookupKeyWrite(c->db,key) != NULL);
(flags & OBJ_SET_XX && lookupKeyWrite(c->db,key) == NULL))
if ((flags & OBJ_SET_NX && found) ||
(flags & OBJ_SET_XX && !found))
{ {
if (!(flags & OBJ_SET_GET)) { if (!(flags & OBJ_SET_GET)) {
addReply(c, abort_reply ? abort_reply : shared.null[c->resp]); addReply(c, abort_reply ? abort_reply : shared.null[c->resp]);
...@@ -94,7 +99,10 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, ...@@ -94,7 +99,10 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
return; return;
} }
genericSetKey(c,c->db,key, val,flags & OBJ_KEEPTTL,1); setkey_flags |= (flags & OBJ_KEEPTTL) ? SETKEY_KEEPTTL : 0;
setkey_flags |= found ? SETKEY_ALREADY_EXIST : SETKEY_DOESNT_EXIST;
setKey(c,c->db,key,val,setkey_flags);
server.dirty++; server.dirty++;
notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id); notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
...@@ -418,7 +426,7 @@ void getdelCommand(client *c) { ...@@ -418,7 +426,7 @@ void getdelCommand(client *c) {
void getsetCommand(client *c) { void getsetCommand(client *c) {
if (getGenericCommand(c) == C_ERR) return; if (getGenericCommand(c) == C_ERR) return;
c->argv[2] = tryObjectEncoding(c->argv[2]); c->argv[2] = tryObjectEncoding(c->argv[2]);
setKey(c,c->db,c->argv[1],c->argv[2]); setKey(c,c->db,c->argv[1],c->argv[2],0);
notifyKeyspaceEvent(NOTIFY_STRING,"set",c->argv[1],c->db->id); notifyKeyspaceEvent(NOTIFY_STRING,"set",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
...@@ -567,7 +575,7 @@ void msetGenericCommand(client *c, int nx) { ...@@ -567,7 +575,7 @@ void msetGenericCommand(client *c, int nx) {
for (j = 1; j < c->argc; j += 2) { for (j = 1; j < c->argc; j += 2) {
c->argv[j+1] = tryObjectEncoding(c->argv[j+1]); c->argv[j+1] = tryObjectEncoding(c->argv[j+1]);
setKey(c,c->db,c->argv[j],c->argv[j+1]); setKey(c,c->db,c->argv[j],c->argv[j+1],0);
notifyKeyspaceEvent(NOTIFY_STRING,"set",c->argv[j],c->db->id); notifyKeyspaceEvent(NOTIFY_STRING,"set",c->argv[j],c->db->id);
} }
server.dirty += (c->argc-1)/2; server.dirty += (c->argc-1)/2;
......
...@@ -2762,7 +2762,7 @@ void zunionInterDiffGenericCommand(client *c, robj *dstkey, int numkeysIndex, in ...@@ -2762,7 +2762,7 @@ void zunionInterDiffGenericCommand(client *c, robj *dstkey, int numkeysIndex, in
if (dstkey) { if (dstkey) {
if (dstzset->zsl->length) { if (dstzset->zsl->length) {
zsetConvertToListpackIfNeeded(dstobj, maxelelen, totelelen); zsetConvertToListpackIfNeeded(dstobj, maxelelen, totelelen);
setKey(c, c->db, dstkey, dstobj); setKey(c, c->db, dstkey, dstobj, 0);
addReplyLongLong(c, zsetLength(dstobj)); addReplyLongLong(c, zsetLength(dstobj));
notifyKeyspaceEvent(NOTIFY_ZSET, notifyKeyspaceEvent(NOTIFY_ZSET,
(op == SET_OP_UNION) ? "zunionstore" : (op == SET_OP_UNION) ? "zunionstore" :
...@@ -2955,7 +2955,7 @@ static void zrangeResultEmitLongLongForStore(zrange_result_handler *handler, ...@@ -2955,7 +2955,7 @@ static void zrangeResultEmitLongLongForStore(zrange_result_handler *handler,
static void zrangeResultFinalizeStore(zrange_result_handler *handler, size_t result_count) static void zrangeResultFinalizeStore(zrange_result_handler *handler, size_t result_count)
{ {
if (result_count) { if (result_count) {
setKey(handler->client, handler->client->db, handler->dstkey, handler->dstobj); setKey(handler->client, handler->client->db, handler->dstkey, handler->dstobj, 0);
addReplyLongLong(handler->client, result_count); addReplyLongLong(handler->client, result_count);
notifyKeyspaceEvent(NOTIFY_ZSET, "zrangestore", handler->dstkey, handler->client->db->id); notifyKeyspaceEvent(NOTIFY_ZSET, "zrangestore", handler->dstkey, handler->client->db->id);
server.dirty++; server.dirty++;
......
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