Redis系列（八）：数据结构List双向链表中阻塞版本之BLPOP、BRPO

void blockForKeys(client *c,1)">int btype,robj **keys,1)">int numkeys,mstime_t timeout,robj *target,streamID *ids) {
    dictEntry *de;
    list *l;
     j;

    c->bpop.timeout = timeout;
    c->bpop.target = target;

    if (target != NULL) incrRefCount(target);

    0; j < numkeys; j++) {
         Allocate our bkinfo structure,associated to each key the client
         * is blocked for. 
        bkinfo *bki = zmalloc(sizeof(*bki));
        if (btype == BLOCKED_STREAM)
            bki->stream_id = ids[j];

         If the key already exists in the dictionary ignore it. */
        if (dictAdd(c->bpop.keys,keys[j],bki) != DICT_OK) {
            zfree(bki);
            continue;
        }
        incrRefCount(keys[j]);

         And in the other "side",to map keys -> clients 
        de = dictFind(c->db->blocking_keys,keys[j]);
        if (de == retval;

             For every key we take a list of clients blocked for it 
            l = listCreate();
            retval = dictAdd(c->db-> DICT_OK);
        }  {
            l = dictGetVal(de);
        }
        listAddNodeTail(l,c);
        bki->listnode = listLast(l);
    }
    blockClient(c,btype);
}

void blockClient(client *c,1)"> btype) {
    c->flags |= CLIENT_BLOCKED;
    c->btype = btype;
    server.blocked_clients++;
    server.blocked_clients_by_type[btype]++;
    addClientToTimeoutTable(c);
}

void addClientToTimeoutTable(client *c) {
    if (c->bpop.timeout == 0) ;
    uint64_t timeout = c->bpop.timeout;
    unsigned char buf[CLIENT_ST_KEYLEN];
    encodeTimeoutKey(buf,c);
    if (raxTryInsert(server.clients_timeout_table,buf,1)">sizeof(buf),NULL))
        c->flags |= CLIENT_IN_TO_TABLE;
}

6379> lpush mylist World" Hello6379> lindex mylist 6379>

void lindexCommand(client *c) {
    robj *o = lookupKeyReadOrReply(c,c->argv[null[c->resp]);
    if (o == NULL || checkType(c,o,OBJ_LIST)) ;
    long index;
    robj *value = NULL;

    if ((getLongFromObjectOrReply(c,1)">2],&index,NULL) != C_OK))
        if (o->encoding == OBJ_ENCODING_QUICKLIST) {
        quicklistEntry entry;
        if (quicklistIndex(o->ptr,index,1)">entry)) {
            if (entry.value) {
                value = createStringObject((char*)entry.value,entry.sz);
            }  {
                value = createStringObjectFromLongLong(entry.longval);
            }
            addReplyBulk(c,value);
            decrRefCount(value);
        }  {
            addReplyNull(c);
        }
    }  {
        serverPanic(Unknown list encoding);
    }
}

 Populate 'entry' with the element at the specified zero-based index
 * where 0 is the head,1 is the element next to head
 * and so on. Negative integers are used in order to count
 * from the tail,-1 is the last element,-2 the penultimate
 * and so on. If the index is out of range 0 is returned.
 *
 * Returns 1 if element found
 * Returns 0 if element not found int quicklistIndex(const quicklist *quicklist,1)">const long  idx,quicklistEntry *entry) {
    quicklistNode *n;
    unsigned long accum = ;
    unsigned  index;
    int forward = idx < 0 ? 0 : 1;  < 0 -> reverse,0+ -> forward 

    initEntry(entry);
    entry->quicklist = quicklist;

    if (!forward) {
        index = (-idx) - 1;
        n = quicklist->tail;
    }  {
        index = idx;
        n = quicklist->head;
    }

    if (index >= quicklist->count)
        return while (likely(n)) {
        if ((accum + n->count) > index) {
            break;
        }  {
            D(Skipping over (%p) %u at accum %lld",(void *)n,n->count,accum);
            accum += n->count;
            n = forward ? n->next : n->prev;
        }
    }

    n)
        ;

    D(Found node: %p at accum %llu,idx %llu,sub+ %llu,sub- %lluvoid *)n,accum,index - accum,(-index) - 1 + accum);

    entry->node = n;
     (forward) {
         forward = normal head-to-tail offset. 
        entry->offset = index - accum;
    }  {
         reverse = need negative offset for tail-to-head,so undo
         * the result of the original if (index < 0) above. 
        entry->offset = (-index) -  accum;
    }

    quicklistDecompressNodeForUse(entry->node);
    entry->zi = ziplistIndex(entry->node->zl,entry->offset);
    ziplistGet(entry->zi,&entry->value,&entry->sz,&entry->longval);
     The caller will use our result,so we don't re-compress here.
     * The caller can recompress or delete the node as needed. ;
}