Redis/redis
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Optimize COPY, RENAME and RESTORE commands with TTL (#14088)

Browse Source 
* DEL optimized to call `kvstoreDictDelete(db->expires ...)` only when
O(1) `kvobjGetExpire() != -1`
* Combines the two funcitons `dbAdd()` and `setExpire()` into a single 
efficient call: `dbAdd(..., long long expire)`. This optimization
eliminates an extra lookup
and also avoid reallocating a new object to store the TTL.

**Benchmarking RESTORE command with TTL:**
```
memtier_benchmark --command "RESTORE bla 1000000 \"\\x00\\x03bbb\\x0c\\x00\\x8e\\x85\\xaf\\x9f\\x0e'#\\x00\" REPLACE" --command-key-pattern=P --data-size=1 --pipeline=1000 --key-maximum=10000000 -c 5 -t 20 --hide-histogram --requests=100000
```
**Results Summary (Average of 3 Runs):**

Metric | unstable | optimize-setExpire | Δ (Improvement)
-- | -- | -- | --
Throughput (ops/sec) | 1,614,176 | 1,737,198 | +7.6%
P50 Latency (ms) | 61.25 | 57.30 | –6.5%
P95 Latency (ms) | 64.77 | 59.99 | –7.4%
P99 Latency (ms) | 73.73 | 70.83 | –3.9%
Max Latency (ms) | 87.55 | 85.84 | –2.0%
This commit is contained in:
Moti Cohen
2025-06-16 15:11:48 +03:00
committed by GitHub
parent 94aebb7324
commit 90178712f6
4 changed files with 71 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cluster.c
+1 -2
View File
@@ -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);
src/db.c
+39 -26
View File
@@ -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 */
src/object.c
+28 -25
View File
@@ -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);
src/server.h
+3 -2
View File
@@ -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);