Redis4.0的大key发现和删除

发表于 | 分类于 |

Redis4.0版本之后引入了memory usage命令和lazyfree机制。不管是对大key的发现，还是解决删除大key造成的阻塞问题都有了很大的提升。

大key发现

memory usage的实现主要在object.c->memoryCommand方法中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
{"memory",memoryCommand,-2,
     "random read-only",
     0,NULL,0,0,0,0,0,0},
     
else if (!strcasecmp(c->argv[1]->ptr,"usage") && c->argc >= 3) {
        dictEntry *de;
        long long samples = OBJ_COMPUTE_SIZE_DEF_SAMPLES;
        for (int j = 3; j < c->argc; j++) {
            if (!strcasecmp(c->argv[j]->ptr,"samples") &&
                j+1 < c->argc)
            {
                if (getLongLongFromObjectOrReply(c,c->argv[j+1],&samples,NULL)
                     == C_ERR) return;
                if (samples < 0) {
                    addReply(c,shared.syntaxerr);
                    return;
                }
                if (samples == 0) samples = LLONG_MAX;;
                j++; /* skip option argument. */
            } else {
                addReply(c,shared.syntaxerr);
                return;
            }
        }
        if ((de = dictFind(c->db->dict,c->argv[2]->ptr)) == NULL) {
            addReplyNull(c);
            return;
        }
        size_t usage = objectComputeSize(dictGetVal(de),samples);
        usage += sdsAllocSize(dictGetKey(de));
        usage += sizeof(dictEntry);
        addReplyLongLong(c,usage);
    }

我们可以看到计算使用内存大小核心逻辑是在objectComputeSize函数中，对不同类型的键值计算方式不一样，这里以hash类型举例。
在使用memory usage命令时可以指定一个抽样元素个数。默认为5，决定了内存计算的准确性和计算成本。
这个值越大，循环次数越多，计算结果越精准，性能损耗也越高

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/*...代码对数据类型进行了分类，此处只取hash类型说明*/
    /*...*/
        /*循环抽样个field，累加获取抽样样本内存值，默认抽样样本为5*/
        while((de = dictNext(di)) != NULL && samples < sample_size) {
            ele = dictGetKey(de);
            ele2 = dictGetVal(de);
            elesize += sdsAllocSize(ele) + sdsAllocSize(ele2);
            elesize += sizeof(struct dictEntry);
            samples++;
        }
        dictReleaseIterator(di);
        /*根据上一步计算的抽样样本内存值除以样本量，再乘以总的filed个数计算总内存值*/
        if (samples) asize += (double)elesize/samples*dictSize(d);
    /*...*/
    }

lazyfree

在Redis4.0版本中，新增了一个删除命令unlink。实现了懒删除方式，减少了在删除大key时引起的阻塞影响。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
{"unlink",unlinkCommand,-2,
     "write fast @keyspace",
     0,NULL,1,-1,1,0,0,0},
     
     

void unlinkCommand(client *c) {
    delGenericCommand(c,1);
}

/* This command implements DEL and LAZYDEL. */
void delGenericCommand(client *c, int lazy) {
    int numdel = 0, j;

    for (j = 1; j < c->argc; j++) {
        expireIfNeeded(c->db,c->argv[j]);
        int deleted  = lazy ? dbAsyncDelete(c->db,c->argv[j]) :
                              dbSyncDelete(c->db,c->argv[j]);
        if (deleted) {
            signalModifiedKey(c->db,c->argv[j]);
            notifyKeyspaceEvent(NOTIFY_GENERIC,
                "del",c->argv[j],c->db->id);
            server.dirty++;
            numdel++;
        }
    }
    addReplyLongLong(c,numdel);
}

我们可以看出，del和unlink命令调用的都是delGenericCommand方法。区别主要在于第二个参数，是否为懒删除标记。
如果是懒删除，调用的是异步删除方法dbAsyncDelete

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
/* Delete a key, value, and associated expiration entry if any, from the DB.
 * If there are enough allocations to free the value object may be put into
 * a lazy free list instead of being freed synchronously. The lazy free list
 * will be reclaimed in a different bio.c thread. */
#define LAZYFREE_THRESHOLD 64
int dbAsyncDelete(redisDb *db, robj *key) {
    /* Deleting an entry from the expires dict will not free the sds of
     * the key, because it is shared with the main dictionary. */
    if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);

    /* If the value is composed of a few allocations, to free in a lazy way
     * is actually just slower... So under a certain limit we just free
     * the object synchronously. */
    dictEntry *de = dictUnlink(db->dict,key->ptr);
    if (de) {
        robj *val = dictGetVal(de);
        /*lazyfreeGetFreeEffort来获取val对象所包含的元素个数*/
        size_t free_effort = lazyfreeGetFreeEffort(val);

        /* If releasing the object is too much work, do it in the background
         * by adding the object to the lazy free list.
         * Note that if the object is shared, to reclaim it now it is not
         * possible. This rarely happens, however sometimes the implementation
         * of parts of the Redis core may call incrRefCount() to protect
         * objects, and then call dbDelete(). In this case we'll fall
         * through and reach the dictFreeUnlinkedEntry() call, that will be
         * equivalent to just calling decrRefCount(). */
        /* 对删除key进行判断，满足阈值条件时进行后台删除 */
        if (free_effort > LAZYFREE_THRESHOLD && val->refcount == 1) {
            atomicIncr(lazyfree_objects,1);
            /*将删除对象放入BIO_LAZY_FREE后台线程任务队列*/
            bioCreateBackgroundJob(BIO_LAZY_FREE,val,NULL,NULL);
            /*将第一步获取到的val值设置为null*/
            dictSetVal(db->dict,de,NULL);
        }
    }

    /* Release the key-val pair, or just the key if we set the val
     * field to NULL in order to lazy free it later. */
    if (de) {
        dictFreeUnlinkedEntry(db->dict,de);
        if (server.cluster_enabled) slotToKeyDel(key);
        return 1;
    } else {
        return 0;
    }
}

从函数实现中可以看出，Redis并不是单纯的将所有懒删除操作都放后台线程中进行。而是会先对需要懒删除的key进行判断，不满足条件的key将会直接进行删除操作。
只有满足条件的key才放入到后台线程任务处理队列中。并且立即将其value设置为NULL，避免造成脏读。

1
2
3
4
5
6
7
8
9
10
11
12
else if (type == BIO_LAZY_FREE) {
           /* What we free changes depending on what arguments are set:
            * arg1 -> free the object at pointer.
            * arg2 & arg3 -> free two dictionaries (a Redis DB).
            * only arg3 -> free the skiplist. */
           if (job->arg1)
               lazyfreeFreeObjectFromBioThread(job->arg1);
           else if (job->arg2 && job->arg3)
               lazyfreeFreeDatabaseFromBioThread(job->arg2,job->arg3);
           else if (job->arg3)
               lazyfreeFreeSlotsMapFromBioThread(job->arg3);
       }