X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/defb5f66a090c614aa70aee9e1923ec571e0e911..5244d6e54ec08666f953124739a498d0537a2bf9:/src/t_list.c diff --git a/src/t_list.c b/src/t_list.c index 7c1b848a..21fd1c2f 100644 --- a/src/t_list.c +++ b/src/t_list.c @@ -198,7 +198,7 @@ void listTypeInsert(listTypeEntry *entry, robj *value, int where) { int listTypeEqual(listTypeEntry *entry, robj *o) { listTypeIterator *li = entry->li; if (li->encoding == REDIS_ENCODING_ZIPLIST) { - redisAssert(o->encoding == REDIS_ENCODING_RAW); + redisAssertWithInfo(NULL,o,o->encoding == REDIS_ENCODING_RAW); return ziplistCompare(entry->zi,o->ptr,sdslen(o->ptr)); } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) { return equalStringObjects(o,listNodeValue(entry->ln)); @@ -235,7 +235,7 @@ void listTypeDelete(listTypeEntry *entry) { void listTypeConvert(robj *subject, int enc) { listTypeIterator *li; listTypeEntry entry; - redisAssert(subject->type == REDIS_LIST); + redisAssertWithInfo(NULL,subject,subject->type == REDIS_LIST); if (enc == REDIS_ENCODING_LINKEDLIST) { list *l = listCreate(); @@ -310,7 +310,7 @@ void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) { if (refval != NULL) { /* Note: we expect refval to be string-encoded because it is *not* the * last argument of the multi-bulk LINSERT. */ - redisAssert(refval->encoding == REDIS_ENCODING_RAW); + redisAssertWithInfo(c,refval,refval->encoding == REDIS_ENCODING_RAW); /* 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 @@ -519,7 +519,12 @@ void lrangeCommand(redisClient *c) { p = ziplistNext(o->ptr,p); } } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { - listNode *ln = listIndex(o->ptr,start); + listNode *ln; + + /* If we are nearest to the end of the list, reach the element + * starting from tail and going backward, as it is faster. */ + if (start > llen/2) start -= llen; + ln = listIndex(o->ptr,start); while(rangelen--) { addReplyBulk(c,ln->value); @@ -640,17 +645,37 @@ void lremCommand(redisClient *c) { * as well. This command was originally proposed by Ezra Zygmuntowicz. */ -void rpoplpushHandlePush(redisClient *c, robj *dstkey, robj *dstobj, robj *value) { - if (!handleClientsWaitingListPush(c,dstkey,value)) { +void rpoplpushHandlePush(redisClient *origclient, redisClient *c, robj *dstkey, robj *dstobj, robj *value) { + robj *aux; + + if (!handleClientsWaitingListPush(origclient,dstkey,value)) { /* Create the list if the key does not exist */ if (!dstobj) { dstobj = createZiplistObject(); dbAdd(c->db,dstkey,dstobj); } else { signalModifiedKey(c->db,dstkey); - server.dirty++; } listTypePush(dstobj,value,REDIS_HEAD); + /* If we are pushing as a result of LPUSH against a key + * watched by BRPOPLPUSH, we need to rewrite the command vector + * as an LPUSH. + * + * If this is called directly by RPOPLPUSH (either directly + * or via a BRPOPLPUSH where the popped list exists) + * we should replicate the RPOPLPUSH command itself. */ + if (c != origclient) { + aux = createStringObject("LPUSH",5); + rewriteClientCommandVector(origclient,3,aux,dstkey,value); + decrRefCount(aux); + } else { + /* Make sure to always use RPOPLPUSH in the replication / AOF, + * even if the original command was BRPOPLPUSH. */ + aux = createStringObject("RPOPLPUSH",9); + rewriteClientCommandVector(origclient,3,aux,c->argv[1],c->argv[2]); + decrRefCount(aux); + } + server.dirty++; } /* Always send the pushed value to the client. */ @@ -666,16 +691,22 @@ void rpoplpushCommand(redisClient *c) { addReply(c,shared.nullbulk); } else { robj *dobj = lookupKeyWrite(c->db,c->argv[2]); + robj *touchedkey = c->argv[1]; + if (dobj && checkType(c,dobj,REDIS_LIST)) return; value = listTypePop(sobj,REDIS_TAIL); - rpoplpushHandlePush(c,c->argv[2],dobj,value); + /* We saved touched key, and protect it, since rpoplpushHandlePush + * may change the client command argument vector. */ + incrRefCount(touchedkey); + rpoplpushHandlePush(c,c,c->argv[2],dobj,value); /* listTypePop returns an object with its refcount incremented */ decrRefCount(value); /* Delete the source list when it is empty */ - if (listTypeLength(sobj) == 0) dbDelete(c->db,c->argv[1]); - signalModifiedKey(c->db,c->argv[1]); + if (listTypeLength(sobj) == 0) dbDelete(c->db,touchedkey); + signalModifiedKey(c->db,touchedkey); + decrRefCount(touchedkey); server.dirty++; } } @@ -743,9 +774,9 @@ void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeout, robj l = listCreate(); retval = dictAdd(c->db->blocking_keys,keys[j],l); incrRefCount(keys[j]); - redisAssert(retval == DICT_OK); + redisAssertWithInfo(c,keys[j],retval == DICT_OK); } else { - l = dictGetEntryVal(de); + l = dictGetVal(de); } listAddNodeTail(l,c); } @@ -760,13 +791,13 @@ void unblockClientWaitingData(redisClient *c) { list *l; int j; - redisAssert(c->bpop.keys != NULL); + redisAssertWithInfo(c,NULL,c->bpop.keys != NULL); /* The client may wait for multiple keys, so unblock it for every key. */ for (j = 0; j < c->bpop.count; j++) { /* Remove this client from the list of clients waiting for this key. */ de = dictFind(c->db->blocking_keys,c->bpop.keys[j]); - redisAssert(de != NULL); - l = dictGetEntryVal(de); + redisAssertWithInfo(c,c->bpop.keys[j],de != NULL); + l = dictGetVal(de); listDelNode(l,listSearchKey(l,c)); /* If the list is empty we need to remove it to avoid wasting memory */ if (listLength(l) == 0) @@ -777,6 +808,7 @@ void unblockClientWaitingData(redisClient *c) { /* Cleanup the client structure */ zfree(c->bpop.keys); c->bpop.keys = NULL; + if (c->bpop.target) decrRefCount(c->bpop.target); c->bpop.target = NULL; c->flags &= ~REDIS_BLOCKED; c->flags |= REDIS_UNBLOCKED; @@ -804,7 +836,7 @@ int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) { de = dictFind(c->db->blocking_keys,key); if (de == NULL) return 0; - clients = dictGetEntryVal(de); + clients = dictGetVal(de); numclients = listLength(clients); /* Try to handle the push as long as there are clients waiting for a push. @@ -816,10 +848,14 @@ int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) { * this happens, it simply tries the next client waiting for a push. */ while (numclients--) { ln = listFirst(clients); - redisAssert(ln != NULL); + redisAssertWithInfo(c,key,ln != NULL); receiver = ln->value; dstkey = receiver->bpop.target; + /* Protect receiver->bpop.target, that will be freed by + * the next unblockClientWaitingData() call. */ + if (dstkey) incrRefCount(dstkey); + /* This should remove the first element of the "clients" list. */ unblockClientWaitingData(receiver); @@ -828,17 +864,16 @@ int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) { addReplyMultiBulkLen(receiver,2); addReplyBulk(receiver,key); addReplyBulk(receiver,ele); - return 1; + return 1; /* Serve just the first client as in B[RL]POP semantics */ } else { /* BRPOPLPUSH, note that receiver->db is always equal to c->db. */ dstobj = lookupKeyWrite(receiver->db,dstkey); - if (dstobj && checkType(receiver,dstobj,REDIS_LIST)) { - decrRefCount(dstkey); - } else { - rpoplpushHandlePush(receiver,dstkey,dstobj,ele); + if (!(dstobj && checkType(receiver,dstobj,REDIS_LIST))) { + rpoplpushHandlePush(c,receiver,dstkey,dstobj,ele); decrRefCount(dstkey); return 1; } + decrRefCount(dstkey); } } @@ -882,6 +917,7 @@ void blockingPopGenericCommand(redisClient *c, int where) { if (listTypeLength(o) != 0) { /* If the list contains elements fall back to the usual * non-blocking POP operation */ + struct redisCommand *orig_cmd; robj *argv[2], **orig_argv; int orig_argc; @@ -889,6 +925,7 @@ void blockingPopGenericCommand(redisClient *c, int where) { * popGenericCommand() as the command takes a single key. */ orig_argv = c->argv; orig_argc = c->argc; + orig_cmd = c->cmd; argv[1] = c->argv[j]; c->argv = argv; c->argc = 2; @@ -906,6 +943,7 @@ void blockingPopGenericCommand(redisClient *c, int where) { /* Fix the client structure with the original stuff */ c->argv = orig_argv; c->argc = orig_argc; + c->cmd = orig_cmd; return; } @@ -957,7 +995,7 @@ void brpoplpushCommand(redisClient *c) { /* The list exists and has elements, so * the regular rpoplpushCommand is executed. */ - redisAssert(listTypeLength(key) > 0); + redisAssertWithInfo(c,key,listTypeLength(key) > 0); rpoplpushCommand(c); } }