redis4.0.9 SET\GET方法从哪里开始server.c里面有每个redis命令对应的执行方法如struct redisCommand redisCommandTable[] = {{module,moduleCommand,-2,as,0,NULL,0,0,0,0,0},{get,getCommand,2,rF,0,N...

redis4.0.9 SET\GET方法

从哪里开始

server.c里面有每个redis命令对应的执行方法
如

struct redisCommand redisCommandTable[] = {
    {"module",moduleCommand,-2,"as",0,NULL,0,0,0,0,0},
    {"get",getCommand,2,"rF",0,NULL,1,1,1,0,0},
    {"set",setCommand,-3,"wm",0,NULL,1,1,1,0,0},
    {"setnx",setnxCommand,3,"wmF",0,NULL,1,1,1,0,0},
    {"setex",setexCommand,4,"wm",0,NULL,1,1,1,0,0},
    {"psetex",psetexCommand,4,"wm",0,NULL,1,1,1,0,0},
    ...
    }

set命令对应setCommand方法，get命令对应getCommand方法

Set in t_string.c

先来看一下setCommand方法，了解一下set命令的流程

/* SET key value [NX] [XX] [EX <seconds>] [PX <milliseconds>] */
void setCommand(client *c) {
    int j;
    robj *expire = NULL;
    int unit = UNIT_SECONDS;
    int flags = OBJ_SET_NO_FLAGS;

    for (j = 3; j < c->argc; j++) {
        char *a = c->argv[j]->ptr;
        robj *next = (j == c->argc-1) ? NULL : c->argv[j+1];

        if ((a[0] == 'n' || a[0] == 'N') &&
            (a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&
            !(flags & OBJ_SET_XX))
        {
            flags |= OBJ_SET_NX;
        } else if ((a[0] == 'x' || a[0] == 'X') &&
                   (a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&
                   !(flags & OBJ_SET_NX))
        {
            flags |= OBJ_SET_XX;
        } else if ((a[0] == 'e' || a[0] == 'E') &&
                   (a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&
                   !(flags & OBJ_SET_PX) && next)
        {
            flags |= OBJ_SET_EX;
            unit = UNIT_SECONDS;
            expire = next;
            j++;
        } else if ((a[0] == 'p' || a[0] == 'P') &&
                   (a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&
                   !(flags & OBJ_SET_EX) && next)
        {
            flags |= OBJ_SET_PX;
            unit = UNIT_MILLISECONDS;
            expire = next;
            j++;
        } else {
            addReply(c,shared.syntaxerr);
            return;
        }
    }

    c->argv[2] = tryObjectEncoding(c->argv[2]);
    setGenericCommand(c,flags,c->argv[1],c->argv[2],expire,unit,NULL,NULL);
}   

• 可以看到是从第4(数组下标为3)个参数开始解析参数的，目的是判断出当前set操作，是NX|XX|EX|PX，如果没有任何命令，会报语法错误
• encoding第3个参数，这是存储的值所在的参数
• set操作执行

tryObjectEncoding会根据类型将值进行encoding，这里包含了很多优化操作，如字符串类型，实际上保存的是一个20位内的数字，会用long来存，原因是long占用的空间更少，这个方法只encoding OBJ_STRING类型

/* Try to encode a string object in order to save space */
robj *tryObjectEncoding(robj *o) {
    long value;
    sds s = o->ptr;
    size_t len;

    /* Make sure this is a string object, the only type we encode
     * in this function. Other types use encoded memory efficient
     * representations but are handled by the commands implementing
     * the type. */
    serverAssertWithInfo(NULL,o,o->type == OBJ_STRING);

    /* We try some specialized encoding only for objects that are
     * RAW or EMBSTR encoded, in other words objects that are still
     * in represented by an actually array of chars. */
    if (!sdsEncodedObject(o)) return o;

    /* It's not safe to encode shared objects: shared objects can be shared
     * everywhere in the "object space" of Redis and may end in places where
     * they are not handled. We handle them only as values in the keyspace. */
     if (o->refcount > 1) return o;

    /* Check if we can represent this string as a long integer.
     * Note that we are sure that a string larger than 20 chars is not
     * representable as a 32 nor 64 bit integer. */
    len = sdslen(s);
    if (len <= 20 && string2l(s,len,&value)) {
        /* This object is encodable as a long. Try to use a shared object.
         * Note that we avoid using shared integers when maxmemory is used
         * because every object needs to have a private LRU field for the LRU
         * algorithm to work well. */
        if ((server.maxmemory == 0 ||
            !(server.maxmemory_policy & MAXMEMORY_FLAG_NO_SHARED_INTEGERS)) &&
            value >= 0 &&
            value < OBJ_SHARED_INTEGERS)
        {
            decrRefCount(o);
            incrRefCount(shared.integers[value]);
            return shared.integers[value];
        } else {
            if (o->encoding == OBJ_ENCODING_RAW) sdsfree(o->ptr);
            o->encoding = OBJ_ENCODING_INT;
            o->ptr = (void*) value;
            return o;
        }
    }

    /* If the string is small and is still RAW encoded,
     * try the EMBSTR encoding which is more efficient.
     * In this representation the object and the SDS string are allocated
     * in the same chunk of memory to save space and cache misses. */
    if (len <= OBJ_ENCODING_EMBSTR_SIZE_LIMIT) {
        robj *emb;

        if (o->encoding == OBJ_ENCODING_EMBSTR) return o;
        emb = createEmbeddedStringObject(s,sdslen(s));
        decrRefCount(o);
        return emb;
    }

    /* We can't encode the object...
     *
     * Do the last try, and at least optimize the SDS string inside
     * the string object to require little space, in case there
     * is more than 10% of free space at the end of the SDS string.
     *
     * We do that only for relatively large strings as this branch
     * is only entered if the length of the string is greater than
     * OBJ_ENCODING_EMBSTR_SIZE_LIMIT. */
    if (o->encoding == OBJ_ENCODING_RAW &&
        sdsavail(s) > len/10)
    {
        o->ptr = sdsRemoveFreeSpace(o->ptr);
    }

    /* Return the original object. */
    return o;
}  

• 方法只encoding String类型的值，目的是压缩字符串占用的空间

• 如果不是指定可压缩类型，直接返回

  #define sdsEncodedObject(objptr) (objptr->encoding == OBJ_ENCODING_RAW || objptr->encoding == OBJ_ENCODING_EMBSTR)

• 共享的对象不压缩

• redis缓存了一定长度的数字 shared.integers[j] 根据 #define OBJ_SHARED_INTEGERS 10000，可以发现，数字范围0~10000，在redis中是共享的，不会分配新的内存，因此，设置10w个key的值为10000以内，value占用的空间是不会变化的

关于为什么只能encoding OBJ_STRING，原因是Redis协议规范中，约定了客户端发给服务端的命令，是一个bulk string数组组成的？？？？代码里面放置的复杂对象，如何变成string？？？？？？

Sending commands to a Redis Server
Now that you are familiar with the RESP serialization format, writing an implementation of a Redis client library will be easy. We can further specify how the interaction between the client and the server works:
A client sends to the Redis server a RESP Array consisting of just Bulk Strings.
A Redis server replies to clients sending any valid RESP data type as reply.

setKey(c->db,key,val);

setKey方法执行的set操作

/* High level Set operation. This function can be used in order to set
 * a key, whatever it was existing or not, to a new object.
 *
 * 1) The ref count of the value object is incremented.
 * 2) clients WATCHing for the destination key notified.
 * 3) The expire time of the key is reset (the key is made persistent).
 *
 * All the new keys in the database should be craeted via this interface. */
void setKey(redisDb *db, robj *key, robj *val) {
    if (lookupKeyWrite(db,key) == NULL) {
        dbAdd(db,key,val);
    } else {
        dbOverwrite(db,key,val);
    }
    incrRefCount(val);
    removeExpire(db,key);
    signalModifiedKey(db,key);
}

在lookupKeyWrite时，会先检查expireIfNeeded

/* Lookup a key for write operations, and as a side effect, if needed, expires
 * the key if its TTL is reached.
 *
 * Returns the linked value object if the key exists or NULL if the key
 * does not exist in the specified DB. */
robj *lookupKeyWrite(redisDb *db, robj *key) {
    expireIfNeeded(db,key);  //获取值时驱动检查过期时间，如果未过期，则什么都不处理，返回0，如果过期了，先把过期事件扩散到slaves或者aof文件，就删除key
    //如果server端启动lazyfree_lazy_expire，则会异步删除，否则同步删除key，
    return lookupKey(db,key,LOOKUP_NONE);
}

• 从这里可以看出，redis的key如果设置了过期时间，并不是到了时间，立即就会被移除，而是基于访问时先检查来做的，先判断是否过期，过期就删除（ps:也有通过一些算法来主动删除key）

  /* This function is called when we are going to perform some operation
  • in a given key, but such key may be already logically expired even if
  • it still exists in the database. The main way this function is called
  • is via lookupKey*() family of functions.
  • The behavior of the function depends on the replication role of the
  • instance, because slave instances do not expire keys, they wait
  • for DELs from the master for consistency matters. However even
  • slaves will try to have a coherent return value for the function,
  • so that read commands executed in the slave side will be able to
  • behave like if the key is expired even if still present (because the
  • master has yet to propagate the DEL).
  • In masters as a side effect of finding a key which is expired, such
  • key will be evicted from the database. Also this may trigger the
  • propagation of a DEL/UNLINK command in AOF / replication stream.
  • The return value of the function is 0 if the key is still valid,
  • otherwise the function returns 1 if the key is expired. */

  int expireIfNeeded(redisDb db, robj key) {
  mstime_t when = getExpire(db,key);
  mstime_t now;

    if (when < 0) return 0; /* No expire for this key */
  
    /* Don't expire anything while loading. It will be done later. */
    if (server.loading) return 0;
  
    /* If we are in the context of a Lua script, we pretend that time is
     * blocked to when the Lua script started. This way a key can expire
     * only the first time it is accessed and not in the middle of the
     * script execution, making propagation to slaves / AOF consistent.
     * See issue #1525 on Github for more information. */
    now = server.lua_caller ? server.lua_time_start : mstime();
  
    /* If we are running in the context of a slave, return ASAP:
     * the slave key expiration is controlled by the master that will
     * send us synthesized DEL operations for expired keys.
     *
     * Still we try to return the right information to the caller,
     * that is, 0 if we think the key should be still valid, 1 if
     * we think the key is expired at this time. */
    if (server.masterhost != NULL) return now > when;
  
    /* Return when this key has not expired */
    if (now <= when) return 0;
  
    /* Delete the key */
    server.stat_expiredkeys++;
    propagateExpire(db,key,server.lazyfree_lazy_expire);  //把过期事件扩散到slaves或者aof文件
    notifyKeyspaceEvent(NOTIFY_EXPIRED,
        "expired",key,db->id);   //通知key为空事件，NOTIFY_KEYSPACE，NOTIFY_KEYEVENT
    return server.lazyfree_lazy_expire ? dbAsyncDelete(db,key) :
                                         dbSyncDelete(db,key);

  }

• 先notify后delete会不会不安全，如果redis是单线程的，那自然就安全了

• 疑点：redis如果完全是多路复用 + 单线程，那是不是在multi命令时，不需要watch某个key了？？

  /* The API provided to the rest of the Redis core is a simple function:
  • notifyKeyspaceEvent(char event, robj key, int dbid);
  • 'event' is a C string representing the event name.
  • 'key' is a Redis object representing the key name.

  • 'dbid' is the database ID where the key lives. /
    void notifyKeyspaceEvent(int type, char event, robj key, int dbid) {
    sds chan;
    robj chanobj, *eventobj;
    int len = -1;
    char buf[24];

    /* If any modules are interested in events, notify the module system now.
    • This bypasses the notifications configuration, but the module engine
    • will only call event subscribers if the event type matches the types
    • they are interested in. */
      moduleNotifyKeyspaceEvent(type, event, key, dbid);

    /* If notifications for this class of events are off, return ASAP. */
    if (!(server.notify_keyspace_events & type)) return;

    eventobj = createStringObject(event,strlen(event));

    /* keyspace@:

    /* keyevent@:

删除key，dict结构

dbSyncDelete方法最终会执行删除key操作dictDelete
可见dic的数据结构类型java的hashMap,为数组+链表结构(ps：这只是其中一种结构，并非全部)

/* Search and remove an element,由此可见dic的数据结构类型java的hashMap,为数组+链表结构 */
static int dictDelete(dict *ht, const void *key) {
    unsigned int h;
    dictEntry *de, *prevde;

    if (ht->size == 0)
        return DICT_ERR;
    h = dictHashKey(ht, key) & ht->sizemask;
    de = ht->table[h];

    prevde = NULL;
    while(de) {
        if (dictCompareHashKeys(ht,key,de->key)) {
            /* Unlink the element from the list */
            if (prevde)
                prevde->next = de->next;
            else
                ht->table[h] = de->next;

            dictFreeEntryKey(ht,de);
            dictFreeEntryVal(ht,de);
            free(de);
            ht->used--;
            return DICT_OK;
        }
        prevde = de;
        de = de->next;
    }
    return DICT_ERR; /* not found */
}

• 通过hashKey 和 sizemask 与操作，获取数组下标
• 删除entry后，会释放key和值占用的空间

dicthashtable结构
/* This is our hash table structure. Every dictionary has two of this as we
* implement incremental rehashing, for the old to the new table. */
typedef struct dictht {
dictEntry **table;
unsigned long size;
unsigned long sizemask;
unsigned long used;
} dictht;

typedef struct dict {
    dictType *type;   //不同的type，会有不同的处理函数，通过type来找到多态方法
    void *privdata;
    dictht ht[2];
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    unsigned long iterators; /* number of iterators currently running */
} dict;

dictExpand

两步走

• 分配新的hash表

• 执行rehash，将老的hash表的值，移动到新的hash表中

  /* Expand or create the hash table /
  int dictExpand(dict d, unsigned long size)
  {
  dictht n; /* the new hash table */
  unsigned long realsize = _dictNextPower(size); //保证realsize为2的倍数

    /* the size is invalid if it is smaller than the number of
     * elements already inside the hash table */
    if (dictIsRehashing(d) || d->ht[0].used > size)
        return DICT_ERR;
  
    /* Rehashing to the same table size is not useful. */
    if (realsize == d->ht[0].size) return DICT_ERR;
  
    /* Allocate the new hash table and initialize all pointers to NULL */
    n.size = realsize;
    n.sizemask = realsize-1;
    n.table = zcalloc(realsize*sizeof(dictEntry*));
    n.used = 0;
  
    /* Is this the first initialization? If so it's not really a rehashing
     * we just set the first hash table so that it can accept keys. */
    if (d->ht[0].table == NULL) {
        d->ht[0] = n;
        return DICT_OK;
    }
  
    /* Prepare a second hash table for incremental rehashing */
    d->ht[1] = n;    //正在rehash，把新的hash表赋给ht[1]
    d->rehashidx = 0;
    return DICT_OK;

  }

• 先判断新的size是否大于当前的size，如果不是，则返回错误
• 修改dict的size，sizemask为size-1,原因是数组下标是从0开始的
• 重新分配内存
• 如果是第一次初始化，分配内存后就直接返回

• 非初始化时，为下次rehashing准备

rehash 操作

/* Performs N steps of incremental rehashing. Returns 1 if there are still
 * keys to move from the old to the new hash table, otherwise 0 is returned.
 *
 * Note that a rehashing step consists in moving a bucket (that may have more
 * than one key as we use chaining) from the old to the new hash table, however
 * since part of the hash table may be composed of empty spaces, it is not
 * guaranteed that this function will rehash even a single bucket, since it
 * will visit at max N*10 empty buckets in total, otherwise the amount of
 * work it does would be unbound and the function may block for a long time. */
int dictRehash(dict *d, int n) {
    int empty_visits = n*10; /* Max number of empty buckets to visit. */
    if (!dictIsRehashing(d)) return 0;

    while(n-- && d->ht[0].used != 0) {
        dictEntry *de, *nextde;

        /* Note that rehashidx can't overflow as we are sure there are more
         * elements because ht[0].used != 0 */
        assert(d->ht[0].size > (unsigned long)d->rehashidx);
        while(d->ht[0].table[d->rehashidx] == NULL) {
            d->rehashidx++;
            if (--empty_visits == 0) return 1;
        }
        de = d->ht[0].table[d->rehashidx];
        /* Move all the keys in this bucket from the old to the new hash HT */
        while(de) {
            uint64_t h;

            nextde = de->next;
            /* Get the index in the new hash table */
            h = dictHashKey(d, de->key) & d->ht[1].sizemask;
            de->next = d->ht[1].table[h];
            d->ht[1].table[h] = de;
            d->ht[0].used--;
            d->ht[1].used++;
            de = nextde;
        }
        d->ht[0].table[d->rehashidx] = NULL;
        d->rehashidx++;
    }

    /* Check if we already rehashed the whole table... */
    if (d->ht[0].used == 0) {
        zfree(d->ht[0].table);
        d->ht[0] = d->ht[1];
        _dictReset(&d->ht[1]);
        d->rehashidx = -1;
        return 0;
    }

    /* More to rehash... */
    return 1;
}

• 根据方法参数n，可以发现rehash操作是慢慢进行的，而不是一蹴而就的，实际上也是这样的，上层方法会根据每执行n次的耗时，和预估的耗时进行比较，如果这次while循环的总的执行时间超过预估时间，就break

  int dictRehashMilliseconds(dict *d, int ms) {
  long long start = timeInMilliseconds();
  int rehashes = 0;
  while(dictRehash(d,100)) {
  rehashes += 100;
  if (timeInMilliseconds()-start > ms) break;
  }
  return rehashes;
  }
  可以参见博客：https://blog.csdn.net/u012658346/article/details/51316029

• 总的来说，就是把ht[0]中所有值移动到ht[1]中，再把ht[1]赋给ht[0]

lookupKey

讲述了较多dict的数据结构，现在继续set方法相关的代码

/* Low level key lookup API, not actually called directly from commands
 * implementations that should instead rely on lookupKeyRead(),
 * lookupKeyWrite() and lookupKeyReadWithFlags(). */
robj *lookupKey(redisDb *db, robj *key, int flags) {
    dictEntry *de = dictFind(db->dict,key->ptr);
    if (de) {
        robj *val = dictGetVal(de);

        /* Update the access time for the ageing algorithm.
         * Don't do it if we have a saving child, as this will trigger
         * a copy on write madness. */
         //当前没有rdb备份进程、aof进程，且非 不更新key last access time时，更新last accesstime
        if (server.rdb_child_pid == -1 &&
            server.aof_child_pid == -1 &&
            !(flags & LOOKUP_NOTOUCH))
        {
            if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
                updateLFU(val);  //走lfu算法
            } else {
                val->lru = LRU_CLOCK(); //走lru算法
            }
        }
        return val;
    } else {
        return NULL;
    }
} 

关键点

• 找key的值，如果没找到就返回NULL
• LFU（Least Frequently Used）：缓存淘汰算法之最少使用原则

• LRU：淘汰最少被访问的key

  参考https://blog.csdn.net/qq_35440678/article/details/53453107

LFU算法

淘汰最少被访问的key算法成为：LFU（Least Frequently Used），将来要被淘汰腾出新空间给新key。
理论上LFU的思想相当简单，只需要给每个key加一个访问计数器。每次访问就自增1，所以也就很容易知道哪些key被访问更频繁。
当然，LFU也会带起其他问题，不单单是针对redis，对于LFU实现：
1、不能使用“移除顶部元素”的方式，keys必须要根据访问计数器进行排序。每访问一次就得遍历所有key找出访问次数最少的key。
2、LFU不能仅仅是只增加每一此访问的计数器。正如我们所讲的，访问模式改变随时变化，因此一个有高访问次数的key，后面很可能没有人继续访问它，因此我们的算法必须要适应超时的情况。

在redis中，第一个问题很好解决：我们可以在LRU的方式一样：随机在缓存池中选举，淘汰其中某项。第二个问题redis还是存在，因此一般对于LFU的思想必须使用一些方式进行减少，或者定期把访问计数器减半。

更新LFU计算器结构

/* Update LFU when an object is accessed.
 * Firstly, decrement the counter if the decrement time is reached.
 * Then logarithmically increment the counter, and update the access time. */
void updateLFU(robj *val) {
    unsigned long counter = LFUDecrAndReturn(val);
    counter = LFULogIncr(counter);
    val->lru = (LFUGetTimeInMinutes()<<8) | counter;
}

redis中，为每个对象额外新增24bit，来存放 上次缩减时间和计数器
16bit : last decr time //时间戳是精确到分钟
8bit:计数器，8bit最大255，如果访问100w此呢，根据随机因子，基于概率来进行计数器+1，服务长时间运行时，概率是趋同的
实际的counter：
你可以配置计数器增长的速率，如果使用默认配置，会发生：

• 100次访问后，计数器=10；
• 1000次访问是是18；
• 10万次访问是142；

• 100万次访问后达到255，不再继续增长；

  /* Logarithmically increment a counter. The greater is the current counter value
  • the less likely is that it gets really implemented. Saturate it at 255. /
    uint8_t LFULogIncr(uint8_t counter) {
    if (counter == 255) return 255;
    double r = (double)rand()/RAND_MAX;
    double baseval = counter - LFU_INIT_VAL; //LFU_INIT_VAL：5，初始化次数，key被添加就会有5次
    if (baseval < 0) baseval = 0;
    double p = 1.0/(basevalserver.lfu_log_factor+1); //server.lfu_log_factor:10
    if (r < p) counter++;
    return counter;
    }
  /* Return the current time in minutes, just taking the least significant
  • 16 bits. The returned time is suitable to be stored as LDT (last decrement
  • time) for the LFU implementation. */
    unsigned long LFUGetTimeInMinutes(void) {
    return (server.unixtime/60) & 65535; //由此可见，unixtime的单位是秒
    }
  /* Update LFU when an object is accessed.
  • Firstly, decrement the counter if the decrement time is reached.
  • Then logarithmically increment the counter, and update the access time. /
    void updateLFU(robj val) {
    unsigned long counter = LFUDecrAndReturn(val);
    counter = LFULogIncr(counter);
    val->lru = (LFUGetTimeInMinutes()<<8) | counter;
    }

接着看find

#define dictGetVal(he) ((he)->v.val)

dictEntry数据结构

dict.h
typedef struct dictEntry {
    void *key;
    union {
        void *val;
        uint64_t u64;
        int64_t s64;  //过期时间，到这个时间就过期了
        double d;
    } v;
    struct dictEntry *next;
} dictEntry;

setExpire(c,c->db,key,mstime()+milliseconds)

/* Set an expire to the specified key. If the expire is set in the context
 * of an user calling a command 'c' is the client, otherwise 'c' is set
 * to NULL. The 'when' parameter is the absolute unix time in milliseconds
 * after which the key will no longer be considered valid. */
void setExpire(client *c, redisDb *db, robj *key, long long when) {
    dictEntry *kde, *de;

    /* Reuse the sds from the main dict in the expire dict */
    kde = dictFind(db->dict,key->ptr);
    serverAssertWithInfo(NULL,key,kde != NULL);
    de = dictAddOrFind(db->expires,dictGetKey(kde));
    dictSetSignedIntegerVal(de,when);

    int writable_slave = server.masterhost && server.repl_slave_ro == 0;
    if (c && writable_slave && !(c->flags & CLIENT_MASTER))
        rememberSlaveKeyWithExpire(db,key);
}

AddOrFind

/* Add or Find:
 * dictAddOrFind() is simply a version of dictAddRaw() that always
 * returns the hash entry of the specified key, even if the key already
 * exists and can't be added (in that case the entry of the already
 * existing key is returned.)
 *
 * See dictAddRaw() for more information. */
dictEntry *dictAddOrFind(dict *d, void *key) {
    dictEntry *entry, *existing;
    entry = dictAddRaw(d,key,&existing);
    return entry ? entry : existing;
}

#define dictSetSignedIntegerVal(entry, _val_) \
do { (entry)->v.s64 = _val_; } while(0)

get命令

void getCommand(client *c) {
    getGenericCommand(c);
}

int getGenericCommand(client *c) {
    robj *o;

    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL)
        return C_OK;

    if (o->type != OBJ_STRING) {
        //这个error并非是指响应的value type判断，而是数据类型的type判断，如这个key本身对应的type是list、set，不能直接用get命令
        addReply(c,shared.wrongtypeerr);
        return C_ERR;
    } else {
        addReplyBulk(c,o);
        return C_OK;
    }
}

robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply) {
    robj *o = lookupKeyRead(c->db, key);
    if (!o) addReply(c,reply);
    return o;
}

/* Like lookupKeyReadWithFlags(), but does not use any flag, which is the
 * common case. */
robj *lookupKeyRead(redisDb *db, robj *key) {
    return lookupKeyReadWithFlags(db,key,LOOKUP_NONE);
}

/* Lookup a key for read operations, or return NULL if the key is not found
 * in the specified DB.
 *
 * As a side effect of calling this function:
 * 1. A key gets expired if it reached it's TTL.
 * 2. The key last access time is updated.
 * 3. The global keys hits/misses stats are updated (reported in INFO).
 *
 * This API should not be used when we write to the key after obtaining
 * the object linked to the key, but only for read only operations.
 *
 * Flags change the behavior of this command:
 *
 *  LOOKUP_NONE (or zero): no special flags are passed.
 *  LOOKUP_NOTOUCH: don't alter the last access time of the key.
 *
 * Note: this function also returns NULL is the key is logically expired
 * but still existing, in case this is a slave, since this API is called only
 * for read operations. Even if the key expiry is master-driven, we can
 * correctly report a key is expired on slaves even if the master is lagging
 * expiring our key via DELs in the replication link. */
robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags) {
    robj *val;

    if (expireIfNeeded(db,key) == 1) {
        /* Key expired. If we are in the context of a master, expireIfNeeded()
         * returns 0 only when the key does not exist at all, so it's safe
         * to return NULL ASAP. */
        if (server.masterhost == NULL) return NULL;

        /* However if we are in the context of a slave, expireIfNeeded() will
         * not really try to expire the key, it only returns information
         * about the "logical" status of the key: key expiring is up to the
         * master in order to have a consistent view of master's data set.
         *
         * However, if the command caller is not the master, and as additional
         * safety measure, the command invoked is a read-only command, we can
         * safely return NULL here, and provide a more consistent behavior
         * to clients accessign expired values in a read-only fashion, that
         * will say the key as non exisitng.
         *
         * Notably this covers GETs when slaves are used to scale reads. */
        if (server.current_client &&
            server.current_client != server.master &&
            server.current_client->cmd &&
            server.current_client->cmd->flags & CMD_READONLY)
        {
            return NULL;
        }
    }
    val = lookupKey(db,key,flags);
    if (val == NULL)
        server.stat_keyspace_misses++;
    else
        server.stat_keyspace_hits++;
    return val;
}

robj:

typedef struct redisObject {
    unsigned type:4;
    unsigned encoding:4;
    unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
                            * LFU data (least significant 8 bits frequency
                            * and most significant 16 bits access time). */
    int refcount;
    void *ptr;
} robj;