diff --git a/src/cluster.c b/src/cluster.c index e3f6e90d7..18a6cdd4b 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -239,7 +239,7 @@ void restoreCommand(client *c) { } /* Create the key and set the TTL if any */ - kvobj *kv = dbAdd(c->db, key, &obj); + kvobj *kv = dbAddInternal(c->db, key, &obj, NULL, ttl ? ttl : -1); /* If minExpiredField was set, then the object is hash with expiration * on fields and need to register it in global HFE DS */ @@ -250,7 +250,6 @@ void restoreCommand(client *c) { } if (ttl) { - kv = setExpire(c,c->db,key,ttl); /* might realloc kvobj */ if (!absttl) { /* Propagate TTL as absolute timestamp */ robj *ttl_obj = createStringObjectFromLongLong(ttl); diff --git a/src/db.c b/src/db.c index d44a7f2d3..2734ac04e 100644 --- a/src/db.c +++ b/src/db.c @@ -321,13 +321,22 @@ kvobj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply) { * * link - Optional link to bucket where the key should be added. * On return, get updated, by need, to the inserted key. + * + * expire - Set expiry of the key. -1 for no expiry. */ -kvobj *dbAddByLink(redisDb *db, robj *key, robj **valref, dictEntryLink *link) { +kvobj *dbAddInternal(redisDb *db, robj *key, robj **valref, dictEntryLink *link, long long expire) { int slot = getKeySlot(key->ptr); + dictEntryLink tmp = NULL; + if (link == NULL) link = &tmp; robj *val = *valref; - kvobj *kv = kvobjSet(key->ptr, val, -1); + int hasExpire = expire != -1; + kvobj *kv = kvobjSet(key->ptr, val, hasExpire); initObjectLRUOrLFU(kv); kvstoreDictSetAtLink(db->keys, slot, kv, link, 1); + + /* Add to expires. Leverage setExpireByLink() to reuse the key link. */ + if (hasExpire) kv = setExpireByLink(NULL, db, key->ptr, expire, *link); + signalKeyAsReady(db, key, kv->type); notifyKeyspaceEvent(NOTIFY_NEW,"new",key,db->id); updateKeysizesHist(db, slot, kv->type, -1, getObjectLength(kv)); /* add hist */ @@ -335,9 +344,13 @@ kvobj *dbAddByLink(redisDb *db, robj *key, robj **valref, dictEntryLink *link) { return kv; } -/* Read dbAddByLink() comment */ +/* Read dbAddInternal() comment */ kvobj *dbAdd(redisDb *db, robj *key, robj **valref) { - return dbAddByLink(db, key, valref, NULL); + return dbAddInternal(db, key, valref, NULL, -1); +} + +kvobj *dbAddByLink(redisDb *db, robj *key, robj **valref, dictEntryLink *link) { + return dbAddInternal(db, key, valref, link, -1); } /* Returns key's hash slot when cluster mode is enabled, or 0 when disabled. @@ -403,7 +416,7 @@ kvobj *dbAddRDBLoad(redisDb *db, sds key, robj **valref, long long expire) { return NULL; /* prepare kvobj for insertion. Pass expire to reserve space for it */ - kvobj *kv = kvobjSet(key, *valref, -1); + kvobj *kv = kvobjSet(key, *valref, expire != -1); initObjectLRUOrLFU(kv); kvstoreDictSetAtLink(db->keys, slot, kv, &bucket, 1); @@ -486,12 +499,14 @@ static void dbSetValue(redisDb *db, robj *key, robj **valref, dictEntryLink link val->lru = old->lru; /* Update expire reference if needed */ long long expire = getExpire(db, key->ptr, old); - kvNew = kvobjSet(key->ptr, val, keepTTL ? expire : -1); + int hasExpire = keepTTL && (expire != -1); + kvNew = kvobjSet(key->ptr, val, hasExpire); kvstoreDictSetAtLink(db->keys, slot, kvNew, &link, 0); /* Replace the old value at its location in the expire space. */ if (expire >= 0) { if (keepTTL) { + kvobjSetExpire(kvNew, expire); /* kvNew not reallocated here */ dictEntryLink exLink = kvstoreDictFindLink(db->expires, slot, key->ptr, NULL); serverAssertWithInfo(NULL, key, exLink != NULL); @@ -643,32 +658,35 @@ int dbGenericDelete(redisDb *db, robj *key, int async, int flags) { link = kvstoreDictTwoPhaseUnlinkFind(db->keys, slot, key->ptr, &table); if (link) { - kvobj *val = dictGetKV(*link); + kvobj *kv = dictGetKV(*link); - int64_t oldlen = (int64_t) getObjectLength(val); - int type = val->type; + int64_t oldlen = (int64_t) getObjectLength(kv); + int type = kv->type; /* If hash object with expiry on fields, remove it from HFE DS of DB */ if (type == OBJ_HASH) - hashTypeRemoveFromExpires(&db->hexpires, val); + hashTypeRemoveFromExpires(&db->hexpires, kv); - /* RM_StringDMA may call dbUnshareStringValue which may free val, so we - * need to incr to retain val */ - incrRefCount(val); /* refcnt=1->2 */ + /* RM_StringDMA may call dbUnshareStringValue which may free kv, so we + * need to incr to retain kv */ + incrRefCount(kv); /* refcnt=1->2 */ /* Tells the module that the key has been unlinked from the database. */ - moduleNotifyKeyUnlink(key,val,db->id,flags); + moduleNotifyKeyUnlink(key, kv, db->id, flags); /* We want to try to unblock any module clients or clients using a blocking XREADGROUP */ signalDeletedKeyAsReady(db,key,type); /* We should call decr before freeObjAsync. If not, the refcount may be * greater than 1, so freeObjAsync doesn't work */ - decrRefCount(val); + decrRefCount(kv); - /* Delete an entry from the expires dict is not decrRefCount of kvobj */ - kvstoreDictDelete(db->expires, slot, key->ptr); + /* Because of dbUnshareStringValue, the val in db may change. */ + kv = dictGetKV(*link); + + /* if expirable, delete an entry from the expires dict is not decrRefCount of kvobj */ + if (kvobjGetExpire(kv) != -1) + kvstoreDictDelete(db->expires, slot, key->ptr); if (async) { - /* Because of dbUnshareStringValue, the val in db may change. */ - freeObjAsync(key, dictGetKV(*link), db->id); + freeObjAsync(key, kv, db->id); /* Set the key to NULL in the main dictionary. */ kvstoreDictSetAtLink(db->keys, slot, NULL, &link, 0); } @@ -1835,8 +1853,7 @@ void renameGenericCommand(client *c, int nx) { minHashExpireTime = hashTypeRemoveFromExpires(&c->db->hexpires, o); dbDelete(c->db,c->argv[1]); - dbAdd(c->db, c->argv[2], &o); - if (expire != -1) o = setExpire(c, c->db, c->argv[2], expire); + dbAddInternal(c->db, c->argv[2], &o, NULL, expire); /* If hash with HFEs, register in db->hexpires */ if (minHashExpireTime != EB_EXPIRE_TIME_INVALID) @@ -2035,11 +2052,7 @@ void copyCommand(client *c) { dbDelete(dst,newkey); } - kvobj *kvCopy = dbAdd(dst, newkey, &newobj); - - /* if key with expiration then set it */ - if (expire != -1) - newobj = setExpire(c, dst, newkey, expire); + kvobj *kvCopy = dbAddInternal(dst, newkey, &newobj, NULL, expire); /* If minExpiredField was set, then the object is hash with expiration * on fields and need to register it in global HFE DS */ diff --git a/src/object.c b/src/object.c index 8a70c0fb4..0675db400 100644 --- a/src/object.c +++ b/src/object.c @@ -36,10 +36,10 @@ * | robj (16) | expiry (8) | key-hdr-size (1) | sdshdr5 "mykey" \0 (7) | * +-----------+------------+------------------+------------------------+ */ -kvobj *kvobjCreate(int type, const sds key, void *ptr, long long expire) { +kvobj *kvobjCreate(int type, const sds key, void *ptr, int hasExpire) { /* Determine embedded key and expiration flags */ serverAssert(key != NULL); - int has_expire = ((expire != -1) || (sdslen(key) >= KEY_SIZE_TO_INCLUDE_EXPIRE_THRESHOLD)); + hasExpire = hasExpire || (sdslen(key) >= KEY_SIZE_TO_INCLUDE_EXPIRE_THRESHOLD); /* Calculate embedded key size */ size_t key_sds_len = sdslen(key); @@ -48,7 +48,7 @@ kvobj *kvobjCreate(int type, const sds key, void *ptr, long long expire) { /* Compute the base object size */ size_t min_size = sizeof(robj); - if (has_expire) min_size += sizeof(long long); + if (hasExpire) min_size += sizeof(long long); min_size += 1 + key_sds_size; /* 1 byte for SDS header size */ /* Allocate object memory */ @@ -62,18 +62,18 @@ kvobj *kvobjCreate(int type, const sds key, void *ptr, long long expire) { o->iskvobj = 1; /* If extra space allows, pre-allocate anyway expiration */ - if ((!has_expire) && (bufsize >= min_size + sizeof(long long))) { - has_expire = 1; + if ((!hasExpire) && (bufsize >= min_size + sizeof(long long))) { + hasExpire = 1; min_size += sizeof(long long); } - o->expirable = has_expire; + o->expirable = hasExpire; /* The memory after the struct where we embedded data. */ char *data = (void *)(o + 1); /* Set the expire field. */ if (o->expirable) { - *(long long *)data = expire; + *(long long *)data = -1; data += sizeof(long long); } @@ -141,7 +141,7 @@ robj *createRawStringObject(const char *ptr, size_t len) { * +-----------+------------------+------------------------+----------------------------+ */ static kvobj *kvobjCreateEmbedString(const char *val_ptr, size_t val_len, - const sds key, long long expire) + const sds key, int hasExpire) { serverAssert(key != NULL); @@ -156,7 +156,7 @@ static kvobj *kvobjCreateEmbedString(const char *val_ptr, size_t val_len, /* Compute base object size */ size_t min_size = sizeof(robj) + val_sds_size; - if (expire != -1) min_size += sizeof(long long); + if (hasExpire != 0) min_size += sizeof(long long); min_size += 1 + key_sds_size; /* 1 byte for SDS header size */ /* Allocate object memory */ @@ -166,7 +166,7 @@ static kvobj *kvobjCreateEmbedString(const char *val_ptr, size_t val_len, o->encoding = OBJ_ENCODING_EMBSTR; o->refcount = 1; o->lru = 0; - o->expirable = (expire != -1); + o->expirable = (hasExpire != 0); o->iskvobj = 1; /* If the allocation has enough space for an expire field, add it even if we @@ -181,7 +181,7 @@ static kvobj *kvobjCreateEmbedString(const char *val_ptr, size_t val_len, /* Set the expire field. */ if (o->expirable) { - *(long long *)data = expire; + *(long long *)data = -1; data += sizeof(long long); } @@ -257,22 +257,25 @@ long long kvobjGetExpire(const kvobj *kv) { * the old object's reference counter is decremented and possibly freed. Use the * returned object instead of 'val' after calling this function. */ kvobj *kvobjSetExpire(kvobj *kv, long long expire) { - if (kv->expirable) { - /* Update existing expire field. */ - unsigned char *data = (void *)(kv + 1); - *(long long *)data = expire; - return kv; - } else if (expire == -1) { - return kv; - } else { - return kvobjSet(kvobjGetKey(kv), kv, expire); + if (!kv->expirable) { + /* Nothing to do if kv not expirable and expire is -1 */ + if (expire == -1) + return kv; + + /* Reallocate kvobj to add expire field. */ + kv = kvobjSet(kvobjGetKey(kv), kv, 1 /*hasExpire*/); } + + /* kv is expirable. Update expire field. */ + unsigned char *data = (void *)(kv + 1); + *(long long *)data = expire; + return kv; } /* This functions may reallocate the value. The new allocation is returned and * the old object's reference counter is decremented and possibly freed. Use the * returned object instead of 'val' after calling this function. */ -kvobj *kvobjSet(sds key, robj *val, long long expire) { +kvobj *kvobjSet(sds key, robj *val, int hasExpire) { if (val->type == OBJ_STRING && val->encoding == OBJ_ENCODING_EMBSTR) { kvobj *kv; size_t len = sdslen(val->ptr); @@ -280,12 +283,12 @@ kvobj *kvobjSet(sds key, robj *val, long long expire) { /* Embed when the sum is up to 64 bytes. */ size_t size = sizeof(kvobj); size += (key != NULL) * (sdslen(key) + 3); /* hdr size (1) + hdr (1) + nullterm (1) */ - size += (expire != -1) * sizeof(long long); + size += (!!hasExpire) * sizeof(long long); size += 4 + len; /* embstr header (3) + nullterm (1) */ if (size <= CACHE_LINE_SIZE) { - kv = kvobjCreateEmbedString(val->ptr, len, key, expire); + kv = kvobjCreateEmbedString(val->ptr, len, key, hasExpire); } else { - kv = kvobjCreate(OBJ_STRING, key, sdsnewlen(val->ptr, len), expire); + kv = kvobjCreate(OBJ_STRING, key, sdsnewlen(val->ptr, len), hasExpire); } kv->lru = val->lru; @@ -310,7 +313,7 @@ kvobj *kvobjSet(sds key, robj *val, long long expire) { * can be duplicated, but for a module type is not always possible. */ serverPanic("Not implemented"); } - robj *new = kvobjCreate(val->type, key, valptr, expire); + robj *new = kvobjCreate(val->type, key, valptr, hasExpire); new->encoding = val->encoding; new->lru = val->lru; decrRefCount(val); diff --git a/src/server.h b/src/server.h index a434dad76..e75fb8c18 100644 --- a/src/server.h +++ b/src/server.h @@ -3065,8 +3065,8 @@ unsigned long long estimateObjectIdleTime(robj *o); void trimStringObjectIfNeeded(robj *o, int trim_small_values); #define sdsEncodedObject(objptr) (objptr->encoding == OBJ_ENCODING_RAW || objptr->encoding == OBJ_ENCODING_EMBSTR) -kvobj *kvobjCreate(int type, const sds key, void *ptr, long long expire); -kvobj *kvobjSet(sds key, robj *val, long long expire); +kvobj *kvobjCreate(int type, const sds key, void *ptr, int hasExpire); +kvobj *kvobjSet(sds key, robj *val, int hasExpire); kvobj *kvobjSetExpire(kvobj *kv, long long expire); sds kvobjGetKey(const kvobj *kv); long long kvobjGetExpire(const kvobj *val); @@ -3636,6 +3636,7 @@ int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle, static inline kvobj *dictGetKV(const dictEntry *de) {return (kvobj *) dictGetKey(de);} kvobj *dbAdd(redisDb *db, robj *key, robj **valref); kvobj *dbAddByLink(redisDb *db, robj *key, robj **valref, dictEntryLink *link); +kvobj *dbAddInternal(redisDb *db, robj *key, robj **valref, dictEntryLink *link, long long expire); kvobj *dbAddRDBLoad(redisDb *db, sds key, robj **valref, long long expire); void dbReplaceValue(redisDb *db, robj *key, kvobj **ioKeyVal, int updateKeySizes); void dbReplaceValueWithLink(redisDb *db, robj *key, robj **val, dictEntryLink link);