Unverified Commit 7b9e9606 authored by debing.sun's avatar debing.sun Committed by GitHub
Browse files

Hash Field Expiration (#13303)

## Background

This PR introduces support for field-level expiration in Redis hashes. Previously, Redis supported expiration only at the key level, but this enhancement allows setting expiration times for individual fields within a hash.

## New commands
* HEXPIRE
* HEXPIREAT
* HEXPIRETIME
* HPERSIST
* HPEXPIRE
* HPEXPIREAT
* HPEXPIRETIME
* HPTTL
* HTTL

## Short example
from @moticless
```sh
127.0.0.1:6379>  hset myhash f1 v1 f2 v2 f3 v3                                                   
(integer) 3
127.0.0.1:6379>  hpexpire myhash 10000 NX fields 2 f2 f3                                         
1) (integer) 1
2) (integer) 1
127.0.0.1:6379>  hpttl myhash fields 3 f1 f2 f3                                                                                                                                                                         
1) (integer) -1
2) (integer) 9997
3) (integer) 9997
127.0.0.1:6379>  hgetall myhash  
1) "f3"
2) "v3"
3) "f2"
4) "v2"
5) "f1"
6) "v1"

... after 10 seconds ...

127.0.0.1:6379>  hgetall myhash  
1) "f1"
2) "v1"
127.0.0.1:6379>
```

## Expiration strategy
1. Integrate active
    Redis periodically performs active expiration and deletion of hash keys that contain expired fields, with a maximum attempt limit.
3. Lazy expiration
    When a client touches fields within a hash, Redis checks if the fields are expired. If a field is expired, it will be deleted. However, we do not delete expired fields during a traversal, we implicitly skip over them.

## RDB changes
Add two new rdb type s`RDB_TYPE_HASH_METADATA` and `RDB_TYPE_HASH_LISTPACK_EX`.

## Notification
1. Add `hpersist` notification for `HPERSIST` command.
5. Add `hexpire` notification for `HEXPIRE`, `HEXPIREAT`, `HPEXPIRE` and `HPEXPIREAT` commands.

## Internal
1. Add new data structure `ebuckets`, which is used to store TTL and keys, enabling quick retrieval of keys based on TTL.
2. Add new data structure `mstr` like sds, which is used to store a string with TTL.

This work was done by @moticless, @tezc, @ronen-kalish, @sundb, I just release it.
parents 2d1bb42c f0389f28
......@@ -94,6 +94,7 @@ int activeExpireCycleTryExpire(redisDb *db, dictEntry *de, long long now) {
#define ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC 25 /* Max % of CPU to use. */
#define ACTIVE_EXPIRE_CYCLE_ACCEPTABLE_STALE 10 /* % of stale keys after which
we do extra efforts. */
#define HFE_ACTIVE_EXPIRE_CYCLE_FIELDS 1000
/* Data used by the expire dict scan callback. */
typedef struct {
......@@ -134,6 +135,53 @@ static inline int isExpiryDictValidForSamplingCb(dict *d) {
return C_OK;
}
/* Active expiration Cycle for hash-fields.
*
* Note that releasing fields is expected to be more predictable and rewarding
* than releasing keys because it is stored in `ebuckets` DS which optimized for
* active expiration and in addition the deletion of fields is simple to handle. */
static inline void activeExpireHashFieldCycle(int type) {
/* Remember current db across calls */
static unsigned int currentDb = 0;
/* Tracks the count of fields actively expired for the current database.
* This count continues as long as it fails to actively expire all expired
* fields of currentDb, indicating a possible need to adjust the value of
* maxToExpire. */
static uint64_t activeExpirySequence = 0;
/* Threshold for adjusting maxToExpire */
const uint32_t EXPIRED_FIELDS_TH = 1000000;
/* Maximum number of fields to actively expire in a single call */
uint32_t maxToExpire = HFE_ACTIVE_EXPIRE_CYCLE_FIELDS;
redisDb *db = server.db + currentDb;
/* If db is empty, move to next db and return */
if (ebIsEmpty(db->hexpires)) {
activeExpirySequence = 0;
currentDb = (currentDb + 1) % server.dbnum;
return;
}
/* If running for a while and didn't manage to active-expire all expired fields of
* currentDb (i.e. activeExpirySequence becomes significant) then adjust maxToExpire */
if ((activeExpirySequence > EXPIRED_FIELDS_TH) && (type == ACTIVE_EXPIRE_CYCLE_SLOW)) {
/* maxToExpire is multiplied by a factor between 1 and 32, proportional to
* the number of times activeExpirySequence exceeded EXPIRED_FIELDS_TH */
uint64_t factor = activeExpirySequence / EXPIRED_FIELDS_TH;
maxToExpire *= (factor<32) ? factor : 32;
}
if (hashTypeDbActiveExpire(db, maxToExpire) == maxToExpire) {
/* active-expire reached maxToExpire limit */
activeExpirySequence += maxToExpire;
} else {
/* Managed to active-expire all expired fields of currentDb */
activeExpirySequence = 0;
currentDb = (currentDb + 1) % server.dbnum;
}
}
void activeExpireCycle(int type) {
/* Adjust the running parameters according to the configured expire
* effort. The default effort is 1, and the maximum configurable effort
......@@ -232,6 +280,11 @@ void activeExpireCycle(int type) {
* distribute the time evenly across DBs. */
current_db++;
/* Interleaving hash-field expiration with key expiration. Better
* call it before handling expired keys because HFE DS is optimized for
* active expiration */
activeExpireHashFieldCycle(type);
if (kvstoreSize(db->expires))
dbs_performed++;
......
......@@ -3,6 +3,7 @@
#include "atomicvar.h"
#include "functions.h"
#include "cluster.h"
#include "ebuckets.h"
static redisAtomic size_t lazyfree_objects = 0;
static redisAtomic size_t lazyfreed_objects = 0;
......@@ -22,7 +23,8 @@ void lazyfreeFreeObject(void *args[]) {
void lazyfreeFreeDatabase(void *args[]) {
kvstore *da1 = args[0];
kvstore *da2 = args[1];
ebuckets oldHfe = args[2];
ebDestroy(&oldHfe, &hashExpireBucketsType, NULL);
size_t numkeys = kvstoreSize(da1);
kvstoreRelease(da1);
kvstoreRelease(da2);
......@@ -201,10 +203,12 @@ void emptyDbAsync(redisDb *db) {
flags |= KVSTORE_FREE_EMPTY_DICTS;
}
kvstore *oldkeys = db->keys, *oldexpires = db->expires;
ebuckets oldHfe = db->hexpires;
db->keys = kvstoreCreate(&dbDictType, slot_count_bits, flags);
db->expires = kvstoreCreate(&dbExpiresDictType, slot_count_bits, flags);
db->hexpires = ebCreate();
atomicIncr(lazyfree_objects, kvstoreSize(oldkeys));
bioCreateLazyFreeJob(lazyfreeFreeDatabase, 2, oldkeys, oldexpires);
bioCreateLazyFreeJob(lazyfreeFreeDatabase, 3, oldkeys, oldexpires, oldHfe);
}
/* Free the key tracking table.
......
This diff is collapsed.
......@@ -49,18 +49,25 @@ unsigned char *lpReplaceInteger(unsigned char *lp, unsigned char **p, long long
unsigned char *lpDelete(unsigned char *lp, unsigned char *p, unsigned char **newp);
unsigned char *lpDeleteRangeWithEntry(unsigned char *lp, unsigned char **p, unsigned long num);
unsigned char *lpDeleteRange(unsigned char *lp, long index, unsigned long num);
unsigned char *lpBatchAppend(unsigned char *lp, listpackEntry *entries, unsigned long len);
unsigned char *lpBatchInsert(unsigned char *lp, unsigned char *p, int where,
listpackEntry *entries, unsigned int len, unsigned char **newp);
unsigned char *lpBatchDelete(unsigned char *lp, unsigned char **ps, unsigned long count);
unsigned char *lpMerge(unsigned char **first, unsigned char **second);
unsigned char *lpDup(unsigned char *lp);
unsigned long lpLength(unsigned char *lp);
unsigned char *lpGet(unsigned char *p, int64_t *count, unsigned char *intbuf);
unsigned char *lpGetValue(unsigned char *p, unsigned int *slen, long long *lval);
int lpGetIntegerValue(unsigned char *p, long long *lval);
unsigned char *lpFind(unsigned char *lp, unsigned char *p, unsigned char *s, uint32_t slen, unsigned int skip);
typedef int (*lpCmp)(const unsigned char *lp, unsigned char *p, void *user, unsigned char *s, long long slen);
unsigned char *lpFindCb(unsigned char *lp, unsigned char *p, void *user, lpCmp cmp, unsigned int skip);
unsigned char *lpFirst(unsigned char *lp);
unsigned char *lpLast(unsigned char *lp);
unsigned char *lpNext(unsigned char *lp, unsigned char *p);
unsigned char *lpPrev(unsigned char *lp, unsigned char *p);
size_t lpBytes(unsigned char *lp);
size_t lpEntrySizeInteger(long long lval);
size_t lpEstimateBytesRepeatedInteger(long long lval, unsigned long rep);
unsigned char *lpSeek(unsigned char *lp, long index);
typedef int (*listpackValidateEntryCB)(unsigned char *p, unsigned int head_count, void *userdata);
......@@ -69,12 +76,15 @@ int lpValidateIntegrity(unsigned char *lp, size_t size, int deep,
unsigned char *lpValidateFirst(unsigned char *lp);
int lpValidateNext(unsigned char *lp, unsigned char **pp, size_t lpbytes);
unsigned int lpCompare(unsigned char *p, unsigned char *s, uint32_t slen);
void lpRandomPair(unsigned char *lp, unsigned long total_count, listpackEntry *key, listpackEntry *val);
void lpRandomPairs(unsigned char *lp, unsigned int count, listpackEntry *keys, listpackEntry *vals);
unsigned int lpRandomPairsUnique(unsigned char *lp, unsigned int count, listpackEntry *keys, listpackEntry *vals);
void lpRandomPair(unsigned char *lp, unsigned long total_count,
listpackEntry *key, listpackEntry *val, int tuple_len);
void lpRandomPairs(unsigned char *lp, unsigned int count,
listpackEntry *keys, listpackEntry *vals, int tuple_len);
unsigned int lpRandomPairsUnique(unsigned char *lp, unsigned int count,
listpackEntry *keys, listpackEntry *vals, int tuple_len);
void lpRandomEntries(unsigned char *lp, unsigned int count, listpackEntry *entries);
unsigned char *lpNextRandom(unsigned char *lp, unsigned char *p, unsigned int *index,
unsigned int remaining, int even_only);
unsigned int remaining, int tuple_len);
int lpSafeToAdd(unsigned char* lp, size_t add);
void lpRepr(unsigned char *lp);
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment