Keyspace events added for more commands.

This commit is contained in:
antirez 2013-01-24 16:20:53 +01:00
parent 8766e81079
commit da04e6ed44
8 changed files with 134 additions and 51 deletions

View File

@ -153,6 +153,7 @@ void setbitCommand(redisClient *c) {
byteval |= ((on & 0x1) << bit); byteval |= ((on & 0x1) << bit);
((uint8_t*)o->ptr)[byte] = byteval; ((uint8_t*)o->ptr)[byte] = byteval;
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("setbit",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
addReply(c, bitval ? shared.cone : shared.czero); addReply(c, bitval ? shared.cone : shared.czero);
} }
@ -346,9 +347,11 @@ void bitopCommand(redisClient *c) {
if (maxlen) { if (maxlen) {
o = createObject(REDIS_STRING,res); o = createObject(REDIS_STRING,res);
setKey(c->db,targetkey,o); setKey(c->db,targetkey,o);
notifyKeyspaceEvent("set",targetkey,c->db->id);
decrRefCount(o); decrRefCount(o);
} else if (dbDelete(c->db,targetkey)) { } else if (dbDelete(c->db,targetkey)) {
signalModifiedKey(c->db,targetkey); signalModifiedKey(c->db,targetkey);
notifyKeyspaceEvent("del",targetkey,c->db->id);
} }
server.dirty++; server.dirty++;
addReplyLongLong(c,maxlen); /* Return the output string length in bytes. */ addReplyLongLong(c,maxlen); /* Return the output string length in bytes. */

View File

@ -587,12 +587,14 @@ void expireGenericCommand(redisClient *c, long long basetime, int unit) {
rewriteClientCommandVector(c,2,aux,key); rewriteClientCommandVector(c,2,aux,key);
decrRefCount(aux); decrRefCount(aux);
signalModifiedKey(c->db,key); signalModifiedKey(c->db,key);
notifyKeyspaceEvent("del",key,c->db->id);
addReply(c, shared.cone); addReply(c, shared.cone);
return; return;
} else { } else {
setExpire(c->db,key,when); setExpire(c->db,key,when);
addReply(c,shared.cone); addReply(c,shared.cone);
signalModifiedKey(c->db,key); signalModifiedKey(c->db,key);
notifyKeyspaceEvent("expire",key,c->db->id);
server.dirty++; server.dirty++;
return; return;
} }

View File

@ -504,9 +504,11 @@ void sortCommand(redisClient *c) {
} }
if (outputlen) { if (outputlen) {
setKey(c->db,storekey,sobj); setKey(c->db,storekey,sobj);
notifyKeyspaceEvent("sortstore",storekey,c->db->id);
server.dirty += outputlen; server.dirty += outputlen;
} else if (dbDelete(c->db,storekey)) { } else if (dbDelete(c->db,storekey)) {
signalModifiedKey(c->db,storekey); signalModifiedKey(c->db,storekey);
notifyKeyspaceEvent("del",storekey,c->db->id);
server.dirty++; server.dirty++;
} }
decrRefCount(sobj); decrRefCount(sobj);
@ -525,5 +527,3 @@ void sortCommand(redisClient *c) {
} }
zfree(vector); zfree(vector);
} }

View File

@ -474,6 +474,7 @@ void hsetCommand(redisClient *c) {
update = hashTypeSet(o,c->argv[2],c->argv[3]); update = hashTypeSet(o,c->argv[2],c->argv[3]);
addReply(c, update ? shared.czero : shared.cone); addReply(c, update ? shared.czero : shared.cone);
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("hset",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
} }
@ -489,6 +490,7 @@ void hsetnxCommand(redisClient *c) {
hashTypeSet(o,c->argv[2],c->argv[3]); hashTypeSet(o,c->argv[2],c->argv[3]);
addReply(c, shared.cone); addReply(c, shared.cone);
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("hset",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
} }
} }
@ -510,6 +512,7 @@ void hmsetCommand(redisClient *c) {
} }
addReply(c, shared.ok); addReply(c, shared.ok);
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("hset",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
} }
@ -543,6 +546,7 @@ void hincrbyCommand(redisClient *c) {
decrRefCount(new); decrRefCount(new);
addReplyLongLong(c,value); addReplyLongLong(c,value);
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("hincrby",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
} }
@ -569,6 +573,7 @@ void hincrbyfloatCommand(redisClient *c) {
hashTypeSet(o,c->argv[2],new); hashTypeSet(o,c->argv[2],new);
addReplyBulk(c,new); addReplyBulk(c,new);
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("hincrbyfloat",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
/* Always replicate HINCRBYFLOAT as an HSET command with the final value /* Always replicate HINCRBYFLOAT as an HSET command with the final value
@ -649,7 +654,7 @@ void hmgetCommand(redisClient *c) {
void hdelCommand(redisClient *c) { void hdelCommand(redisClient *c) {
robj *o; robj *o;
int j, deleted = 0; int j, deleted = 0, keyremoved = 0;
if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL || if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,o,REDIS_HASH)) return; checkType(c,o,REDIS_HASH)) return;
@ -659,12 +664,15 @@ void hdelCommand(redisClient *c) {
deleted++; deleted++;
if (hashTypeLength(o) == 0) { if (hashTypeLength(o) == 0) {
dbDelete(c->db,c->argv[1]); dbDelete(c->db,c->argv[1]);
keyremoved = 1;
break; break;
} }
} }
} }
if (deleted) { if (deleted) {
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("hdel",c->argv[1],c->db->id);
if (keyremoved) notifyKeyspaceEvent("del",c->argv[1],c->db->id);
server.dirty += deleted; server.dirty += deleted;
} }
addReplyLongLong(c,deleted); addReplyLongLong(c,deleted);

View File

@ -316,7 +316,12 @@ void pushGenericCommand(redisClient *c, int where) {
pushed++; pushed++;
} }
addReplyLongLong(c, waiting + (lobj ? listTypeLength(lobj) : 0)); addReplyLongLong(c, waiting + (lobj ? listTypeLength(lobj) : 0));
if (pushed) signalModifiedKey(c->db,c->argv[1]); if (pushed) {
char *event = (where == REDIS_HEAD) ? "lpush" : "rpush";
signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent(event,c->argv[1],c->db->id);
}
server.dirty += pushed; server.dirty += pushed;
} }
@ -338,10 +343,6 @@ void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) {
checkType(c,subject,REDIS_LIST)) return; checkType(c,subject,REDIS_LIST)) return;
if (refval != NULL) { if (refval != NULL) {
/* Note: we expect refval to be string-encoded because it is *not* the
* last argument of the multi-bulk LINSERT. */
redisAssertWithInfo(c,refval,refval->encoding == REDIS_ENCODING_RAW);
/* We're not sure if this value can be inserted yet, but we cannot /* We're not sure if this value can be inserted yet, but we cannot
* convert the list inside the iterator. We don't want to loop over * convert the list inside the iterator. We don't want to loop over
* the list twice (once to see if the value can be inserted and once * the list twice (once to see if the value can be inserted and once
@ -366,6 +367,7 @@ void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) {
ziplistLen(subject->ptr) > server.list_max_ziplist_entries) ziplistLen(subject->ptr) > server.list_max_ziplist_entries)
listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST); listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST);
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("linsert",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
} else { } else {
/* Notify client of a failed insert */ /* Notify client of a failed insert */
@ -373,8 +375,11 @@ void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) {
return; return;
} }
} else { } else {
char *event = (where == REDIS_HEAD) ? "lpush" : "rpush";
listTypePush(subject,val,where); listTypePush(subject,val,where);
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent(event,c->argv[1],c->db->id);
server.dirty++; server.dirty++;
} }
@ -469,6 +474,7 @@ void lsetCommand(redisClient *c) {
decrRefCount(value); decrRefCount(value);
addReply(c,shared.ok); addReply(c,shared.ok);
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("lset",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
} }
} else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
@ -481,6 +487,7 @@ void lsetCommand(redisClient *c) {
incrRefCount(value); incrRefCount(value);
addReply(c,shared.ok); addReply(c,shared.ok);
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("lset",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
} }
} else { } else {
@ -496,9 +503,15 @@ void popGenericCommand(redisClient *c, int where) {
if (value == NULL) { if (value == NULL) {
addReply(c,shared.nullbulk); addReply(c,shared.nullbulk);
} else { } else {
char *event = (where == REDIS_HEAD) ? "lpop" : "rpop";
addReplyBulk(c,value); addReplyBulk(c,value);
decrRefCount(value); decrRefCount(value);
if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]); notifyKeyspaceEvent(event,c->argv[1],c->db->id);
if (listTypeLength(o) == 0) {
notifyKeyspaceEvent("del",c->argv[1],c->db->id);
dbDelete(c->db,c->argv[1]);
}
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
server.dirty++; server.dirty++;
} }
@ -618,7 +631,12 @@ void ltrimCommand(redisClient *c) {
} else { } else {
redisPanic("Unknown list encoding"); redisPanic("Unknown list encoding");
} }
if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]);
notifyKeyspaceEvent("ltrim",c->argv[1],c->db->id);
if (listTypeLength(o) == 0) {
dbDelete(c->db,c->argv[1]);
notifyKeyspaceEvent("del",c->argv[1],c->db->id);
}
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
server.dirty++; server.dirty++;
addReply(c,shared.ok); addReply(c,shared.ok);
@ -693,6 +711,7 @@ void rpoplpushHandlePush(redisClient *c, robj *dstkey, robj *dstobj, robj *value
} }
signalModifiedKey(c->db,dstkey); signalModifiedKey(c->db,dstkey);
listTypePush(dstobj,value,REDIS_HEAD); listTypePush(dstobj,value,REDIS_HEAD);
notifyKeyspaceEvent("lpush",dstkey,c->db->id);
/* Always send the pushed value to the client. */ /* Always send the pushed value to the client. */
addReplyBulk(c,value); addReplyBulk(c,value);
} }
@ -722,7 +741,11 @@ void rpoplpushCommand(redisClient *c) {
decrRefCount(value); decrRefCount(value);
/* Delete the source list when it is empty */ /* Delete the source list when it is empty */
if (listTypeLength(sobj) == 0) dbDelete(c->db,touchedkey); notifyKeyspaceEvent("rpop",touchedkey,c->db->id);
if (listTypeLength(sobj) == 0) {
dbDelete(c->db,touchedkey);
notifyKeyspaceEvent("del",touchedkey,c->db->id);
}
signalModifiedKey(c->db,touchedkey); signalModifiedKey(c->db,touchedkey);
decrRefCount(touchedkey); decrRefCount(touchedkey);
server.dirty++; server.dirty++;
@ -1046,6 +1069,7 @@ void blockingPopGenericCommand(redisClient *c, int where) {
} else { } else {
if (listTypeLength(o) != 0) { if (listTypeLength(o) != 0) {
/* Non empty list, this is like a non normal [LR]POP. */ /* Non empty list, this is like a non normal [LR]POP. */
char *event = (where == REDIS_HEAD) ? "lpop" : "rpop";
robj *value = listTypePop(o,where); robj *value = listTypePop(o,where);
redisAssert(value != NULL); redisAssert(value != NULL);
@ -1053,7 +1077,11 @@ void blockingPopGenericCommand(redisClient *c, int where) {
addReplyBulk(c,c->argv[j]); addReplyBulk(c,c->argv[j]);
addReplyBulk(c,value); addReplyBulk(c,value);
decrRefCount(value); decrRefCount(value);
if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[j]); notifyKeyspaceEvent(event,c->argv[j],c->db->id);
if (listTypeLength(o) == 0) {
dbDelete(c->db,c->argv[j]);
notifyKeyspaceEvent("del",c->argv[j],c->db->id);
}
signalModifiedKey(c->db,c->argv[j]); signalModifiedKey(c->db,c->argv[j]);
server.dirty++; server.dirty++;

View File

@ -266,14 +266,17 @@ void saddCommand(redisClient *c) {
c->argv[j] = tryObjectEncoding(c->argv[j]); c->argv[j] = tryObjectEncoding(c->argv[j]);
if (setTypeAdd(set,c->argv[j])) added++; if (setTypeAdd(set,c->argv[j])) added++;
} }
if (added) signalModifiedKey(c->db,c->argv[1]); if (added) {
signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("sadd",c->argv[1],c->db->id);
}
server.dirty += added; server.dirty += added;
addReplyLongLong(c,added); addReplyLongLong(c,added);
} }
void sremCommand(redisClient *c) { void sremCommand(redisClient *c) {
robj *set; robj *set;
int j, deleted = 0; int j, deleted = 0, keyremoved = 0;
if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL || if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,set,REDIS_SET)) return; checkType(c,set,REDIS_SET)) return;
@ -283,12 +286,15 @@ void sremCommand(redisClient *c) {
deleted++; deleted++;
if (setTypeSize(set) == 0) { if (setTypeSize(set) == 0) {
dbDelete(c->db,c->argv[1]); dbDelete(c->db,c->argv[1]);
keyremoved = 1;
break; break;
} }
} }
} }
if (deleted) { if (deleted) {
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("srem",c->argv[1],c->db->id);
if (keyremoved) notifyKeyspaceEvent("del",c->argv[1],c->db->id);
server.dirty += deleted; server.dirty += deleted;
} }
addReplyLongLong(c,deleted); addReplyLongLong(c,deleted);
@ -322,9 +328,13 @@ void smoveCommand(redisClient *c) {
addReply(c,shared.czero); addReply(c,shared.czero);
return; return;
} }
notifyKeyspaceEvent("srem",c->argv[1],c->db->id);
/* Remove the src set from the database when empty */ /* Remove the src set from the database when empty */
if (setTypeSize(srcset) == 0) dbDelete(c->db,c->argv[1]); if (setTypeSize(srcset) == 0) {
dbDelete(c->db,c->argv[1]);
notifyKeyspaceEvent("del",c->argv[1],c->db->id);
}
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
signalModifiedKey(c->db,c->argv[2]); signalModifiedKey(c->db,c->argv[2]);
server.dirty++; server.dirty++;
@ -336,7 +346,10 @@ void smoveCommand(redisClient *c) {
} }
/* An extra key has changed when ele was successfully added to dstset */ /* An extra key has changed when ele was successfully added to dstset */
if (setTypeAdd(dstset,ele)) server.dirty++; if (setTypeAdd(dstset,ele)) {
server.dirty++;
notifyKeyspaceEvent("sadd",c->argv[2],c->db->id);
}
addReply(c,shared.cone); addReply(c,shared.cone);
} }
@ -378,6 +391,7 @@ void spopCommand(redisClient *c) {
incrRefCount(ele); incrRefCount(ele);
setTypeRemove(set,ele); setTypeRemove(set,ele);
} }
notifyKeyspaceEvent("spop",c->argv[1],c->db->id);
/* Replicate/AOF this command as an SREM operation */ /* Replicate/AOF this command as an SREM operation */
aux = createStringObject("SREM",4); aux = createStringObject("SREM",4);
@ -386,7 +400,10 @@ void spopCommand(redisClient *c) {
decrRefCount(aux); decrRefCount(aux);
addReplyBulk(c,ele); addReplyBulk(c,ele);
if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]); if (setTypeSize(set) == 0) {
dbDelete(c->db,c->argv[1]);
notifyKeyspaceEvent("del",c->argv[1],c->db->id);
}
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
server.dirty++; server.dirty++;
} }
@ -691,6 +708,7 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum,
if (setTypeSize(dstset) > 0) { if (setTypeSize(dstset) > 0) {
dbAdd(c->db,dstkey,dstset); dbAdd(c->db,dstkey,dstset);
addReplyLongLong(c,setTypeSize(dstset)); addReplyLongLong(c,setTypeSize(dstset));
notifyKeyspaceEvent("sinterstore",dstkey,c->db->id);
} else { } else {
decrRefCount(dstset); decrRefCount(dstset);
addReply(c,shared.czero); addReply(c,shared.czero);
@ -856,6 +874,9 @@ void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *
if (setTypeSize(dstset) > 0) { if (setTypeSize(dstset) > 0) {
dbAdd(c->db,dstkey,dstset); dbAdd(c->db,dstkey,dstset);
addReplyLongLong(c,setTypeSize(dstset)); addReplyLongLong(c,setTypeSize(dstset));
notifyKeyspaceEvent(
op == REDIS_OP_UNION ? "sunionstore" : "sdiffstore",
dstkey,c->db->id);
} else { } else {
decrRefCount(dstset); decrRefCount(dstset);
addReply(c,shared.czero); addReply(c,shared.czero);

View File

@ -110,6 +110,7 @@ void getsetCommand(redisClient *c) {
if (getGenericCommand(c) == REDIS_ERR) return; if (getGenericCommand(c) == REDIS_ERR) return;
c->argv[2] = tryObjectEncoding(c->argv[2]); c->argv[2] = tryObjectEncoding(c->argv[2]);
setKey(c->db,c->argv[1],c->argv[2]); setKey(c->db,c->argv[1],c->argv[2]);
notifyKeyspaceEvent("set",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
} }
@ -171,6 +172,7 @@ void setrangeCommand(redisClient *c) {
o->ptr = sdsgrowzero(o->ptr,offset+sdslen(value)); o->ptr = sdsgrowzero(o->ptr,offset+sdslen(value));
memcpy((char*)o->ptr+offset,value,sdslen(value)); memcpy((char*)o->ptr+offset,value,sdslen(value));
signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent("setrange",c->argv[1],c->db->id);
server.dirty++; server.dirty++;
} }
addReplyLongLong(c,sdslen(o->ptr)); addReplyLongLong(c,sdslen(o->ptr));
@ -255,7 +257,7 @@ void msetGenericCommand(redisClient *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->db,c->argv[j],c->argv[j+1]); setKey(c->db,c->argv[j],c->argv[j+1]);
notifyKeyspaceEvent("set",c->argv[j+1],c->db->id); notifyKeyspaceEvent("set",c->argv[j],c->db->id);
} }
server.dirty += (c->argc-1)/2; server.dirty += (c->argc-1)/2;
addReply(c, nx ? shared.cone : shared.ok); addReply(c, nx ? shared.cone : shared.ok);

View File

@ -844,9 +844,9 @@ void zaddGenericCommand(redisClient *c, int incr) {
robj *ele; robj *ele;
robj *zobj; robj *zobj;
robj *curobj; robj *curobj;
double score = 0, *scores, curscore = 0.0; double score = 0, *scores = NULL, curscore = 0.0;
int j, elements = (c->argc-2)/2; int j, elements = (c->argc-2)/2;
int added = 0; int added = 0, updated = 0;
if (c->argc % 2) { if (c->argc % 2) {
addReply(c,shared.syntaxerr); addReply(c,shared.syntaxerr);
@ -859,11 +859,7 @@ void zaddGenericCommand(redisClient *c, int incr) {
scores = zmalloc(sizeof(double)*elements); scores = zmalloc(sizeof(double)*elements);
for (j = 0; j < elements; j++) { for (j = 0; j < elements; j++) {
if (getDoubleFromObjectOrReply(c,c->argv[2+j*2],&scores[j],NULL) if (getDoubleFromObjectOrReply(c,c->argv[2+j*2],&scores[j],NULL)
!= REDIS_OK) != REDIS_OK) goto cleanup;
{
zfree(scores);
return;
}
} }
/* Lookup the key and create the sorted set if does not exist. */ /* Lookup the key and create the sorted set if does not exist. */
@ -880,8 +876,7 @@ void zaddGenericCommand(redisClient *c, int incr) {
} else { } else {
if (zobj->type != REDIS_ZSET) { if (zobj->type != REDIS_ZSET) {
addReply(c,shared.wrongtypeerr); addReply(c,shared.wrongtypeerr);
zfree(scores); goto cleanup;
return;
} }
} }
@ -898,10 +893,7 @@ void zaddGenericCommand(redisClient *c, int incr) {
score += curscore; score += curscore;
if (isnan(score)) { if (isnan(score)) {
addReplyError(c,nanerr); addReplyError(c,nanerr);
/* Don't need to check if the sorted set is empty goto cleanup;
* because we know it has at least one element. */
zfree(scores);
return;
} }
} }
@ -909,9 +901,8 @@ void zaddGenericCommand(redisClient *c, int incr) {
if (score != curscore) { if (score != curscore) {
zobj->ptr = zzlDelete(zobj->ptr,eptr); zobj->ptr = zzlDelete(zobj->ptr,eptr);
zobj->ptr = zzlInsert(zobj->ptr,ele,score); zobj->ptr = zzlInsert(zobj->ptr,ele,score);
signalModifiedKey(c->db,key);
server.dirty++; server.dirty++;
updated++;
} }
} else { } else {
/* Optimize: check if the element is too large or the list /* Optimize: check if the element is too large or the list
@ -921,10 +912,8 @@ void zaddGenericCommand(redisClient *c, int incr) {
zsetConvert(zobj,REDIS_ENCODING_SKIPLIST); zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
if (sdslen(ele->ptr) > server.zset_max_ziplist_value) if (sdslen(ele->ptr) > server.zset_max_ziplist_value)
zsetConvert(zobj,REDIS_ENCODING_SKIPLIST); zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
signalModifiedKey(c->db,key);
server.dirty++; server.dirty++;
if (!incr) added++; added++;
} }
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr; zset *zs = zobj->ptr;
@ -943,8 +932,7 @@ void zaddGenericCommand(redisClient *c, int incr) {
addReplyError(c,nanerr); addReplyError(c,nanerr);
/* Don't need to check if the sorted set is empty /* Don't need to check if the sorted set is empty
* because we know it has at least one element. */ * because we know it has at least one element. */
zfree(scores); goto cleanup;
return;
} }
} }
@ -956,29 +944,32 @@ void zaddGenericCommand(redisClient *c, int incr) {
znode = zslInsert(zs->zsl,score,curobj); znode = zslInsert(zs->zsl,score,curobj);
incrRefCount(curobj); /* Re-inserted in skiplist. */ incrRefCount(curobj); /* Re-inserted in skiplist. */
dictGetVal(de) = &znode->score; /* Update score ptr. */ dictGetVal(de) = &znode->score; /* Update score ptr. */
signalModifiedKey(c->db,key);
server.dirty++; server.dirty++;
updated++;
} }
} else { } else {
znode = zslInsert(zs->zsl,score,ele); znode = zslInsert(zs->zsl,score,ele);
incrRefCount(ele); /* Inserted in skiplist. */ incrRefCount(ele); /* Inserted in skiplist. */
redisAssertWithInfo(c,NULL,dictAdd(zs->dict,ele,&znode->score) == DICT_OK); redisAssertWithInfo(c,NULL,dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
incrRefCount(ele); /* Added to dictionary. */ incrRefCount(ele); /* Added to dictionary. */
signalModifiedKey(c->db,key);
server.dirty++; server.dirty++;
if (!incr) added++; added++;
} }
} else { } else {
redisPanic("Unknown sorted set encoding"); redisPanic("Unknown sorted set encoding");
} }
} }
zfree(scores);
if (incr) /* ZINCRBY */ if (incr) /* ZINCRBY */
addReplyDouble(c,score); addReplyDouble(c,score);
else /* ZADD */ else /* ZADD */
addReplyLongLong(c,added); addReplyLongLong(c,added);
cleanup:
zfree(scores);
if (added || updated) {
signalModifiedKey(c->db,key);
notifyKeyspaceEvent(incr ? "zincr" : "zadd", key, c->db->id);
}
} }
void zaddCommand(redisClient *c) { void zaddCommand(redisClient *c) {
@ -992,7 +983,7 @@ void zincrbyCommand(redisClient *c) {
void zremCommand(redisClient *c) { void zremCommand(redisClient *c) {
robj *key = c->argv[1]; robj *key = c->argv[1];
robj *zobj; robj *zobj;
int deleted = 0, j; int deleted = 0, keyremoved = 0, j;
if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL || if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
checkType(c,zobj,REDIS_ZSET)) return; checkType(c,zobj,REDIS_ZSET)) return;
@ -1038,6 +1029,8 @@ void zremCommand(redisClient *c) {
} }
if (deleted) { if (deleted) {
notifyKeyspaceEvent("zrem",key,c->db->id);
if (keyremoved) notifyKeyspaceEvent("del",key,c->db->id);
signalModifiedKey(c->db,key); signalModifiedKey(c->db,key);
server.dirty += deleted; server.dirty += deleted;
} }
@ -1048,6 +1041,7 @@ void zremrangebyscoreCommand(redisClient *c) {
robj *key = c->argv[1]; robj *key = c->argv[1];
robj *zobj; robj *zobj;
zrangespec range; zrangespec range;
int keyremoved = 0;
unsigned long deleted; unsigned long deleted;
/* Parse the range arguments. */ /* Parse the range arguments. */
@ -1061,17 +1055,27 @@ void zremrangebyscoreCommand(redisClient *c) {
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
zobj->ptr = zzlDeleteRangeByScore(zobj->ptr,range,&deleted); zobj->ptr = zzlDeleteRangeByScore(zobj->ptr,range,&deleted);
if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key); if (zzlLength(zobj->ptr) == 0) {
dbDelete(c->db,key);
keyremoved = 1;
}
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr; zset *zs = zobj->ptr;
deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict); deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict);
if (htNeedsResize(zs->dict)) dictResize(zs->dict); if (htNeedsResize(zs->dict)) dictResize(zs->dict);
if (dictSize(zs->dict) == 0) dbDelete(c->db,key); if (dictSize(zs->dict) == 0) {
dbDelete(c->db,key);
keyremoved = 1;
}
} else { } else {
redisPanic("Unknown sorted set encoding"); redisPanic("Unknown sorted set encoding");
} }
if (deleted) signalModifiedKey(c->db,key); if (deleted) {
signalModifiedKey(c->db,key);
notifyKeyspaceEvent("zrembyscore",key,c->db->id);
if (keyremoved) notifyKeyspaceEvent("del",key,c->db->id);
}
server.dirty += deleted; server.dirty += deleted;
addReplyLongLong(c,deleted); addReplyLongLong(c,deleted);
} }
@ -1083,6 +1087,7 @@ void zremrangebyrankCommand(redisClient *c) {
long end; long end;
int llen; int llen;
unsigned long deleted; unsigned long deleted;
int keyremoved = 0;
if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
(getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;
@ -1107,19 +1112,29 @@ void zremrangebyrankCommand(redisClient *c) {
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
/* Correct for 1-based rank. */ /* Correct for 1-based rank. */
zobj->ptr = zzlDeleteRangeByRank(zobj->ptr,start+1,end+1,&deleted); zobj->ptr = zzlDeleteRangeByRank(zobj->ptr,start+1,end+1,&deleted);
if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key); if (zzlLength(zobj->ptr) == 0) {
dbDelete(c->db,key);
keyremoved = 1;
}
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr; zset *zs = zobj->ptr;
/* Correct for 1-based rank. */ /* Correct for 1-based rank. */
deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict); deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
if (htNeedsResize(zs->dict)) dictResize(zs->dict); if (htNeedsResize(zs->dict)) dictResize(zs->dict);
if (dictSize(zs->dict) == 0) dbDelete(c->db,key); if (dictSize(zs->dict) == 0) {
dbDelete(c->db,key);
keyremoved = 1;
}
} else { } else {
redisPanic("Unknown sorted set encoding"); redisPanic("Unknown sorted set encoding");
} }
if (deleted) signalModifiedKey(c->db,key); if (deleted) {
signalModifiedKey(c->db,key);
notifyKeyspaceEvent("zrembyrank",key,c->db->id);
if (keyremoved) notifyKeyspaceEvent("del",key,c->db->id);
}
server.dirty += deleted; server.dirty += deleted;
addReplyLongLong(c,deleted); addReplyLongLong(c,deleted);
} }
@ -1661,6 +1676,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
if (dbDelete(c->db,dstkey)) { if (dbDelete(c->db,dstkey)) {
signalModifiedKey(c->db,dstkey); signalModifiedKey(c->db,dstkey);
notifyKeyspaceEvent("del",dstkey,c->db->id);
touched = 1; touched = 1;
server.dirty++; server.dirty++;
} }
@ -1673,6 +1689,9 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
dbAdd(c->db,dstkey,dstobj); dbAdd(c->db,dstkey,dstobj);
addReplyLongLong(c,zsetLength(dstobj)); addReplyLongLong(c,zsetLength(dstobj));
if (!touched) signalModifiedKey(c->db,dstkey); if (!touched) signalModifiedKey(c->db,dstkey);
notifyKeyspaceEvent(
(op == REDIS_OP_UNION) ? "zunionstore" : "zinterstore",
dstkey,c->db->id);
server.dirty++; server.dirty++;
} else { } else {
decrRefCount(dstobj); decrRefCount(dstobj);