Unverified Commit 0637b4ea authored by YaacovHazan's avatar YaacovHazan Committed by GitHub
Browse files

Redis 7.4 RC2 (#13371)

Upgrade urgency LOW: This is the second Release Candidate for Redis 7.4.

Performance and resource utilization improvements
=================================================
* #13296 Optimize CPU cache efficiency

Changes to new 7.4 new features (compared to 7.4 RC1)
=====================================================
* #13343 Hash - expiration of individual fields: when key does not exist
- reply with an array (nonexisting code for each field)
* #13329 Hash - expiration of individual fields: new keyspace event:
`hexpired`

Modules API - Potentially breaking changes to new 7.4 features (compared
to 7.4 RC1)

====================================================================================
* #13326 Hash - expiration of individual fields: avoid lazy expire when
called from a Modules API function
parents 4606f91d 4000bb2e
...@@ -175,6 +175,7 @@ void rdbCheckSetupSignals(void) { ...@@ -175,6 +175,7 @@ void rdbCheckSetupSignals(void) {
* otherwise the already open file 'fp' is checked. */ * otherwise the already open file 'fp' is checked. */
int redis_check_rdb(char *rdbfilename, FILE *fp) { int redis_check_rdb(char *rdbfilename, FILE *fp) {
uint64_t dbid; uint64_t dbid;
int selected_dbid = -1;
int type, rdbver; int type, rdbver;
char buf[1024]; char buf[1024];
long long expiretime, now = mstime(); long long expiretime, now = mstime();
...@@ -246,6 +247,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -246,6 +247,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
if ((dbid = rdbLoadLen(&rdb,NULL)) == RDB_LENERR) if ((dbid = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
goto eoferr; goto eoferr;
rdbCheckInfo("Selecting DB ID %llu", (unsigned long long)dbid); rdbCheckInfo("Selecting DB ID %llu", (unsigned long long)dbid);
selected_dbid = dbid;
continue; /* Read type again. */ continue; /* Read type again. */
} else if (type == RDB_OPCODE_RESIZEDB) { } else if (type == RDB_OPCODE_RESIZEDB) {
/* RESIZEDB: Hint about the size of the keys in the currently /* RESIZEDB: Hint about the size of the keys in the currently
...@@ -331,7 +333,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -331,7 +333,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
rdbstate.keys++; rdbstate.keys++;
/* Read value */ /* Read value */
rdbstate.doing = RDB_CHECK_DOING_READ_OBJECT_VALUE; rdbstate.doing = RDB_CHECK_DOING_READ_OBJECT_VALUE;
if ((val = rdbLoadObject(type,&rdb,key->ptr,NULL,NULL,NULL)) == NULL) if ((val = rdbLoadObject(type,&rdb,key->ptr,selected_dbid,NULL)) == NULL)
goto eoferr; goto eoferr;
/* Check if the key already expired. */ /* Check if the key already expired. */
if (expiretime != -1 && expiretime < now) if (expiretime != -1 && expiretime < now)
......
...@@ -189,6 +189,9 @@ int canFeedReplicaReplBuffer(client *replica) { ...@@ -189,6 +189,9 @@ int canFeedReplicaReplBuffer(client *replica) {
/* Don't feed replicas that are still waiting for BGSAVE to start. */ /* Don't feed replicas that are still waiting for BGSAVE to start. */
if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_START) return 0; if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_START) return 0;
/* Don't feed replicas that are going to be closed ASAP. */
if (replica->flags & CLIENT_CLOSE_ASAP) return 0;
return 1; return 1;
} }
......
...@@ -74,6 +74,7 @@ extern scriptFlag scripts_flags_def[]; ...@@ -74,6 +74,7 @@ extern scriptFlag scripts_flags_def[];
void luaEnvInit(void); void luaEnvInit(void);
lua_State *createLuaState(void); lua_State *createLuaState(void);
dict *getLuaScripts(void);
uint64_t scriptFlagsToCmdFlags(uint64_t cmd_flags, uint64_t script_flags); uint64_t scriptFlagsToCmdFlags(uint64_t cmd_flags, uint64_t script_flags);
int scriptPrepareForRun(scriptRunCtx *r_ctx, client *engine_client, client *caller, const char *funcname, uint64_t script_flags, int ro); int scriptPrepareForRun(scriptRunCtx *r_ctx, client *engine_client, client *caller, const char *funcname, uint64_t script_flags, int ro);
void scriptResetRun(scriptRunCtx *r_ctx); void scriptResetRun(scriptRunCtx *r_ctx);
......
...@@ -334,6 +334,10 @@ uint64_t dictObjHash(const void *key) { ...@@ -334,6 +334,10 @@ uint64_t dictObjHash(const void *key) {
return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr)); return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
} }
uint64_t dictPtrHash(const void *key) {
return dictGenHashFunction((unsigned char*)&key,sizeof(key));
}
uint64_t dictSdsHash(const void *key) { uint64_t dictSdsHash(const void *key) {
return dictGenHashFunction((unsigned char*)key, sdslen((char*)key)); return dictGenHashFunction((unsigned char*)key, sdslen((char*)key));
} }
......
...@@ -2071,7 +2071,8 @@ struct redisServer { ...@@ -2071,7 +2071,8 @@ struct redisServer {
char *locale_collate; char *locale_collate;
}; };
#define MAX_KEYS_BUFFER 256 /* we use 6 so that all getKeyResult fits a cacheline */
#define MAX_KEYS_BUFFER 6
typedef struct { typedef struct {
int pos; /* The position of the key within the client array */ int pos; /* The position of the key within the client array */
...@@ -2084,12 +2085,12 @@ typedef struct { ...@@ -2084,12 +2085,12 @@ typedef struct {
* for returning channel information. * for returning channel information.
*/ */
typedef struct { typedef struct {
int numkeys; /* Number of key indices return */
int size; /* Available array size */
keyReference keysbuf[MAX_KEYS_BUFFER]; /* Pre-allocated buffer, to save heap allocations */ keyReference keysbuf[MAX_KEYS_BUFFER]; /* Pre-allocated buffer, to save heap allocations */
keyReference *keys; /* Key indices array, points to keysbuf or heap */ keyReference *keys; /* Key indices array, points to keysbuf or heap */
int numkeys; /* Number of key indices return */
int size; /* Available array size */
} getKeysResult; } getKeysResult;
#define GETKEYS_RESULT_INIT { {{0}}, NULL, 0, MAX_KEYS_BUFFER } #define GETKEYS_RESULT_INIT { 0, MAX_KEYS_BUFFER, {{0}}, NULL }
/* Key specs definitions. /* Key specs definitions.
* *
...@@ -3163,7 +3164,9 @@ robj *setTypeDup(robj *o); ...@@ -3163,7 +3164,9 @@ robj *setTypeDup(robj *o);
typedef struct listpackEx { typedef struct listpackEx {
ExpireMeta meta; /* To be used in order to register the hash in the ExpireMeta meta; /* To be used in order to register the hash in the
global ebuckets (i.e. db->hexpires) with next, global ebuckets (i.e. db->hexpires) with next,
minimum, hash-field to expire. */ minimum, hash-field to expire. TTL value might be
inaccurate up-to few seconds due to optimization
consideration. */
sds key; /* reference to the key, same one that stored in sds key; /* reference to the key, same one that stored in
db->dict. Will be used from active-expiration flow db->dict. Will be used from active-expiration flow
for notification and deletion of the object, if for notification and deletion of the object, if
...@@ -3178,7 +3181,9 @@ typedef struct dictExpireMetadata { ...@@ -3178,7 +3181,9 @@ typedef struct dictExpireMetadata {
ExpireMeta expireMeta; /* embedded ExpireMeta in dict. ExpireMeta expireMeta; /* embedded ExpireMeta in dict.
To be used in order to register the hash in the To be used in order to register the hash in the
global ebuckets (i.e db->hexpires) with next, global ebuckets (i.e db->hexpires) with next,
minimum, hash-field to expire */ minimum, hash-field to expire. TTL value might be
inaccurate up-to few seconds due to optimization
consideration. */
ebuckets hfe; /* DS of Hash Fields Expiration, associated to each hash */ ebuckets hfe; /* DS of Hash Fields Expiration, associated to each hash */
sds key; /* reference to the key, same one that stored in sds key; /* reference to the key, same one that stored in
db->dict. Will be used from active-expiration flow db->dict. Will be used from active-expiration flow
...@@ -3191,9 +3196,16 @@ typedef struct dictExpireMetadata { ...@@ -3191,9 +3196,16 @@ typedef struct dictExpireMetadata {
#define HASH_SET_TAKE_VALUE (1<<1) #define HASH_SET_TAKE_VALUE (1<<1)
#define HASH_SET_COPY 0 #define HASH_SET_COPY 0
/* Hash field lazy expiration flags. Used by core hashTypeGetValue() and its callers */
#define HFE_LAZY_EXPIRE (0) /* Delete expired field, and if last field also the hash */
#define HFE_LAZY_AVOID_FIELD_DEL (1<<0) /* Avoid deleting expired field */
#define HFE_LAZY_AVOID_HASH_DEL (1<<1) /* Avoid deleting hash if the field is the last one */
#define HFE_LAZY_NO_NOTIFICATION (1<<2) /* Do not send notification, used when multiple fields
* may expire and only one notification is desired. */
void hashTypeConvert(robj *o, int enc, ebuckets *hexpires); void hashTypeConvert(robj *o, int enc, ebuckets *hexpires);
void hashTypeTryConversion(redisDb *db, robj *subject, robj **argv, int start, int end); void hashTypeTryConversion(redisDb *db, robj *subject, robj **argv, int start, int end);
int hashTypeExists(redisDb *db, robj *o, sds key, int *isHashDeleted); int hashTypeExists(redisDb *db, robj *o, sds key, int hfeFlags, int *isHashDeleted);
int hashTypeDelete(robj *o, void *key, int isSdsField); int hashTypeDelete(robj *o, void *key, int isSdsField);
unsigned long hashTypeLength(const robj *o, int subtractExpiredFields); unsigned long hashTypeLength(const robj *o, int subtractExpiredFields);
hashTypeIterator *hashTypeInitIterator(robj *subject); hashTypeIterator *hashTypeInitIterator(robj *subject);
...@@ -3210,20 +3222,17 @@ void hashTypeCurrentObject(hashTypeIterator *hi, int what, unsigned char **vstr, ...@@ -3210,20 +3222,17 @@ void hashTypeCurrentObject(hashTypeIterator *hi, int what, unsigned char **vstr,
unsigned int *vlen, long long *vll, uint64_t *expireTime); unsigned int *vlen, long long *vll, uint64_t *expireTime);
sds hashTypeCurrentObjectNewSds(hashTypeIterator *hi, int what); sds hashTypeCurrentObjectNewSds(hashTypeIterator *hi, int what);
hfield hashTypeCurrentObjectNewHfield(hashTypeIterator *hi); hfield hashTypeCurrentObjectNewHfield(hashTypeIterator *hi);
robj *hashTypeGetValueObject(redisDb *db, robj *o, sds field, int *isHashDeleted); robj *hashTypeGetValueObject(redisDb *db, robj *o, sds field, int hfeFlags, int *isHashDeleted);
int hashTypeSet(redisDb *db, robj *o, sds field, sds value, int flags); int hashTypeSet(redisDb *db, robj *o, sds field, sds value, int flags);
robj *hashTypeDup(robj *o, sds newkey, uint64_t *minHashExpire); robj *hashTypeDup(robj *o, sds newkey, uint64_t *minHashExpire);
uint64_t hashTypeRemoveFromExpires(ebuckets *hexpires, robj *o); uint64_t hashTypeRemoveFromExpires(ebuckets *hexpires, robj *o);
void hashTypeAddToExpires(redisDb *db, sds key, robj *hashObj, uint64_t expireTime); void hashTypeAddToExpires(redisDb *db, sds key, robj *hashObj, uint64_t expireTime);
void hashTypeFree(robj *o); void hashTypeFree(robj *o);
int hashTypeIsExpired(const robj *o, uint64_t expireAt); int hashTypeIsExpired(const robj *o, uint64_t expireAt);
uint64_t hashTypeGetMinExpire(robj *o);
unsigned char *hashTypeListpackGetLp(robj *o); unsigned char *hashTypeListpackGetLp(robj *o);
uint64_t hashTypeGetMinExpire(robj *o); uint64_t hashTypeGetMinExpire(robj *o, int accurate);
void hashTypeUpdateKeyRef(robj *o, sds newkey); void hashTypeUpdateKeyRef(robj *o, sds newkey);
ebuckets *hashTypeGetDictMetaHFE(dict *d); ebuckets *hashTypeGetDictMetaHFE(dict *d);
uint64_t hashTypeGetMinExpire(robj *keyObj);
uint64_t hashTypeGetNextTimeToExpire(robj *o);
void initDictExpireMetadata(sds key, robj *o); void initDictExpireMetadata(sds key, robj *o);
struct listpackEx *listpackExCreate(void); struct listpackEx *listpackExCreate(void);
void listpackExAddNew(robj *o, char *field, size_t flen, void listpackExAddNew(robj *o, char *field, size_t flen,
...@@ -3531,6 +3540,7 @@ void startEvictionTimeProc(void); ...@@ -3531,6 +3540,7 @@ void startEvictionTimeProc(void);
/* Keys hashing / comparison functions for dict.c hash tables. */ /* Keys hashing / comparison functions for dict.c hash tables. */
uint64_t dictSdsHash(const void *key); uint64_t dictSdsHash(const void *key);
uint64_t dictPtrHash(const void *key);
uint64_t dictSdsCaseHash(const void *key); uint64_t dictSdsCaseHash(const void *key);
int dictSdsKeyCompare(dict *d, const void *key1, const void *key2); int dictSdsKeyCompare(dict *d, const void *key1, const void *key2);
int dictSdsMstrKeyCompare(dict *d, const void *sdsLookup, const void *mstrStored); int dictSdsMstrKeyCompare(dict *d, const void *sdsLookup, const void *mstrStored);
......
...@@ -95,7 +95,7 @@ robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) { ...@@ -95,7 +95,7 @@ robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
/* Retrieve value from hash by the field name. The returned object /* Retrieve value from hash by the field name. The returned object
* is a new object with refcount already incremented. */ * is a new object with refcount already incremented. */
int isHashDeleted; int isHashDeleted;
o = hashTypeGetValueObject(db, o, fieldobj->ptr, &isHashDeleted); o = hashTypeGetValueObject(db, o, fieldobj->ptr, HFE_LAZY_EXPIRE, &isHashDeleted);
if (isHashDeleted) if (isHashDeleted)
goto noobj; goto noobj;
......
...@@ -14,24 +14,43 @@ ...@@ -14,24 +14,43 @@
* update the expiration time of the hash object in global HFE DS. */ * update the expiration time of the hash object in global HFE DS. */
#define HASH_NEW_EXPIRE_DIFF_THRESHOLD max(4000, 1<<EB_BUCKET_KEY_PRECISION) #define HASH_NEW_EXPIRE_DIFF_THRESHOLD max(4000, 1<<EB_BUCKET_KEY_PRECISION)
/* Returned by hashTypeGetValue() */ /* Reserve 2 bits out of hash-field expiration time for possible future lightweight
* indexing/categorizing of fields. It can be achieved by hacking HFE as follows:
*
* HPEXPIREAT key [ 2^47 + USER_INDEX ] FIELDS numfields field [field …]
*
* Redis will also need to expose kind of HEXPIRESCAN and HEXPIRECOUNT for this
* idea. Yet to be better defined.
*
* HFE_MAX_ABS_TIME_MSEC constraint must be enforced only at API level. Internally,
* the expiration time can be up to EB_EXPIRE_TIME_MAX for future readiness.
*/
#define HFE_MAX_ABS_TIME_MSEC (EB_EXPIRE_TIME_MAX >> 2)
typedef enum GetFieldRes { typedef enum GetFieldRes {
/* common (Used by hashTypeGet* value family) */ /* common (Used by hashTypeGet* value family) */
GETF_OK = 0, GETF_OK = 0, /* The field was found. */
GETF_NOT_FOUND, /* The field was not found. */ GETF_NOT_FOUND, /* The field was not found. */
GETF_EXPIRED, /* Logically expired (Might be lazy deleted or not) */
/* used only by hashTypeGetValue() */
GETF_EXPIRED, /* Logically expired but not yet deleted. */
GETF_EXPIRED_HASH, /* Delete hash since retrieved field was expired and GETF_EXPIRED_HASH, /* Delete hash since retrieved field was expired and
* it was the last field in the hash. */ * it was the last field in the hash. */
} GetFieldRes; } GetFieldRes;
/* ActiveExpireCtx passed to hashTypeActiveExpire() */
typedef struct ExpireCtx {
uint32_t fieldsToExpireQuota;
redisDb *db;
} ExpireCtx;
typedef listpackEntry CommonEntry; /* extend usage beyond lp */
/* hash field expiration (HFE) funcs */ /* hash field expiration (HFE) funcs */
static ExpireAction onFieldExpire(eItem item, void *ctx); static ExpireAction onFieldExpire(eItem item, void *ctx);
static ExpireMeta* hfieldGetExpireMeta(const eItem field); static ExpireMeta* hfieldGetExpireMeta(const eItem field);
static ExpireMeta *hashGetExpireMeta(const eItem hash); static ExpireMeta *hashGetExpireMeta(const eItem hash);
static void hexpireGenericCommand(client *c, const char *cmd, long long basetime, int unit); static void hexpireGenericCommand(client *c, const char *cmd, long long basetime, int unit);
static ExpireAction hashTypeActiveExpire(eItem hashObj, void *ctx); static ExpireAction hashTypeActiveExpire(eItem hashObj, void *ctx);
static uint64_t hashTypeExpire(robj *o, ExpireCtx *expireCtx, int updateGlobalHFE);
static void hfieldPersist(robj *hashObj, hfield field); static void hfieldPersist(robj *hashObj, hfield field);
static void propagateHashFieldDeletion(redisDb *db, sds key, char *field, size_t fieldLen); static void propagateHashFieldDeletion(redisDb *db, sds key, char *field, size_t fieldLen);
...@@ -108,12 +127,6 @@ EbucketsType hashFieldExpireBucketsType = { ...@@ -108,12 +127,6 @@ EbucketsType hashFieldExpireBucketsType = {
.itemsAddrAreOdd = 1, /* Addresses of hfield (mstr) are odd!! */ .itemsAddrAreOdd = 1, /* Addresses of hfield (mstr) are odd!! */
}; };
/* ActiveExpireCtx passed to hashTypeActiveExpire() */
typedef struct ActiveExpireCtx {
uint32_t fieldsToExpireQuota;
redisDb *db;
} ActiveExpireCtx;
/* OnFieldExpireCtx passed to OnFieldExpire() */ /* OnFieldExpireCtx passed to OnFieldExpire() */
typedef struct OnFieldExpireCtx { typedef struct OnFieldExpireCtx {
robj *hashObj; robj *hashObj;
...@@ -152,36 +165,26 @@ static inline int isDictWithMetaHFE(dict *d) { ...@@ -152,36 +165,26 @@ static inline int isDictWithMetaHFE(dict *d) {
} }
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
* setex* - Set field OR field's expiration * setex* - Set field's expiration
* *
* Whereas setting plain fields is rather straightforward, setting expiration * Setting expiration time to fields might be time-consuming and complex since
* time to fields might be time-consuming and complex since each update of * each update of expiration time, not only updates `ebuckets` of corresponding
* expiration time, not only updates `ebuckets` of corresponding hash, but also * hash, but also might update `ebuckets` of global HFE DS. It is required to opt
* might update `ebuckets` of global HFE DS. It is required to opt sequence of * sequence of field updates with expirartion for a given hash, such that only
* field updates with expirartion for a given hash, such that only once done, * once done, the global HFE DS will get updated.
* the global HFE DS will get updated.
* *
* To do so, follow the scheme: * To do so, follow the scheme:
* 1. Call hashTypeSetExInit() to initialize the HashTypeSetEx struct. * 1. Call hashTypeSetExInit() to initialize the HashTypeSetEx struct.
* 2. Call hashTypeSetEx() one time or more, for each field/expiration update. * 2. Call hashTypeSetEx() one time or more, for each field/expiration update.
* 3. Call hashTypeSetExDone() for notification and update of global HFE. * 3. Call hashTypeSetExDone() for notification and update of global HFE.
*
* If expiration is not required, then avoid this API and use instead hashTypeSet()
*----------------------------------------------------------------------------*/ *----------------------------------------------------------------------------*/
/* Returned value of hashTypeSetEx() */ /* Returned value of hashTypeSetEx() */
typedef enum SetExRes { typedef enum SetExRes {
/* Common res from hashTypeSetEx() */
HSETEX_OK = 1, /* Expiration time set/updated as expected */ HSETEX_OK = 1, /* Expiration time set/updated as expected */
/* If provided HashTypeSetEx struct to hashTypeSetEx() */
HSETEX_NO_FIELD = -2, /* No such hash-field */ HSETEX_NO_FIELD = -2, /* No such hash-field */
HSETEX_NO_CONDITION_MET = 0, /* Specified NX | XX | GT | LT condition not met */ HSETEX_NO_CONDITION_MET = 0, /* Specified NX | XX | GT | LT condition not met */
HSETEX_DELETED = 2, /* Field deleted because the specified time is in the past */ HSETEX_DELETED = 2, /* Field deleted because the specified time is in the past */
/* If not provided HashTypeSetEx struct to hashTypeSetEx() (plain HSET) */
HSET_UPDATE = 4, /* Update of the field without expiration time */
} SetExRes; } SetExRes;
/* Used by httlGenericCommand() */ /* Used by httlGenericCommand() */
...@@ -190,20 +193,6 @@ typedef enum GetExpireTimeRes { ...@@ -190,20 +193,6 @@ typedef enum GetExpireTimeRes {
HFE_GET_NO_TTL = -1, /* No TTL attached to the field */ HFE_GET_NO_TTL = -1, /* No TTL attached to the field */
} GetExpireTimeRes; } GetExpireTimeRes;
/* on fail return HSETEX_NO_CONDITION_MET */
typedef enum FieldSetCond {
FIELD_CREATE_OR_OVRWRT = 0,
FIELD_DONT_CREATE = 1,
FIELD_DONT_CREATE2 = 2, /* on fail return HSETEX_NO_FIELD */
FIELD_DONT_OVRWRT = 3
} FieldSetCond;
typedef enum FieldGet { /* TBD */
FIELD_GET_NONE = 0,
FIELD_GET_NEW = 1,
FIELD_GET_OLD = 2
} FieldGet;
typedef enum ExpireSetCond { typedef enum ExpireSetCond {
HFE_NX = 1<<0, HFE_NX = 1<<0,
HFE_XX = 1<<1, HFE_XX = 1<<1,
...@@ -211,16 +200,10 @@ typedef enum ExpireSetCond { ...@@ -211,16 +200,10 @@ typedef enum ExpireSetCond {
HFE_LT = 1<<3 HFE_LT = 1<<3
} ExpireSetCond; } ExpireSetCond;
typedef struct HashTypeSet {
sds value;
int flags;
} HashTypeSet;
/* Used by hashTypeSetEx() for setting fields or their expiry */ /* Used by hashTypeSetEx() for setting fields or their expiry */
typedef struct HashTypeSetEx { typedef struct HashTypeSetEx {
/*** config ***/ /*** config ***/
FieldSetCond fieldSetCond; /* [DCF | DOF] */
ExpireSetCond expireSetCond; /* [XX | NX | GT | LT] */ ExpireSetCond expireSetCond; /* [XX | NX | GT | LT] */
/*** metadata ***/ /*** metadata ***/
...@@ -239,14 +222,10 @@ typedef struct HashTypeSetEx { ...@@ -239,14 +222,10 @@ typedef struct HashTypeSetEx {
const char *cmd; const char *cmd;
} HashTypeSetEx; } HashTypeSetEx;
static SetExRes hashTypeSetExListpack(redisDb *db, robj *o, sds field, HashTypeSet *setParams,
uint64_t expireAt, HashTypeSetEx *exParams);
int hashTypeSetExInit(robj *key, robj *o, client *c, redisDb *db, const char *cmd, int hashTypeSetExInit(robj *key, robj *o, client *c, redisDb *db, const char *cmd,
FieldSetCond fieldSetCond, ExpireSetCond expireSetCond, HashTypeSetEx *ex); ExpireSetCond expireSetCond, HashTypeSetEx *ex);
SetExRes hashTypeSetEx(redisDb *db, robj *o, sds field, HashTypeSet *setKeyVal, SetExRes hashTypeSetEx(robj *o, sds field, uint64_t expireAt, HashTypeSetEx *exInfo);
uint64_t expireAt, HashTypeSetEx *exInfo);
void hashTypeSetExDone(HashTypeSetEx *e); void hashTypeSetExDone(HashTypeSetEx *e);
...@@ -445,7 +424,7 @@ void listpackExExpire(redisDb *db, robj *o, ExpireInfo *info) { ...@@ -445,7 +424,7 @@ void listpackExExpire(redisDb *db, robj *o, ExpireInfo *info) {
if (expired) if (expired)
lpt->lp = lpDeleteRange(lpt->lp, 0, expired * 3); lpt->lp = lpDeleteRange(lpt->lp, 0, expired * 3);
min = hashTypeGetNextTimeToExpire(o); min = hashTypeGetMinExpire(o, 1 /*accurate*/);
info->nextExpireTime = min; info->nextExpireTime = min;
} }
...@@ -561,8 +540,10 @@ SetExRes hashTypeSetExpiryListpack(HashTypeSetEx *ex, sds field, ...@@ -561,8 +540,10 @@ SetExRes hashTypeSetExpiryListpack(HashTypeSetEx *ex, sds field,
ex->minExpireFields = prevExpire; ex->minExpireFields = prevExpire;
} }
/* if expiration time is in the past */ /* If expired, then delete the field and propagate the deletion.
* If replica, continue like the field is valid */
if (unlikely(checkAlreadyExpired(expireAt))) { if (unlikely(checkAlreadyExpired(expireAt))) {
propagateHashFieldDeletion(ex->db, ex->key->ptr, field, sdslen(field));
hashTypeDelete(ex->hashObj, field, 1); hashTypeDelete(ex->hashObj, field, 1);
ex->fieldDeleted++; ex->fieldDeleted++;
return HSETEX_DELETED; return HSETEX_DELETED;
...@@ -712,23 +693,21 @@ GetFieldRes hashTypeGetFromHashTable(robj *o, sds field, sds *value, uint64_t *e ...@@ -712,23 +693,21 @@ GetFieldRes hashTypeGetFromHashTable(robj *o, sds field, sds *value, uint64_t *e
/* Higher level function of hashTypeGet*() that returns the hash value /* Higher level function of hashTypeGet*() that returns the hash value
* associated with the specified field. * associated with the specified field.
* Arguments:
* hfeFlags - Lookup for HFE_LAZY_* flags
* *
* Returned: * Returned:
* - GetFieldRes: OK: Return Field's valid value * GetFieldRes - Result of get operation
* NOT_FOUND: Field was not found. * vstr, vlen - if string, ref in either *vstr and *vlen if it's
* EXPIRED: Field is expired and Lazy deleted
* EXPIRED_HASH: Returned only if the field is the last one in the
* hash and the hash is deleted.
* - vstr, vlen : if string, ref in either *vstr and *vlen if it's
* returned in string form, * returned in string form,
* - vll : or stored in *vll if it's returned as a number. * vll - or stored in *vll if it's returned as a number.
* If *vll is populated *vstr is set to NULL, so the caller can * If *vll is populated *vstr is set to NULL, so the caller can
* always check the function return by checking the return value * always check the function return by checking the return value
* for GETF_OK and checking if vll (or vstr) is NULL. * for GETF_OK and checking if vll (or vstr) is NULL.
* *
*/ */
GetFieldRes hashTypeGetValue(redisDb *db, robj *o, sds field, unsigned char **vstr, GetFieldRes hashTypeGetValue(redisDb *db, robj *o, sds field, unsigned char **vstr,
unsigned int *vlen, long long *vll) { unsigned int *vlen, long long *vll, int hfeFlags) {
uint64_t expiredAt; uint64_t expiredAt;
sds key; sds key;
GetFieldRes res; GetFieldRes res;
...@@ -753,14 +732,24 @@ GetFieldRes hashTypeGetValue(redisDb *db, robj *o, sds field, unsigned char **vs ...@@ -753,14 +732,24 @@ GetFieldRes hashTypeGetValue(redisDb *db, robj *o, sds field, unsigned char **vs
serverPanic("Unknown hash encoding"); serverPanic("Unknown hash encoding");
} }
/* Don't expire anything while loading. It will be done later. */ if (expiredAt >= (uint64_t) commandTimeSnapshot())
if ( (server.loading) ||
(server.lazy_expire_disabled) ||
((server.masterhost) && (server.current_client && (server.current_client->flags & CLIENT_MASTER))) ||
(expiredAt >= (uint64_t) commandTimeSnapshot()) )
return GETF_OK; return GETF_OK;
/* Got expired. Extract attached key from LISTPACK_EX/HT */ if (server.masterhost) {
/* If CLIENT_MASTER, assume valid as long as it didn't get delete */
if (server.current_client && (server.current_client->flags & CLIENT_MASTER))
return GETF_OK;
/* If user client, then act as if expired, but don't delete! */
return GETF_EXPIRED;
}
if ((server.loading) ||
(server.lazy_expire_disabled) ||
(hfeFlags & HFE_LAZY_AVOID_FIELD_DEL) ||
(isPausedActionsWithUpdate(PAUSE_ACTION_EXPIRE)))
return GETF_EXPIRED;
if (o->encoding == OBJ_ENCODING_LISTPACK_EX) if (o->encoding == OBJ_ENCODING_LISTPACK_EX)
key = ((listpackEx *) o->ptr)->key; key = ((listpackEx *) o->ptr)->key;
else else
...@@ -771,15 +760,19 @@ GetFieldRes hashTypeGetValue(redisDb *db, robj *o, sds field, unsigned char **vs ...@@ -771,15 +760,19 @@ GetFieldRes hashTypeGetValue(redisDb *db, robj *o, sds field, unsigned char **vs
propagateHashFieldDeletion(db, key, field, sdslen(field)); propagateHashFieldDeletion(db, key, field, sdslen(field));
/* If the field is the last one in the hash, then the hash will be deleted */ /* If the field is the last one in the hash, then the hash will be deleted */
if (hashTypeLength(o, 0) == 0) { res = GETF_EXPIRED;
robj *keyObj = createStringObject(key, sdslen(key)); robj *keyObj = createStringObject(key, sdslen(key));
notifyKeyspaceEvent(NOTIFY_GENERIC, "del", keyObj, db->id); if (!(hfeFlags & HFE_LAZY_NO_NOTIFICATION))
notifyKeyspaceEvent(NOTIFY_HASH, "hexpired", keyObj, db->id);
if ((hashTypeLength(o, 0) == 0) && (!(hfeFlags & HFE_LAZY_AVOID_HASH_DEL))) {
if (!(hfeFlags & HFE_LAZY_NO_NOTIFICATION))
notifyKeyspaceEvent(NOTIFY_GENERIC, "del", keyObj, db->id);
dbDelete(db,keyObj); dbDelete(db,keyObj);
decrRefCount(keyObj); res = GETF_EXPIRED_HASH;
return GETF_EXPIRED_HASH;
} }
signalModifiedKey(NULL, db, keyObj);
return GETF_EXPIRED; decrRefCount(keyObj);
return res;
} }
/* Like hashTypeGetValue() but returns a Redis object, which is useful for /* Like hashTypeGetValue() but returns a Redis object, which is useful for
...@@ -787,24 +780,25 @@ GetFieldRes hashTypeGetValue(redisDb *db, robj *o, sds field, unsigned char **vs ...@@ -787,24 +780,25 @@ GetFieldRes hashTypeGetValue(redisDb *db, robj *o, sds field, unsigned char **vs
* The function returns NULL if the field is not found in the hash. Otherwise * The function returns NULL if the field is not found in the hash. Otherwise
* a newly allocated string object with the value is returned. * a newly allocated string object with the value is returned.
* *
* hfeFlags - Lookup HFE_LAZY_* flags
* isHashDeleted - If attempted to access expired field and it's the last field * isHashDeleted - If attempted to access expired field and it's the last field
* in the hash, then the hash will as well be deleted. In this case, * in the hash, then the hash will as well be deleted. In this case,
* isHashDeleted will be set to 1. * isHashDeleted will be set to 1.
*/ */
robj *hashTypeGetValueObject(redisDb *db, robj *o, sds field, int *isHashDeleted) { robj *hashTypeGetValueObject(redisDb *db, robj *o, sds field, int hfeFlags, int *isHashDeleted) {
unsigned char *vstr; unsigned char *vstr;
unsigned int vlen; unsigned int vlen;
long long vll; long long vll;
*isHashDeleted = 0; /*default*/ if (isHashDeleted) *isHashDeleted = 0;
GetFieldRes res = hashTypeGetValue(db,o,field,&vstr,&vlen,&vll); GetFieldRes res = hashTypeGetValue(db,o,field,&vstr,&vlen,&vll, hfeFlags);
if (res == GETF_OK) { if (res == GETF_OK) {
if (vstr) return createStringObject((char*)vstr,vlen); if (vstr) return createStringObject((char*)vstr,vlen);
else return createStringObjectFromLongLong(vll); else return createStringObjectFromLongLong(vll);
} }
if (res == GETF_EXPIRED_HASH) if ((res == GETF_EXPIRED_HASH) && (isHashDeleted))
*isHashDeleted = 1; *isHashDeleted = 1;
/* GETF_EXPIRED_HASH, GETF_EXPIRED, GETF_NOT_FOUND */ /* GETF_EXPIRED_HASH, GETF_EXPIRED, GETF_NOT_FOUND */
...@@ -814,19 +808,21 @@ robj *hashTypeGetValueObject(redisDb *db, robj *o, sds field, int *isHashDeleted ...@@ -814,19 +808,21 @@ robj *hashTypeGetValueObject(redisDb *db, robj *o, sds field, int *isHashDeleted
/* Test if the specified field exists in the given hash. If the field is /* Test if the specified field exists in the given hash. If the field is
* expired (HFE), then it will be lazy deleted * expired (HFE), then it will be lazy deleted
* *
* Returns 1 if the field exists, and 0 when it doesn't. * hfeFlags - Lookup HFE_LAZY_* flags
*
* isHashDeleted - If attempted to access expired field and it is the last field * isHashDeleted - If attempted to access expired field and it is the last field
* in the hash, then the hash will as well be deleted. In this case, * in the hash, then the hash will as well be deleted. In this case,
* isHashDeleted will be set to 1. * isHashDeleted will be set to 1.
*
* Returns 1 if the field exists, and 0 when it doesn't.
*/ */
int hashTypeExists(redisDb *db, robj *o, sds field, int *isHashDeleted) { int hashTypeExists(redisDb *db, robj *o, sds field, int hfeFlags, int *isHashDeleted) {
unsigned char *vstr = NULL; unsigned char *vstr = NULL;
unsigned int vlen = UINT_MAX; unsigned int vlen = UINT_MAX;
long long vll = LLONG_MAX; long long vll = LLONG_MAX;
GetFieldRes res = hashTypeGetValue(db, o, field, &vstr, &vlen, &vll); GetFieldRes res = hashTypeGetValue(db, o, field, &vstr, &vlen, &vll, hfeFlags);
*isHashDeleted = (res == GETF_EXPIRED_HASH) ? 1 : 0; if (isHashDeleted)
*isHashDeleted = (res == GETF_EXPIRED_HASH) ? 1 : 0;
return (res == GETF_OK) ? 1 : 0; return (res == GETF_OK) ? 1 : 0;
} }
...@@ -839,7 +835,7 @@ int hashTypeExists(redisDb *db, robj *o, sds field, int *isHashDeleted) { ...@@ -839,7 +835,7 @@ int hashTypeExists(redisDb *db, robj *o, sds field, int *isHashDeleted) {
* *
* HASH_SET_TAKE_FIELD -- The SDS field ownership passes to the function. * HASH_SET_TAKE_FIELD -- The SDS field ownership passes to the function.
* HASH_SET_TAKE_VALUE -- The SDS value ownership passes to the function. * HASH_SET_TAKE_VALUE -- The SDS value ownership passes to the function.
* HASH_SET_KEEP_FIELD -- keep original field along with TTL if already exists * HASH_SET_KEEP_TTL -- keep original TTL if field already exists
* *
* When the flags are used the caller does not need to release the passed * When the flags are used the caller does not need to release the passed
* SDS string(s). It's up to the function to use the string to create a new * SDS string(s). It's up to the function to use the string to create a new
...@@ -851,204 +847,254 @@ int hashTypeExists(redisDb *db, robj *o, sds field, int *isHashDeleted) { ...@@ -851,204 +847,254 @@ int hashTypeExists(redisDb *db, robj *o, sds field, int *isHashDeleted) {
*/ */
#define HASH_SET_TAKE_FIELD (1<<0) #define HASH_SET_TAKE_FIELD (1<<0)
#define HASH_SET_TAKE_VALUE (1<<1) #define HASH_SET_TAKE_VALUE (1<<1)
#define HASH_SET_KEEP_FIELD (1<<2) #define HASH_SET_KEEP_TTL (1<<2)
#define HASH_SET_COPY 0 #define HASH_SET_COPY 0
int hashTypeSet(redisDb *db, robj *o, sds field, sds value, int flags) { int hashTypeSet(redisDb *db, robj *o, sds field, sds value, int flags) {
HashTypeSet set = {value, flags}; int update = 0;
return (hashTypeSetEx(db, o, field, &set, 0, NULL) == HSET_UPDATE) ? 1 : 0;
}
SetExRes hashTypeSetExpiry(HashTypeSetEx *ex, sds field, uint64_t expireAt, dictEntry **de) { /* Check if the field is too long for listpack, and convert before adding the item.
dict *ht = ex->hashObj->ptr; * This is needed for HINCRBY* case since in other commands this is handled early by
dictEntry *newEntry = NULL, *existingEntry = NULL; * hashTypeTryConversion, so this check will be a NOP. */
if (o->encoding == OBJ_ENCODING_LISTPACK ||
o->encoding == OBJ_ENCODING_LISTPACK_EX) {
if (sdslen(field) > server.hash_max_listpack_value || sdslen(value) > server.hash_max_listpack_value)
hashTypeConvert(o, OBJ_ENCODING_HT, &db->hexpires);
}
/* New field with expiration metadata */ if (o->encoding == OBJ_ENCODING_LISTPACK) {
hfield hfNew = hfieldNew(field, sdslen(field), 1 /*withExpireMeta*/); unsigned char *zl, *fptr, *vptr;
if ((ex->fieldSetCond == FIELD_DONT_CREATE) || (ex->fieldSetCond == FIELD_DONT_CREATE2)) { zl = o->ptr;
if ((existingEntry = dictFind(ht, field)) == NULL) { fptr = lpFirst(zl);
hfieldFree(hfNew); if (fptr != NULL) {
return (ex->fieldSetCond == FIELD_DONT_CREATE) ? fptr = lpFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);
HSETEX_NO_CONDITION_MET : HSETEX_NO_FIELD; if (fptr != NULL) {
/* Grab pointer to the value (fptr points to the field) */
vptr = lpNext(zl, fptr);
serverAssert(vptr != NULL);
/* Replace value */
zl = lpReplace(zl, &vptr, (unsigned char*)value, sdslen(value));
update = 1;
}
} }
} else {
if (!update) {
/* Push new field/value pair onto the tail of the listpack */
zl = lpAppend(zl, (unsigned char*)field, sdslen(field));
zl = lpAppend(zl, (unsigned char*)value, sdslen(value));
}
o->ptr = zl;
/* Check if the listpack needs to be converted to a hash table */
if (hashTypeLength(o, 0) > server.hash_max_listpack_entries)
hashTypeConvert(o, OBJ_ENCODING_HT, &db->hexpires);
} else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) {
unsigned char *fptr = NULL, *vptr = NULL, *tptr = NULL;
listpackEx *lpt = o->ptr;
long long expireTime = HASH_LP_NO_TTL;
fptr = lpFirst(lpt->lp);
if (fptr != NULL) {
fptr = lpFind(lpt->lp, fptr, (unsigned char*)field, sdslen(field), 2);
if (fptr != NULL) {
/* Grab pointer to the value (fptr points to the field) */
vptr = lpNext(lpt->lp, fptr);
serverAssert(vptr != NULL);
/* Replace value */
lpt->lp = lpReplace(lpt->lp, &vptr, (unsigned char *) value, sdslen(value));
update = 1;
fptr = lpPrev(lpt->lp, vptr);
serverAssert(fptr != NULL);
tptr = lpNext(lpt->lp, vptr);
serverAssert(tptr && lpGetIntegerValue(tptr, &expireTime));
if (flags & HASH_SET_KEEP_TTL) {
/* keep old field along with TTL */
} else if (expireTime != HASH_LP_NO_TTL) {
/* re-insert field and override TTL */
listpackExUpdateExpiry(o, field, fptr, vptr, HASH_LP_NO_TTL);
}
}
}
if (!update)
listpackExAddNew(o, field, sdslen(field), value, sdslen(value),
HASH_LP_NO_TTL);
/* Check if the listpack needs to be converted to a hash table */
if (hashTypeLength(o, 0) > server.hash_max_listpack_entries)
hashTypeConvert(o, OBJ_ENCODING_HT, &db->hexpires);
} else if (o->encoding == OBJ_ENCODING_HT) {
hfield newField = hfieldNew(field, sdslen(field), 0);
dict *ht = o->ptr;
dictEntry *de, *existing;
/* stored key is different than lookup key */
dictUseStoredKeyApi(ht, 1); dictUseStoredKeyApi(ht, 1);
newEntry = dictAddRaw(ht, hfNew, &existingEntry); de = dictAddRaw(ht, newField, &existing);
dictUseStoredKeyApi(ht, 0); dictUseStoredKeyApi(ht, 0);
}
if (newEntry) {
*de = newEntry;
if (ex->expireSetCond & (HFE_XX | HFE_LT | HFE_GT)) { /* If field already exists, then update "field". "Value" will be set afterward */
dictDelete(ht, field); if (de == NULL) {
return HSETEX_NO_CONDITION_MET; if (flags & HASH_SET_KEEP_TTL) {
/* keep old field along with TTL */
hfieldFree(newField);
} else {
/* If attached TTL to the old field, then remove it from hash's private ebuckets */
hfield oldField = dictGetKey(existing);
hfieldPersist(o, oldField);
hfieldFree(oldField);
dictSetKey(ht, existing, newField);
}
sdsfree(dictGetVal(existing));
update = 1;
de = existing;
} }
} else { /* field exist */
*de = existingEntry;
if (ex->fieldSetCond == FIELD_DONT_OVRWRT) { if (flags & HASH_SET_TAKE_VALUE) {
hfieldFree(hfNew); dictSetVal(ht, de, value);
return HSETEX_NO_CONDITION_MET; flags &= ~HASH_SET_TAKE_VALUE;
} else {
dictSetVal(ht, de, sdsdup(value));
} }
} else {
serverPanic("Unknown hash encoding");
}
hfield hfOld = dictGetKey(existingEntry); /* Free SDS strings we did not referenced elsewhere if the flags
* want this function to be responsible. */
if (flags & HASH_SET_TAKE_FIELD && field) sdsfree(field);
if (flags & HASH_SET_TAKE_VALUE && value) sdsfree(value);
return update;
}
/* If field doesn't have expiry metadata attached */ SetExRes hashTypeSetExpiryHT(HashTypeSetEx *exInfo, sds field, uint64_t expireAt) {
if (!hfieldIsExpireAttached(hfOld)) { dict *ht = exInfo->hashObj->ptr;
dictEntry *existingEntry = NULL;
/* For fields without expiry, LT condition is considered valid */ /* New field with expiration metadata */
if (ex->expireSetCond & (HFE_XX | HFE_GT)) { hfield hfNew = hfieldNew(field, sdslen(field), 1 /*withExpireMeta*/);
hfieldFree(hfNew);
return HSETEX_NO_CONDITION_MET;
}
/* Delete old field. Below goanna dictSetKey(..,hfNew) */ if ((existingEntry = dictFind(ht, field)) == NULL) {
hfieldFree(hfOld); hfieldFree(hfNew);
return HSETEX_NO_FIELD;
}
hfield hfOld = dictGetKey(existingEntry);
} else { /* field has ExpireMeta struct attached */ /* If field doesn't have expiry metadata attached */
if (!hfieldIsExpireAttached(hfOld)) {
/* No need for hfNew (Just modify expire-time of existing field) */ /* For fields without expiry, LT condition is considered valid */
if (exInfo->expireSetCond & (HFE_XX | HFE_GT)) {
hfieldFree(hfNew); hfieldFree(hfNew);
return HSETEX_NO_CONDITION_MET;
}
uint64_t prevExpire = hfieldGetExpireTime(hfOld); /* Delete old field. Below goanna dictSetKey(..,hfNew) */
hfieldFree(hfOld);
/* If field has valid expiration time, then check GT|LT|NX */ } else { /* field has ExpireMeta struct attached */
if (prevExpire != EB_EXPIRE_TIME_INVALID) {
if (((ex->expireSetCond == HFE_GT) && (prevExpire >= expireAt)) ||
((ex->expireSetCond == HFE_LT) && (prevExpire <= expireAt)) ||
(ex->expireSetCond == HFE_NX) )
return HSETEX_NO_CONDITION_MET;
/* remove old expiry time from hash's private ebuckets */ /* No need for hfNew (Just modify expire-time of existing field) */
dictExpireMetadata *dm = (dictExpireMetadata *) dictMetadata(ht); hfieldFree(hfNew);
ebRemove(&dm->hfe, &hashFieldExpireBucketsType, hfOld);
/* Track of minimum expiration time (only later update global HFE DS) */ uint64_t prevExpire = hfieldGetExpireTime(hfOld);
if (ex->minExpireFields > prevExpire)
ex->minExpireFields = prevExpire;
} else { /* If field has valid expiration time, then check GT|LT|NX */
/* field has invalid expiry. No need to ebRemove() */ if (prevExpire != EB_EXPIRE_TIME_INVALID) {
if (((exInfo->expireSetCond == HFE_GT) && (prevExpire >= expireAt)) ||
((exInfo->expireSetCond == HFE_LT) && (prevExpire <= expireAt)) ||
(exInfo->expireSetCond == HFE_NX) )
return HSETEX_NO_CONDITION_MET;
/* Check XX|LT|GT */ /* remove old expiry time from hash's private ebuckets */
if (ex->expireSetCond & (HFE_XX | HFE_GT)) dictExpireMetadata *dm = (dictExpireMetadata *) dictMetadata(ht);
return HSETEX_NO_CONDITION_MET; ebRemove(&dm->hfe, &hashFieldExpireBucketsType, hfOld);
}
/* Track of minimum expiration time (only later update global HFE DS) */
if (exInfo->minExpireFields > prevExpire)
exInfo->minExpireFields = prevExpire;
/* Reuse hfOld as hfNew and rewrite its expiry with ebAdd() */ } else {
hfNew = hfOld; /* field has invalid expiry. No need to ebRemove() */
/* Check XX|LT|GT */
if (exInfo->expireSetCond & (HFE_XX | HFE_GT))
return HSETEX_NO_CONDITION_MET;
} }
dictSetKey(ht, existingEntry, hfNew); /* Reuse hfOld as hfNew and rewrite its expiry with ebAdd() */
hfNew = hfOld;
} }
/* if expiration time is in the past */ dictSetKey(ht, existingEntry, hfNew);
/* If expired, then delete the field and propagate the deletion.
* If replica, continue like the field is valid */
if (unlikely(checkAlreadyExpired(expireAt))) { if (unlikely(checkAlreadyExpired(expireAt))) {
hashTypeDelete(ex->hashObj, field, 1); /* replicas should not initiate deletion of fields */
ex->fieldDeleted++; propagateHashFieldDeletion(exInfo->db, exInfo->key->ptr, field, sdslen(field));
hashTypeDelete(exInfo->hashObj, field, 1);
exInfo->fieldDeleted++;
return HSETEX_DELETED; return HSETEX_DELETED;
} }
if (ex->minExpireFields > expireAt) if (exInfo->minExpireFields > expireAt)
ex->minExpireFields = expireAt; exInfo->minExpireFields = expireAt;
dictExpireMetadata *dm = (dictExpireMetadata *) dictMetadata(ht); dictExpireMetadata *dm = (dictExpireMetadata *) dictMetadata(ht);
ebAdd(&dm->hfe, &hashFieldExpireBucketsType, hfNew, expireAt); ebAdd(&dm->hfe, &hashFieldExpireBucketsType, hfNew, expireAt);
ex->fieldUpdated++; exInfo->fieldUpdated++;
return HSETEX_OK; return HSETEX_OK;
} }
/* /*
* Set fields OR field's expiration (See also `setex*` comment above) * Set field expiration
* *
* Take care to call first hashTypeSetExInit() and then call this function. * Take care to call first hashTypeSetExInit() and then call this function.
* Finally, call hashTypeSetExDone() to notify and update global HFE DS. * Finally, call hashTypeSetExDone() to notify and update global HFE DS.
*
* NOTE: this functions is also called during RDB load to set dict-encoded
* fields with and without expiration.
*/ */
SetExRes hashTypeSetEx(redisDb *db, robj *o, sds field, HashTypeSet *setKeyVal, SetExRes hashTypeSetEx(robj *o, sds field, uint64_t expireAt, HashTypeSetEx *exInfo)
uint64_t expireAt, HashTypeSetEx *exInfo)
{ {
SetExRes res = HSETEX_OK; if (o->encoding == OBJ_ENCODING_LISTPACK_EX)
int isSetKeyValue = (setKeyVal) ? 1 : 0;
int isSetExpire = (exInfo) ? 1 : 0;
int flags = (setKeyVal) ? setKeyVal->flags : 0;
/* Check if the field is too long for listpack, and convert before adding the item.
* This is needed for HINCRBY* case since in other commands this is handled early by
* hashTypeTryConversion, so this check will be a NOP. */
if (o->encoding == OBJ_ENCODING_LISTPACK ||
o->encoding == OBJ_ENCODING_LISTPACK_EX)
{ {
if ( (isSetKeyValue) && unsigned char *fptr = NULL, *vptr = NULL, *tptr = NULL;
(sdslen(field) > server.hash_max_listpack_value ||
sdslen(setKeyVal->value) > server.hash_max_listpack_value) )
{
hashTypeConvert(o, OBJ_ENCODING_HT, &db->hexpires);
} else {
res = hashTypeSetExListpack(db, o, field, setKeyVal, expireAt, exInfo);
goto SetExDone; /*done*/
}
}
if (o->encoding != OBJ_ENCODING_HT) listpackEx *lpt = o->ptr;
serverPanic("Unknown hash encoding"); long long expireTime = HASH_LP_NO_TTL;
/*** now deal with HT ***/ if ((fptr = lpFirst(lpt->lp)) == NULL)
hfield newField; return HSETEX_NO_FIELD;
dict *ht = o->ptr;
dictEntry *de;
/* If needed to set the field along with expiry */ fptr = lpFind(lpt->lp, fptr, (unsigned char*)field, sdslen(field), 2);
if (isSetExpire) {
res = hashTypeSetExpiry(exInfo, field, expireAt, &de);
if (res != HSETEX_OK) goto SetExDone;
} else {
dictEntry *existing;
/* Cannot leverage HASH_SET_TAKE_FIELD since hfield is not of type sds */
newField = hfieldNew(field, sdslen(field), 0);
/* stored key is different than lookup key */ if (!fptr)
dictUseStoredKeyApi(ht, 1); return HSETEX_NO_FIELD;
de = dictAddRaw(ht, newField, &existing);
dictUseStoredKeyApi(ht, 0);
/* If field already exists, then update "field". "Value" will be set afterward */ /* Grab pointer to the value (fptr points to the field) */
if (de == NULL) { vptr = lpNext(lpt->lp, fptr);
if (flags & HASH_SET_KEEP_FIELD) { serverAssert(vptr != NULL);
/* Not keep old field along with TTL */
hfieldFree(newField);
} else {
/* If attached TTL to the old field, then remove it from hash's private ebuckets */
hfield oldField = dictGetKey(existing);
hfieldPersist(o, oldField);
hfieldFree(oldField);
dictSetKey(ht, existing, newField);
}
sdsfree(dictGetVal(existing));
res = HSET_UPDATE;
de = existing;
}
}
/* If need to set value */ tptr = lpNext(lpt->lp, vptr);
if (isSetKeyValue) { serverAssert(tptr && lpGetIntegerValue(tptr, &expireTime));
if (flags & HASH_SET_TAKE_VALUE) {
dictSetVal(ht, de, setKeyVal->value); /* update TTL */
flags &= ~HASH_SET_TAKE_VALUE; return hashTypeSetExpiryListpack(exInfo, field, fptr, vptr, tptr, expireAt);
} else { } else if (o->encoding == OBJ_ENCODING_HT) {
dictSetVal(ht, de, sdsdup(setKeyVal->value)); /* If needed to set the field along with expiry */
} return hashTypeSetExpiryHT(exInfo, field, expireAt);
} else {
serverPanic("Unknown hash encoding");
} }
SetExDone: return HSETEX_OK; /* never reach here */
/* Free SDS strings we did not referenced elsewhere if the flags
* want this function to be responsible. */
if (flags & HASH_SET_TAKE_FIELD && field) sdsfree(field);
if (flags & HASH_SET_TAKE_VALUE && setKeyVal->value) sdsfree(setKeyVal->value);
return res;
} }
void initDictExpireMetadata(sds key, robj *o) { void initDictExpireMetadata(sds key, robj *o) {
...@@ -1060,18 +1106,11 @@ void initDictExpireMetadata(sds key, robj *o) { ...@@ -1060,18 +1106,11 @@ void initDictExpireMetadata(sds key, robj *o) {
m->expireMeta.trash = 1; /* mark as trash (as long it wasn't ebAdd()) */ m->expireMeta.trash = 1; /* mark as trash (as long it wasn't ebAdd()) */
} }
/* /* Init HashTypeSetEx struct before calling hashTypeSetEx() */
* Init HashTypeSetEx struct before calling hashTypeSetEx() int hashTypeSetExInit(robj *key, robj *o, client *c, redisDb *db, const char *cmd,
*
* Don't have to provide client and "cmd". If provided, then notification once
* done by function hashTypeSetExDone().
*/
int hashTypeSetExInit(robj *key, robj *o, client *c, redisDb *db, const char *cmd, FieldSetCond fieldSetCond,
ExpireSetCond expireSetCond, HashTypeSetEx *ex) ExpireSetCond expireSetCond, HashTypeSetEx *ex)
{ {
dict *ht = o->ptr; dict *ht = o->ptr;
ex->fieldSetCond = fieldSetCond;
ex->expireSetCond = expireSetCond; ex->expireSetCond = expireSetCond;
ex->minExpire = EB_EXPIRE_TIME_INVALID; ex->minExpire = EB_EXPIRE_TIME_INVALID;
ex->c = c; ex->c = c;
...@@ -1084,20 +1123,20 @@ int hashTypeSetExInit(robj *key, robj *o, client *c, redisDb *db, const char *cm ...@@ -1084,20 +1123,20 @@ int hashTypeSetExInit(robj *key, robj *o, client *c, redisDb *db, const char *cm
ex->minExpireFields = EB_EXPIRE_TIME_INVALID; ex->minExpireFields = EB_EXPIRE_TIME_INVALID;
/* Take care that HASH support expiration */ /* Take care that HASH support expiration */
if (ex->hashObj->encoding == OBJ_ENCODING_LISTPACK) { if (o->encoding == OBJ_ENCODING_LISTPACK) {
hashTypeConvert(ex->hashObj, OBJ_ENCODING_LISTPACK_EX, &c->db->hexpires); hashTypeConvert(o, OBJ_ENCODING_LISTPACK_EX, &c->db->hexpires);
listpackEx *lpt = ex->hashObj->ptr; listpackEx *lpt = o->ptr;
dictEntry *de = dbFind(c->db, key->ptr); dictEntry *de = dbFind(c->db, key->ptr);
serverAssert(de != NULL); serverAssert(de != NULL);
lpt->key = dictGetKey(de); lpt->key = dictGetKey(de);
} else if (ex->hashObj->encoding == OBJ_ENCODING_HT) { } else if (o->encoding == OBJ_ENCODING_HT) {
/* Take care dict has HFE metadata */ /* Take care dict has HFE metadata */
if (!isDictWithMetaHFE(ht)) { if (!isDictWithMetaHFE(ht)) {
/* Realloc (only header of dict) with metadata for hash-field expiration */ /* Realloc (only header of dict) with metadata for hash-field expiration */
dictTypeAddMeta(&ht, &mstrHashDictTypeWithHFE); dictTypeAddMeta(&ht, &mstrHashDictTypeWithHFE);
dictExpireMetadata *m = (dictExpireMetadata *) dictMetadata(ht); dictExpireMetadata *m = (dictExpireMetadata *) dictMetadata(ht);
ex->hashObj->ptr = ht; o->ptr = ht;
/* Find the key in the keyspace. Need to keep reference to the key for /* Find the key in the keyspace. Need to keep reference to the key for
* notifications or even removal of the hash */ * notifications or even removal of the hash */
...@@ -1111,7 +1150,8 @@ int hashTypeSetExInit(robj *key, robj *o, client *c, redisDb *db, const char *cm ...@@ -1111,7 +1150,8 @@ int hashTypeSetExInit(robj *key, robj *o, client *c, redisDb *db, const char *cm
} }
} }
ex->minExpire = hashTypeGetMinExpire(ex->hashObj); /* Read minExpire from attached ExpireMeta to the hash */
ex->minExpire = hashTypeGetMinExpire(o, 0);
return C_OK; return C_OK;
} }
...@@ -1123,15 +1163,17 @@ void hashTypeSetExDone(HashTypeSetEx *ex) { ...@@ -1123,15 +1163,17 @@ void hashTypeSetExDone(HashTypeSetEx *ex) {
/* Notify keyspace event, update dirty count and update global HFE DS */ /* Notify keyspace event, update dirty count and update global HFE DS */
if (ex->fieldDeleted + ex->fieldUpdated > 0) { if (ex->fieldDeleted + ex->fieldUpdated > 0) {
if (ex->c) { server.dirty += ex->fieldDeleted + ex->fieldUpdated;
server.dirty += ex->fieldDeleted + ex->fieldUpdated;
signalModifiedKey(ex->c, ex->db, ex->key);
notifyKeyspaceEvent(NOTIFY_HASH, "hexpire", ex->key, ex->db->id);
}
if (ex->fieldDeleted && hashTypeLength(ex->hashObj, 0) == 0) { if (ex->fieldDeleted && hashTypeLength(ex->hashObj, 0) == 0) {
dbDelete(ex->db,ex->key); dbDelete(ex->db,ex->key);
if (ex->c) notifyKeyspaceEvent(NOTIFY_GENERIC,"del",ex->key, ex->db->id); signalModifiedKey(ex->c, ex->db, ex->key);
notifyKeyspaceEvent(NOTIFY_HASH, "hexpired", ex->key, ex->db->id);
notifyKeyspaceEvent(NOTIFY_GENERIC,"del",ex->key, ex->db->id);
} else { } else {
signalModifiedKey(ex->c, ex->db, ex->key);
notifyKeyspaceEvent(NOTIFY_HASH, ex->fieldDeleted ? "hexpired" : "hexpire",
ex->key, ex->db->id);
/* If minimum HFE of the hash is smaller than expiration time of the /* If minimum HFE of the hash is smaller than expiration time of the
* specified fields in the command as well as it is smaller or equal * specified fields in the command as well as it is smaller or equal
* than expiration time provided in the command, then the minimum * than expiration time provided in the command, then the minimum
...@@ -1139,8 +1181,8 @@ void hashTypeSetExDone(HashTypeSetEx *ex) { ...@@ -1139,8 +1181,8 @@ void hashTypeSetExDone(HashTypeSetEx *ex) {
if ((ex->minExpire < ex->minExpireFields)) if ((ex->minExpire < ex->minExpireFields))
return; return;
/* retrieve new expired time. It might have changed. */ /* Retrieve new expired time. It might have changed. */
uint64_t newMinExpire = hashTypeGetNextTimeToExpire(ex->hashObj); uint64_t newMinExpire = hashTypeGetMinExpire(ex->hashObj, 1 /*accurate*/);
/* Calculate the diff between old minExpire and newMinExpire. If it is /* Calculate the diff between old minExpire and newMinExpire. If it is
* only few seconds, then don't have to update global HFE DS. At the worst * only few seconds, then don't have to update global HFE DS. At the worst
...@@ -1161,99 +1203,6 @@ void hashTypeSetExDone(HashTypeSetEx *ex) { ...@@ -1161,99 +1203,6 @@ void hashTypeSetExDone(HashTypeSetEx *ex) {
} }
} }
/* Check if the field is too long for listpack, and convert before adding the item.
* This is needed for HINCRBY* case since in other commands this is handled early by
* hashTypeTryConversion, so this check will be a NOP. */
static SetExRes hashTypeSetExListpack(redisDb *db, robj *o, sds field, HashTypeSet *setParams,
uint64_t expireAt, HashTypeSetEx *exParams)
{
int res = HSETEX_OK;
unsigned char *fptr = NULL, *vptr = NULL, *tptr = NULL;
if (o->encoding == OBJ_ENCODING_LISTPACK) {
/* If reached here, then no need to set expiration. Otherwise, as precond
* listpack is converted to listpackex by hashTypeSetExInit() */
unsigned char *zl = o->ptr;
fptr = lpFirst(zl);
if (fptr != NULL) {
fptr = lpFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);
if (fptr != NULL) {
/* Grab pointer to the value (fptr points to the field) */
vptr = lpNext(zl, fptr);
serverAssert(vptr != NULL);
res = HSET_UPDATE;
/* Replace value */
zl = lpReplace(zl, &vptr, (unsigned char *) setParams->value, sdslen(setParams->value));
}
}
if (res != HSET_UPDATE) {
/* Push new field/value pair onto the tail of the listpack */
zl = lpAppend(zl, (unsigned char*)field, sdslen(field));
zl = lpAppend(zl, (unsigned char*)setParams->value, sdslen(setParams->value));
}
o->ptr = zl;
goto out;
} else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) {
listpackEx *lpt = o->ptr;
long long expireTime = HASH_LP_NO_TTL;
fptr = lpFirst(lpt->lp);
if (fptr != NULL) {
fptr = lpFind(lpt->lp, fptr, (unsigned char*)field, sdslen(field), 2);
if (fptr != NULL) {
/* Grab pointer to the value (fptr points to the field) */
vptr = lpNext(lpt->lp, fptr);
serverAssert(vptr != NULL);
if (setParams) {
/* Replace value */
lpt->lp = lpReplace(lpt->lp, &vptr,
(unsigned char *) setParams->value,
sdslen(setParams->value));
fptr = lpPrev(lpt->lp, vptr);
serverAssert(fptr != NULL);
res = HSET_UPDATE;
}
tptr = lpNext(lpt->lp, vptr);
serverAssert(tptr && lpGetIntegerValue(tptr, &expireTime));
/* Keep, update or clear TTL */
if (setParams && setParams->flags & HASH_SET_KEEP_FIELD) {
/* keep old field along with TTL */
} else if (exParams) {
res = hashTypeSetExpiryListpack(exParams, field, fptr, vptr, tptr,
expireAt);
if (res != HSETEX_OK)
goto out;
} else if (res == HSET_UPDATE && expireTime != HASH_LP_NO_TTL) {
/* Clear TTL */
listpackExUpdateExpiry(o, field, fptr, vptr, HASH_LP_NO_TTL);
}
}
}
if (!fptr) {
if (setParams) {
listpackExAddNew(o, field, sdslen(field),
setParams->value, sdslen(setParams->value),
exParams ? expireAt : HASH_LP_NO_TTL);
} else {
res = HSETEX_NO_FIELD;
}
}
}
out:
/* Check if the listpack needs to be converted to a hash table */
if (hashTypeLength(o, 0) > server.hash_max_listpack_entries)
hashTypeConvert(o, OBJ_ENCODING_HT, &db->hexpires);
return res;
}
/* Delete an element from a hash. /* Delete an element from a hash.
* *
* Return 1 on deleted and 0 on not found. * Return 1 on deleted and 0 on not found.
...@@ -1640,7 +1589,7 @@ void hashTypeConvertListpackEx(robj *o, int enc, ebuckets *hexpires) { ...@@ -1640,7 +1589,7 @@ void hashTypeConvertListpackEx(robj *o, int enc, ebuckets *hexpires) {
dict *dict; dict *dict;
dictExpireMetadata *dictExpireMeta; dictExpireMetadata *dictExpireMeta;
listpackEx *lpt = o->ptr; listpackEx *lpt = o->ptr;
uint64_t minExpire = hashTypeGetMinExpire(o); uint64_t minExpire = hashTypeGetMinExpire(o, 0);
if (hexpires && lpt->meta.trash != 1) if (hexpires && lpt->meta.trash != 1)
ebRemove(hexpires, &hashExpireBucketsType, o); ebRemove(hexpires, &hashExpireBucketsType, o);
...@@ -1805,7 +1754,7 @@ void hashReplyFromListpackEntry(client *c, listpackEntry *e) { ...@@ -1805,7 +1754,7 @@ void hashReplyFromListpackEntry(client *c, listpackEntry *e) {
* 'key' and 'val' will be set to hold the element. * 'key' and 'val' will be set to hold the element.
* The memory in them is not to be freed or modified by the caller. * The memory in them is not to be freed or modified by the caller.
* 'val' can be NULL in which case it's not extracted. */ * 'val' can be NULL in which case it's not extracted. */
void hashTypeRandomElement(robj *hashobj, unsigned long hashsize, listpackEntry *key, listpackEntry *val) { void hashTypeRandomElement(robj *hashobj, unsigned long hashsize, CommonEntry *key, CommonEntry *val) {
if (hashobj->encoding == OBJ_ENCODING_HT) { if (hashobj->encoding == OBJ_ENCODING_HT) {
dictEntry *de = dictGetFairRandomKey(hashobj->ptr); dictEntry *de = dictGetFairRandomKey(hashobj->ptr);
hfield field = dictGetKey(de); hfield field = dictGetKey(de);
...@@ -1817,9 +1766,10 @@ void hashTypeRandomElement(robj *hashobj, unsigned long hashsize, listpackEntry ...@@ -1817,9 +1766,10 @@ void hashTypeRandomElement(robj *hashobj, unsigned long hashsize, listpackEntry
val->slen = sdslen(s); val->slen = sdslen(s);
} }
} else if (hashobj->encoding == OBJ_ENCODING_LISTPACK) { } else if (hashobj->encoding == OBJ_ENCODING_LISTPACK) {
lpRandomPair(hashobj->ptr, hashsize, key, val, 2); lpRandomPair(hashobj->ptr, hashsize, (listpackEntry *) key, (listpackEntry *) val, 2);
} else if (hashobj->encoding == OBJ_ENCODING_LISTPACK_EX) { } else if (hashobj->encoding == OBJ_ENCODING_LISTPACK_EX) {
lpRandomPair(hashTypeListpackGetLp(hashobj), hashsize, key, val, 3); lpRandomPair(hashTypeListpackGetLp(hashobj), hashsize, (listpackEntry *) key,
(listpackEntry *) val, 3);
} else { } else {
serverPanic("Unknown hash encoding"); serverPanic("Unknown hash encoding");
} }
...@@ -1840,38 +1790,61 @@ void hashTypeRandomElement(robj *hashobj, unsigned long hashsize, listpackEntry ...@@ -1840,38 +1790,61 @@ void hashTypeRandomElement(robj *hashobj, unsigned long hashsize, listpackEntry
* by returning ACT_REMOVE_EXP_ITEM. * by returning ACT_REMOVE_EXP_ITEM.
* - If hash has no more fields afterward, it will remove the hash from keyspace. * - If hash has no more fields afterward, it will remove the hash from keyspace.
*/ */
static ExpireAction hashTypeActiveExpire(eItem _hashObj, void *ctx) { static ExpireAction hashTypeActiveExpire(eItem item, void *ctx) {
robj *hashObj = (robj *) _hashObj; ExpireCtx *expireCtx = ctx;
ActiveExpireCtx *activeExpireCtx = (ActiveExpireCtx *) ctx;
sds keystr = NULL;
ExpireInfo info = {0};
/* If no more quota left for this callback, stop */ /* If no more quota left for this callback, stop */
if (activeExpireCtx->fieldsToExpireQuota == 0) if (expireCtx->fieldsToExpireQuota == 0)
return ACT_STOP_ACTIVE_EXP; return ACT_STOP_ACTIVE_EXP;
if (hashObj->encoding == OBJ_ENCODING_LISTPACK_EX) { uint64_t nextExpTime = hashTypeExpire((robj *) item, expireCtx, 0);
info = (ExpireInfo){
.maxToExpire = activeExpireCtx->fieldsToExpireQuota, /* If hash has no more fields to expire or got deleted, indicate
* to remove it from HFE DB to the caller ebExpire() */
if (nextExpTime == EB_EXPIRE_TIME_INVALID || nextExpTime == 0) {
return ACT_REMOVE_EXP_ITEM;
} else {
/* Hash has more fields to expire. Update next expiration time of the hash
* and indicate to add it back to global HFE DS */
ebSetMetaExpTime(hashGetExpireMeta(item), nextExpTime);
return ACT_UPDATE_EXP_ITEM;
}
}
/* Delete all expired fields from the hash and delete the hash if left empty.
*
* updateGlobalHFE - If the hash should be updated in the global HFE DS with new
* expiration time in case expired fields were deleted.
*
* Return next Expire time of the hash
* - 0 if hash got deleted
* - EB_EXPIRE_TIME_INVALID if no more fields to expire
*/
static uint64_t hashTypeExpire(robj *o, ExpireCtx *expireCtx, int updateGlobalHFE) {
uint64_t noExpireLeftRes = EB_EXPIRE_TIME_INVALID;
redisDb *db = expireCtx->db;
sds keystr = NULL;
ExpireInfo info = {0};
if (o->encoding == OBJ_ENCODING_LISTPACK_EX) {
info = (ExpireInfo) {
.maxToExpire = expireCtx->fieldsToExpireQuota,
.now = commandTimeSnapshot(), .now = commandTimeSnapshot(),
.itemsExpired = 0}; .itemsExpired = 0};
listpackExExpire(activeExpireCtx->db, hashObj, &info); listpackExExpire(db, o, &info);
server.stat_expired_hash_fields += info.itemsExpired; server.stat_expired_hash_fields += info.itemsExpired;
keystr = ((listpackEx*)hashObj->ptr)->key; keystr = ((listpackEx*)o->ptr)->key;
} else { } else {
serverAssert(hashObj->encoding == OBJ_ENCODING_HT); serverAssert(o->encoding == OBJ_ENCODING_HT);
dict *d = hashObj->ptr; dict *d = o->ptr;
dictExpireMetadata *dictExpireMeta = (dictExpireMetadata *) dictMetadata(d); dictExpireMetadata *dictExpireMeta = (dictExpireMetadata *) dictMetadata(d);
OnFieldExpireCtx onFieldExpireCtx = { OnFieldExpireCtx onFieldExpireCtx = { .hashObj = o, .db = db };
.hashObj = hashObj,
.db = activeExpireCtx->db
};
info = (ExpireInfo){ info = (ExpireInfo){
.maxToExpire = activeExpireCtx->fieldsToExpireQuota, .maxToExpire = expireCtx->fieldsToExpireQuota,
.onExpireItem = onFieldExpire, .onExpireItem = onFieldExpire,
.ctx = &onFieldExpireCtx, .ctx = &onFieldExpireCtx,
.now = commandTimeSnapshot() .now = commandTimeSnapshot()
...@@ -1882,58 +1855,102 @@ static ExpireAction hashTypeActiveExpire(eItem _hashObj, void *ctx) { ...@@ -1882,58 +1855,102 @@ static ExpireAction hashTypeActiveExpire(eItem _hashObj, void *ctx) {
} }
/* Update quota left */ /* Update quota left */
activeExpireCtx->fieldsToExpireQuota -= info.itemsExpired; expireCtx->fieldsToExpireQuota -= info.itemsExpired;
/* If hash has no more fields to expire, remove it from HFE DB */ /* In some cases, a field might have been deleted without updating the global DS.
if (info.nextExpireTime == EB_EXPIRE_TIME_INVALID) { * As a result, active-expire might not expire any fields, in such cases,
if (hashTypeLength(hashObj, 0) == 0) { * we don't need to send notifications or perform other operations for this key. */
robj *key = createStringObject(keystr, sdslen(keystr)); if (info.itemsExpired) {
dbDelete(activeExpireCtx->db, key); robj *key = createStringObject(keystr, sdslen(keystr));
notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key, activeExpireCtx->db->id); notifyKeyspaceEvent(NOTIFY_HASH, "hexpired", key, db->id);
server.dirty++;
signalModifiedKey(NULL, &server.db[0], key); if (updateGlobalHFE)
decrRefCount(key); ebRemove(&db->hexpires, &hashExpireBucketsType, o);
if (hashTypeLength(o, 0) == 0) {
dbDelete(db, key);
notifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, db->id);
noExpireLeftRes = 0;
} else {
if ((updateGlobalHFE) && (info.nextExpireTime != EB_EXPIRE_TIME_INVALID))
ebAdd(&db->hexpires, &hashExpireBucketsType, o, info.nextExpireTime);
} }
return ACT_REMOVE_EXP_ITEM;
} else { signalModifiedKey(NULL, db, key);
/* Hash has more fields to expire. Update next expiration time of the hash decrRefCount(key);
* and indicate to add it back to global HFE DS */
ebSetMetaExpTime(hashGetExpireMeta(hashObj), info.nextExpireTime);
return ACT_UPDATE_EXP_ITEM;
} }
/* return 0 if hash got deleted, EB_EXPIRE_TIME_INVALID if no more fields
* with expiration. Else return next expiration time */
return (info.nextExpireTime == EB_EXPIRE_TIME_INVALID) ? noExpireLeftRes : info.nextExpireTime;
} }
/* Return the next/minimum expiry time of the hash-field. This is useful if a /* Delete all expired fields in hash if needed (Currently used only by HRANDFIELD)
* field with the minimum expiry is deleted, and you want to get the next *
* minimum expiry. Otherwise, consider using hashTypeGetMinExpire() which will * Return 1 if the entire hash was deleted, 0 otherwise.
* be faster. If there is no field with expiry, returns EB_EXPIRE_TIME_INVALID */ * This function might be pricy in case there are many expired fields.
uint64_t hashTypeGetNextTimeToExpire(robj *o) { */
if (o->encoding == OBJ_ENCODING_LISTPACK) { static int hashTypeExpireIfNeeded(redisDb *db, robj *o) {
return EB_EXPIRE_TIME_INVALID; uint64_t nextExpireTime;
} else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) { uint64_t minExpire = hashTypeGetMinExpire(o, 1 /*accurate*/);
return listpackExGetMinExpire(o);
} else {
serverAssert(o->encoding == OBJ_ENCODING_HT);
dict *d = o->ptr; /* Nothing to expire */
if (!isDictWithMetaHFE(d)) if ((mstime_t) minExpire >= commandTimeSnapshot())
return EB_EXPIRE_TIME_INVALID; return 0;
dictExpireMetadata *expireMeta = (dictExpireMetadata *) dictMetadata(d); /* Follow expireIfNeeded() conditions of when not lazy-expire */
return ebGetNextTimeToExpire(expireMeta->hfe, &hashFieldExpireBucketsType); if ( (server.loading) ||
} (server.lazy_expire_disabled) ||
(server.masterhost) || /* master-client or user-client, don't delete */
(isPausedActionsWithUpdate(PAUSE_ACTION_EXPIRE)))
return 0;
/* Take care to expire all the fields */
ExpireCtx expireCtx = { .db = db, .fieldsToExpireQuota = UINT32_MAX };
nextExpireTime = hashTypeExpire(o, &expireCtx, 1);
/* return 1 if the entire hash was deleted */
return nextExpireTime == 0;
} }
/* Return the next/minimum expiry time of the hash-field. /* Return the next/minimum expiry time of the hash-field.
* If not found, return EB_EXPIRE_TIME_INVALID */ * accurate=1 - Return the exact time by looking into the object DS.
uint64_t hashTypeGetMinExpire(robj *o) { * accurate=0 - Return the minimum expiration time maintained in expireMeta which
* might not be accurate due to optimization reasons.
*
* If not found, return EB_EXPIRE_TIME_INVALID
*/
uint64_t hashTypeGetMinExpire(robj *o, int accurate) {
ExpireMeta *expireMeta = NULL; ExpireMeta *expireMeta = NULL;
if (!accurate) {
if (o->encoding == OBJ_ENCODING_LISTPACK) {
return EB_EXPIRE_TIME_INVALID;
} else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) {
listpackEx *lpt = o->ptr;
expireMeta = &lpt->meta;
} else {
serverAssert(o->encoding == OBJ_ENCODING_HT);
dict *d = o->ptr;
if (!isDictWithMetaHFE(d))
return EB_EXPIRE_TIME_INVALID;
expireMeta = &((dictExpireMetadata *) dictMetadata(d))->expireMeta;
}
/* Keep aside next hash-field expiry before updating HFE DS. Verify it is not trash */
if (expireMeta->trash == 1)
return EB_EXPIRE_TIME_INVALID;
return ebGetMetaExpTime(expireMeta);
}
/* accurate == 1 */
if (o->encoding == OBJ_ENCODING_LISTPACK) { if (o->encoding == OBJ_ENCODING_LISTPACK) {
return EB_EXPIRE_TIME_INVALID; return EB_EXPIRE_TIME_INVALID;
} else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) { } else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) {
listpackEx *lpt = o->ptr; return listpackExGetMinExpire(o);
expireMeta = &lpt->meta;
} else { } else {
serverAssert(o->encoding == OBJ_ENCODING_HT); serverAssert(o->encoding == OBJ_ENCODING_HT);
...@@ -1941,14 +1958,9 @@ uint64_t hashTypeGetMinExpire(robj *o) { ...@@ -1941,14 +1958,9 @@ uint64_t hashTypeGetMinExpire(robj *o) {
if (!isDictWithMetaHFE(d)) if (!isDictWithMetaHFE(d))
return EB_EXPIRE_TIME_INVALID; return EB_EXPIRE_TIME_INVALID;
expireMeta = &((dictExpireMetadata *) dictMetadata(d))->expireMeta; dictExpireMetadata *expireMeta = (dictExpireMetadata *) dictMetadata(d);
return ebGetNextTimeToExpire(expireMeta->hfe, &hashFieldExpireBucketsType);
} }
/* Keep aside next hash-field expiry before updating HFE DS. Verify it is not trash */
if (expireMeta->trash == 1)
return EB_EXPIRE_TIME_INVALID;
return ebGetMetaExpTime(expireMeta);
} }
uint64_t hashTypeRemoveFromExpires(ebuckets *hexpires, robj *o) { uint64_t hashTypeRemoveFromExpires(ebuckets *hexpires, robj *o) {
...@@ -2014,7 +2026,7 @@ void hashTypeAddToExpires(redisDb *db, sds key, robj *hashObj, uint64_t expireTi ...@@ -2014,7 +2026,7 @@ void hashTypeAddToExpires(redisDb *db, sds key, robj *hashObj, uint64_t expireTi
* Returns number of fields active-expired. * Returns number of fields active-expired.
*/ */
uint64_t hashTypeDbActiveExpire(redisDb *db, uint32_t maxFieldsToExpire) { uint64_t hashTypeDbActiveExpire(redisDb *db, uint32_t maxFieldsToExpire) {
ActiveExpireCtx ctx = { .db = db, .fieldsToExpireQuota = maxFieldsToExpire }; ExpireCtx ctx = { .db = db, .fieldsToExpireQuota = maxFieldsToExpire };
ExpireInfo info = { ExpireInfo info = {
.maxToExpire = UINT64_MAX, /* Only maxFieldsToExpire play a role */ .maxToExpire = UINT64_MAX, /* Only maxFieldsToExpire play a role */
.onExpireItem = hashTypeActiveExpire, .onExpireItem = hashTypeActiveExpire,
...@@ -2079,7 +2091,7 @@ void hsetnxCommand(client *c) { ...@@ -2079,7 +2091,7 @@ void hsetnxCommand(client *c) {
robj *o; robj *o;
if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return; if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
if (hashTypeExists(c->db, o, c->argv[2]->ptr, &isHashDeleted)) { if (hashTypeExists(c->db, o, c->argv[2]->ptr, HFE_LAZY_EXPIRE, &isHashDeleted)) {
addReply(c, shared.czero); addReply(c, shared.czero);
return; return;
} }
...@@ -2137,7 +2149,8 @@ void hincrbyCommand(client *c) { ...@@ -2137,7 +2149,8 @@ void hincrbyCommand(client *c) {
if (getLongLongFromObjectOrReply(c,c->argv[3],&incr,NULL) != C_OK) return; if (getLongLongFromObjectOrReply(c,c->argv[3],&incr,NULL) != C_OK) return;
if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return; if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
GetFieldRes res = hashTypeGetValue(c->db,o,c->argv[2]->ptr,&vstr,&vlen,&value); GetFieldRes res = hashTypeGetValue(c->db,o,c->argv[2]->ptr,&vstr,&vlen,&value,
HFE_LAZY_EXPIRE);
if (res == GETF_OK) { if (res == GETF_OK) {
if (vstr) { if (vstr) {
if (string2ll((char*)vstr,vlen,&value) == 0) { if (string2ll((char*)vstr,vlen,&value) == 0) {
...@@ -2162,7 +2175,7 @@ void hincrbyCommand(client *c) { ...@@ -2162,7 +2175,7 @@ void hincrbyCommand(client *c) {
} }
value += incr; value += incr;
new = sdsfromlonglong(value); new = sdsfromlonglong(value);
hashTypeSet(c->db, o,c->argv[2]->ptr,new,HASH_SET_TAKE_VALUE | HASH_SET_KEEP_FIELD); hashTypeSet(c->db, o,c->argv[2]->ptr,new,HASH_SET_TAKE_VALUE | HASH_SET_KEEP_TTL);
addReplyLongLong(c,value); addReplyLongLong(c,value);
signalModifiedKey(c,c->db,c->argv[1]); signalModifiedKey(c,c->db,c->argv[1]);
notifyKeyspaceEvent(NOTIFY_HASH,"hincrby",c->argv[1],c->db->id); notifyKeyspaceEvent(NOTIFY_HASH,"hincrby",c->argv[1],c->db->id);
...@@ -2183,7 +2196,8 @@ void hincrbyfloatCommand(client *c) { ...@@ -2183,7 +2196,8 @@ void hincrbyfloatCommand(client *c) {
return; return;
} }
if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return; if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
GetFieldRes res = hashTypeGetValue(c->db, o,c->argv[2]->ptr,&vstr,&vlen,&ll); GetFieldRes res = hashTypeGetValue(c->db, o,c->argv[2]->ptr,&vstr,&vlen,&ll,
HFE_LAZY_EXPIRE);
if (res == GETF_OK) { if (res == GETF_OK) {
if (vstr) { if (vstr) {
if (string2ld((char*)vstr,vlen,&value) == 0) { if (string2ld((char*)vstr,vlen,&value) == 0) {
...@@ -2211,7 +2225,7 @@ void hincrbyfloatCommand(client *c) { ...@@ -2211,7 +2225,7 @@ void hincrbyfloatCommand(client *c) {
char buf[MAX_LONG_DOUBLE_CHARS]; char buf[MAX_LONG_DOUBLE_CHARS];
int len = ld2string(buf,sizeof(buf),value,LD_STR_HUMAN); int len = ld2string(buf,sizeof(buf),value,LD_STR_HUMAN);
new = sdsnewlen(buf,len); new = sdsnewlen(buf,len);
hashTypeSet(c->db, o,c->argv[2]->ptr,new,HASH_SET_TAKE_VALUE | HASH_SET_KEEP_FIELD); hashTypeSet(c->db, o,c->argv[2]->ptr,new,HASH_SET_TAKE_VALUE | HASH_SET_KEEP_TTL);
addReplyBulkCBuffer(c,buf,len); addReplyBulkCBuffer(c,buf,len);
signalModifiedKey(c,c->db,c->argv[1]); signalModifiedKey(c,c->db,c->argv[1]);
notifyKeyspaceEvent(NOTIFY_HASH,"hincrbyfloat",c->argv[1],c->db->id); notifyKeyspaceEvent(NOTIFY_HASH,"hincrbyfloat",c->argv[1],c->db->id);
...@@ -2227,7 +2241,7 @@ void hincrbyfloatCommand(client *c) { ...@@ -2227,7 +2241,7 @@ void hincrbyfloatCommand(client *c) {
decrRefCount(newobj); decrRefCount(newobj);
} }
static GetFieldRes addHashFieldToReply(client *c, robj *o, sds field) { static GetFieldRes addHashFieldToReply(client *c, robj *o, sds field, int hfeFlags) {
if (o == NULL) { if (o == NULL) {
addReplyNull(c); addReplyNull(c);
return GETF_NOT_FOUND; return GETF_NOT_FOUND;
...@@ -2237,7 +2251,7 @@ static GetFieldRes addHashFieldToReply(client *c, robj *o, sds field) { ...@@ -2237,7 +2251,7 @@ static GetFieldRes addHashFieldToReply(client *c, robj *o, sds field) {
unsigned int vlen = UINT_MAX; unsigned int vlen = UINT_MAX;
long long vll = LLONG_MAX; long long vll = LLONG_MAX;
GetFieldRes res = hashTypeGetValue(c->db, o, field, &vstr, &vlen, &vll); GetFieldRes res = hashTypeGetValue(c->db, o, field, &vstr, &vlen, &vll, hfeFlags);
if (res == GETF_OK) { if (res == GETF_OK) {
if (vstr) { if (vstr) {
addReplyBulkCBuffer(c, vstr, vlen); addReplyBulkCBuffer(c, vstr, vlen);
...@@ -2256,13 +2270,14 @@ void hgetCommand(client *c) { ...@@ -2256,13 +2270,14 @@ void hgetCommand(client *c) {
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) == NULL || if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) == NULL ||
checkType(c,o,OBJ_HASH)) return; checkType(c,o,OBJ_HASH)) return;
addHashFieldToReply(c, o, c->argv[2]->ptr); addHashFieldToReply(c, o, c->argv[2]->ptr, HFE_LAZY_EXPIRE);
} }
void hmgetCommand(client *c) { void hmgetCommand(client *c) {
GetFieldRes res = GETF_OK; GetFieldRes res = GETF_OK;
robj *o; robj *o;
int i; int i;
int expired = 0, deleted = 0;
/* Don't abort when the key cannot be found. Non-existing keys are empty /* Don't abort when the key cannot be found. Non-existing keys are empty
* hashes, where HMGET should respond with a series of null bulks. */ * hashes, where HMGET should respond with a series of null bulks. */
...@@ -2271,17 +2286,22 @@ void hmgetCommand(client *c) { ...@@ -2271,17 +2286,22 @@ void hmgetCommand(client *c) {
addReplyArrayLen(c, c->argc-2); addReplyArrayLen(c, c->argc-2);
for (i = 2; i < c->argc ; i++) { for (i = 2; i < c->argc ; i++) {
if (!deleted) {
res = addHashFieldToReply(c, o, c->argv[i]->ptr); res = addHashFieldToReply(c, o, c->argv[i]->ptr, HFE_LAZY_NO_NOTIFICATION);
expired += (res == GETF_EXPIRED);
/* If hash got lazy expired since all fields are expired (o is invalid), deleted += (res == GETF_EXPIRED_HASH);
* then fill the rest with trivial nulls and return */ } else {
if (res == GETF_EXPIRED_HASH) { /* If hash got lazy expired since all fields are expired (o is invalid),
while (++i < c->argc) * then fill the rest with trivial nulls and return. */
addReplyNull(c); addReplyNull(c);
return;
} }
} }
if (expired) {
notifyKeyspaceEvent(NOTIFY_HASH, "hexpired", c->argv[1], c->db->id);
if (deleted)
notifyKeyspaceEvent(NOTIFY_GENERIC, "del", c->argv[1], c->db->id);
}
} }
void hdelCommand(client *c) { void hdelCommand(client *c) {
...@@ -2330,7 +2350,8 @@ void hstrlenCommand(client *c) { ...@@ -2330,7 +2350,8 @@ void hstrlenCommand(client *c) {
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL || if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,o,OBJ_HASH)) return; checkType(c,o,OBJ_HASH)) return;
GetFieldRes res = hashTypeGetValue(c->db, o, c->argv[2]->ptr, &vstr, &vlen, &vll); GetFieldRes res = hashTypeGetValue(c->db, o, c->argv[2]->ptr, &vstr, &vlen, &vll,
HFE_LAZY_EXPIRE);
if (res == GETF_NOT_FOUND || res == GETF_EXPIRED || res == GETF_EXPIRED_HASH) { if (res == GETF_NOT_FOUND || res == GETF_EXPIRED || res == GETF_EXPIRED_HASH) {
addReply(c, shared.czero); addReply(c, shared.czero);
...@@ -2387,7 +2408,7 @@ void genericHgetallCommand(client *c, int flags) { ...@@ -2387,7 +2408,7 @@ void genericHgetallCommand(client *c, int flags) {
/* Skip expired fields if the hash has an expire time set at global HFE DS. We could /* Skip expired fields if the hash has an expire time set at global HFE DS. We could
* set it to constant 1, but then it will make another lookup for each field expiration */ * set it to constant 1, but then it will make another lookup for each field expiration */
int skipExpiredFields = (EB_EXPIRE_TIME_INVALID == hashTypeGetMinExpire(o)) ? 0 : 1; int skipExpiredFields = (EB_EXPIRE_TIME_INVALID == hashTypeGetMinExpire(o, 0)) ? 0 : 1;
while (hashTypeNext(hi, skipExpiredFields) != C_ERR) { while (hashTypeNext(hi, skipExpiredFields) != C_ERR) {
if (flags & OBJ_HASH_KEY) { if (flags & OBJ_HASH_KEY) {
...@@ -2421,11 +2442,11 @@ void hgetallCommand(client *c) { ...@@ -2421,11 +2442,11 @@ void hgetallCommand(client *c) {
void hexistsCommand(client *c) { void hexistsCommand(client *c) {
robj *o; robj *o;
int isHashDeleted;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL || if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,o,OBJ_HASH)) return; checkType(c,o,OBJ_HASH)) return;
addReply(c,hashTypeExists(c->db,o,c->argv[2]->ptr,&isHashDeleted) ? shared.cone : shared.czero); addReply(c,hashTypeExists(c->db,o,c->argv[2]->ptr,HFE_LAZY_EXPIRE, NULL) ?
shared.cone : shared.czero);
} }
void hscanCommand(client *c) { void hscanCommand(client *c) {
...@@ -2473,8 +2494,6 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) { ...@@ -2473,8 +2494,6 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
if ((hash = lookupKeyReadOrReply(c,c->argv[1],shared.emptyarray)) if ((hash = lookupKeyReadOrReply(c,c->argv[1],shared.emptyarray))
== NULL || checkType(c,hash,OBJ_HASH)) return; == NULL || checkType(c,hash,OBJ_HASH)) return;
/* TODO: Active-expire */
size = hashTypeLength(hash, 0);
if(l >= 0) { if(l >= 0) {
count = (unsigned long) l; count = (unsigned long) l;
...@@ -2483,6 +2502,15 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) { ...@@ -2483,6 +2502,15 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
uniq = 0; uniq = 0;
} }
/* Delete all expired fields. If the entire hash got deleted then return empty array. */
if (hashTypeExpireIfNeeded(c->db, hash)) {
addReply(c, shared.emptyarray);
return;
}
/* Delete expired fields */
size = hashTypeLength(hash, 0);
/* If count is zero, serve it ASAP to avoid special cases later. */ /* If count is zero, serve it ASAP to avoid special cases later. */
if (count == 0) { if (count == 0) {
addReply(c,shared.emptyarray); addReply(c,shared.emptyarray);
...@@ -2586,64 +2614,50 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) { ...@@ -2586,64 +2614,50 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
} }
/* CASE 3: /* CASE 3:
* The number of elements inside the hash is not greater than * The number of elements inside the hash of type dict is not greater than
* HRANDFIELD_SUB_STRATEGY_MUL times the number of requested elements. * HRANDFIELD_SUB_STRATEGY_MUL times the number of requested elements.
* In this case we create a hash from scratch with all the elements, and * In this case we create an array of dictEntry pointers from the original hash,
* subtract random elements to reach the requested number of elements. * and subtract random elements to reach the requested number of elements.
* *
* This is done because if the number of requested elements is just * This is done because if the number of requested elements is just
* a bit less than the number of elements in the hash, the natural approach * a bit less than the number of elements in the hash, the natural approach
* used into CASE 4 is highly inefficient. */ * used into CASE 4 is highly inefficient. */
if (count*HRANDFIELD_SUB_STRATEGY_MUL > size) { if (count*HRANDFIELD_SUB_STRATEGY_MUL > size) {
/* Hashtable encoding (generic implementation) */ /* Hashtable encoding (generic implementation) */
dict *d = dictCreate(&sdsReplyDictType); /* without metadata! */ dict *ht = hash->ptr;
dictExpand(d, size); dictIterator *di;
hashTypeIterator *hi = hashTypeInitIterator(hash); dictEntry *de;
unsigned long idx = 0;
/* Add all the elements into the temporary dictionary. */
while ((hashTypeNext(hi, 0)) != C_ERR) { /* Allocate a temporary array of pointers to stored key-values in dict and
int ret = DICT_ERR; * assist it to remove random elements to reach the right count. */
sds key, value = NULL; struct FieldValPair {
hfield field;
key = hashTypeCurrentObjectNewSds(hi,OBJ_HASH_KEY); sds value;
if (withvalues) } *pairs = zmalloc(sizeof(struct FieldValPair) * size);
value = hashTypeCurrentObjectNewSds(hi,OBJ_HASH_VALUE);
ret = dictAdd(d, key, value); /* Add all the elements into the temporary array. */
di = dictGetIterator(ht);
serverAssert(ret == DICT_OK); while((de = dictNext(di)) != NULL)
} pairs[idx++] = (struct FieldValPair) {dictGetKey(de), dictGetVal(de)};
serverAssert(dictSize(d) == size); dictReleaseIterator(di);
hashTypeReleaseIterator(hi);
/* Remove random elements to reach the right count. */ /* Remove random elements to reach the right count. */
while (size > count) { while (size > count) {
dictEntry *de; unsigned long toDiscardIdx = rand() % size;
de = dictGetFairRandomKey(d); pairs[toDiscardIdx] = pairs[--size];
dictUseStoredKeyApi(d, 1);
dictUnlink(d,dictGetKey(de));
dictUseStoredKeyApi(d, 0);
sdsfree(dictGetKey(de));
sdsfree(dictGetVal(de));
dictFreeUnlinkedEntry(d,de);
size--;
} }
/* Reply with what's in the dict and release memory */ /* Reply with what's in the array */
dictIterator *di; for (idx = 0; idx < size; idx++) {
dictEntry *de;
di = dictGetIterator(d);
while ((de = dictNext(di)) != NULL) {
sds key = dictGetKey(de);
sds value = dictGetVal(de);
if (withvalues && c->resp > 2) if (withvalues && c->resp > 2)
addReplyArrayLen(c,2); addReplyArrayLen(c,2);
addReplyBulkSds(c, key); addReplyBulkCBuffer(c, pairs[idx].field, hfieldlen(pairs[idx].field));
if (withvalues) if (withvalues)
addReplyBulkSds(c, value); addReplyBulkCBuffer(c, pairs[idx].value, sdslen(pairs[idx].value));
} }
dictReleaseIterator(di); zfree(pairs);
dictRelease(d);
} }
/* CASE 4: We have a big hash compared to the requested number of elements. /* CASE 4: We have a big hash compared to the requested number of elements.
...@@ -2651,43 +2665,78 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) { ...@@ -2651,43 +2665,78 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
* to the temporary hash, trying to eventually get enough unique elements * to the temporary hash, trying to eventually get enough unique elements
* to reach the specified count. */ * to reach the specified count. */
else { else {
/* Allocate temporary dictUnique to find unique elements. Just keep ref
* to key-value from the original hash. This dict relaxes hash function
* to be based on field's pointer */
dictType uniqueDictType = { .hashFunction = dictPtrHash };
dict *dictUnique = dictCreate(&uniqueDictType);
dictExpand(dictUnique, count);
/* Hashtable encoding (generic implementation) */ /* Hashtable encoding (generic implementation) */
unsigned long added = 0; unsigned long added = 0;
listpackEntry key, value;
dict *d = dictCreate(&hashDictType);
dictExpand(d, count);
while(added < count) { while(added < count) {
hashTypeRandomElement(hash, size, &key, withvalues? &value : NULL); dictEntry *de = dictGetFairRandomKey(hash->ptr);
serverAssert(de != NULL);
hfield field = dictGetKey(de);
sds value = dictGetVal(de);
/* Try to add the object to the dictionary. If it already exists /* Try to add the object to the dictionary. If it already exists
* free it, otherwise increment the number of objects we have * free it, otherwise increment the number of objects we have
* in the result dictionary. */ * in the result dictionary. */
sds skey = hashSdsFromListpackEntry(&key); if (dictAdd(dictUnique, field, value) != DICT_OK)
if (dictAdd(d,skey,NULL) != DICT_OK) {
sdsfree(skey);
continue; continue;
}
added++; added++;
/* We can reply right away, so that we don't need to store the value in the dict. */ /* We can reply right away, so that we don't need to store the value in the dict. */
if (withvalues && c->resp > 2) if (withvalues && c->resp > 2)
addReplyArrayLen(c,2); addReplyArrayLen(c,2);
hashReplyFromListpackEntry(c, &key);
addReplyBulkCBuffer(c, field, hfieldlen(field));
if (withvalues) if (withvalues)
hashReplyFromListpackEntry(c, &value); addReplyBulkCBuffer(c, value, sdslen(value));
} }
/* Release memory */ /* Release memory */
dictRelease(d); dictRelease(dictUnique);
} }
} }
/* HRANDFIELD key [<count> [WITHVALUES]] */ /*
* HRANDFIELD - Return a random field from the hash value stored at key.
* CLI usage: HRANDFIELD key [<count> [WITHVALUES]]
*
* Considerations for the current imp of HRANDFIELD & HFE feature:
* HRANDFIELD might access any of the fields in the hash as some of them might
* be expired. And so the Implementation of HRANDFIELD along with HFEs
* might be one of the two options:
* 1. Expire hash-fields before diving into handling HRANDFIELD.
* 2. Refine HRANDFIELD cases to deal with expired fields.
*
* Regarding the first option, as reference, the command RANDOMKEY also declares
* on O(1) complexity, yet might be stuck on a very long (but not infinite) loop
* trying to find non-expired keys. Furthermore RANDOMKEY also evicts expired keys
* along the way even though it is categorized as a read-only command. Note that
* the case of HRANDFIELD is more lightweight versus RANDOMKEY since HFEs have
* much more effective and aggressive active-expiration for fields behind.
*
* The second option introduces additional implementation complexity to HRANDFIELD.
* We could further refine HRANDFIELD cases to differentiate between scenarios
* with many expired fields versus few expired fields, and adjust based on the
* percentage of expired fields. However, this approach could still lead to long
* loops or necessitate expiring fields before selecting them. For the “lightweight”
* cases it is also expected to have a lightweight expiration.
*
* Considering the pros and cons, and the fact that HRANDFIELD is an infrequent
* command (particularly with HFEs) and the fact we have effective active-expiration
* behind for hash-fields, it is better to keep it simple and choose the option #1.
*/
void hrandfieldCommand(client *c) { void hrandfieldCommand(client *c) {
long l; long l;
int withvalues = 0; int withvalues = 0;
robj *hash; robj *hash;
listpackEntry ele; CommonEntry ele;
if (c->argc >= 3) { if (c->argc >= 3) {
if (getRangeLongFromObjectOrReply(c,c->argv[2],-LONG_MAX,LONG_MAX,&l,NULL) != C_OK) return; if (getRangeLongFromObjectOrReply(c,c->argv[2],-LONG_MAX,LONG_MAX,&l,NULL) != C_OK) return;
...@@ -2711,8 +2760,18 @@ void hrandfieldCommand(client *c) { ...@@ -2711,8 +2760,18 @@ void hrandfieldCommand(client *c) {
return; return;
} }
/* Delete all expired fields. If the entire hash got deleted then return null. */
if (hashTypeExpireIfNeeded(c->db, hash)) {
addReply(c,shared.null[c->resp]);
return;
}
hashTypeRandomElement(hash,hashTypeLength(hash, 0),&ele,NULL); hashTypeRandomElement(hash,hashTypeLength(hash, 0),&ele,NULL);
hashReplyFromListpackEntry(c, &ele);
if (ele.sval)
addReplyBulkCBuffer(c, ele.sval, ele.slen);
else
addReplyBulkLongLong(c, ele.lval);
} }
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
...@@ -2847,6 +2906,7 @@ static ExpireMeta *hashGetExpireMeta(const eItem hash) { ...@@ -2847,6 +2906,7 @@ static ExpireMeta *hashGetExpireMeta(const eItem hash) {
serverPanic("Unknown encoding: %d", hashObj->encoding); serverPanic("Unknown encoding: %d", hashObj->encoding);
} }
} }
/* HTTL key <FIELDS count field [field ...]> */ /* HTTL key <FIELDS count field [field ...]> */
static void httlGenericCommand(client *c, const char *cmd, long long basetime, int unit) { static void httlGenericCommand(client *c, const char *cmd, long long basetime, int unit) {
UNUSED(cmd); UNUSED(cmd);
...@@ -2854,8 +2914,9 @@ static void httlGenericCommand(client *c, const char *cmd, long long basetime, i ...@@ -2854,8 +2914,9 @@ static void httlGenericCommand(client *c, const char *cmd, long long basetime, i
long numFields = 0, numFieldsAt = 3; long numFields = 0, numFieldsAt = 3;
/* Read the hash object */ /* Read the hash object */
if ((hashObj = lookupKeyReadOrReply(c, c->argv[1], shared.emptyarray)) == NULL || hashObj = lookupKeyRead(c->db, c->argv[1]);
checkType(c, hashObj, OBJ_HASH)) return; if (checkType(c, hashObj, OBJ_HASH))
return;
if (strcasecmp(c->argv[numFieldsAt-1]->ptr, "FIELDS")) { if (strcasecmp(c->argv[numFieldsAt-1]->ptr, "FIELDS")) {
addReplyError(c, "Mandatory argument FIELDS is missing or not at the right position"); addReplyError(c, "Mandatory argument FIELDS is missing or not at the right position");
...@@ -2868,8 +2929,18 @@ static void httlGenericCommand(client *c, const char *cmd, long long basetime, i ...@@ -2868,8 +2929,18 @@ static void httlGenericCommand(client *c, const char *cmd, long long basetime, i
return; return;
/* Verify `numFields` is consistent with number of arguments */ /* Verify `numFields` is consistent with number of arguments */
if (numFields > (c->argc - numFieldsAt - 1)) { if (numFields != (c->argc - numFieldsAt - 1)) {
addReplyError(c, "Parameter `numFields` is more than number of arguments"); addReplyError(c, "The `numfields` parameter must match the number of arguments");
return;
}
/* Non-existing keys and empty hashes are the same thing. It also means
* fields in the command don't exist in the hash key. */
if (!hashObj) {
addReplyArrayLen(c, numFields);
for (int i = 0; i < numFields; i++) {
addReplyLongLong(c, HFE_GET_NO_FIELD);
}
return; return;
} }
...@@ -2969,27 +3040,52 @@ static void httlGenericCommand(client *c, const char *cmd, long long basetime, i ...@@ -2969,27 +3040,52 @@ static void httlGenericCommand(client *c, const char *cmd, long long basetime, i
* unit is either UNIT_SECONDS or UNIT_MILLISECONDS, and is only used for * unit is either UNIT_SECONDS or UNIT_MILLISECONDS, and is only used for
* the argv[2] parameter. The basetime is always specified in milliseconds. * the argv[2] parameter. The basetime is always specified in milliseconds.
* *
* Additional flags are supported and parsed via parseExtendedExpireArguments */ * PROPAGATE TO REPLICA:
* The command will be translated into HPEXPIREAT and the expiration time will be
* converted to absolute time in milliseconds.
*
* As we need to propagate H(P)EXPIRE(AT) command to the replica, each field that
* is mentioned in the command should be categorized into one of the four options:
* 1. Field’s expiration time updated successfully - Propagate it to replica as
* part of the HPEXPIREAT command.
* 2. The field got deleted since the time is in the past - propagate also HDEL
* command to delete the field. Also remove the field from the propagated
* HPEXPIREAT command.
* 3. Condition not met for the field - Remove the field from the propagated
* HPEXPIREAT command.
* 4. Field doesn't exists - Remove the field from propagated HPEXPIREAT command.
*
* If none of the provided fields match option #1, that is provided time of the
* command is in the past, then avoid propagating the HPEXPIREAT command to the
* replica.
*
* This approach is aligned with existing EXPIRE command. If a given key has already
* expired, then DEL will be propagated instead of EXPIRE command. If condition
* not met, then command will be rejected. Otherwise, EXPIRE command will be
* propagated for given key.
*/
static void hexpireGenericCommand(client *c, const char *cmd, long long basetime, int unit) { static void hexpireGenericCommand(client *c, const char *cmd, long long basetime, int unit) {
long numFields = 0, numFieldsAt = 4; long numFields = 0, numFieldsAt = 4;
long long expire; /* unix time in msec */ long long expire; /* unix time in msec */
int expireSetCond = 0; int fieldAt, fieldsNotSet = 0, expireSetCond = 0;
robj *hashObj, *keyArg = c->argv[1], *expireArg = c->argv[2]; robj *hashObj, *keyArg = c->argv[1], *expireArg = c->argv[2];
/* Read the hash object */ /* Read the hash object */
if ((hashObj = lookupKeyWriteOrReply(c, keyArg, shared.emptyarray)) == NULL || hashObj = lookupKeyWrite(c->db, keyArg);
checkType(c, hashObj, OBJ_HASH)) return; if (checkType(c, hashObj, OBJ_HASH))
return;
/* Read the expiry time from command */ /* Read the expiry time from command */
if (getLongLongFromObjectOrReply(c, expireArg, &expire, NULL) != C_OK) if (getLongLongFromObjectOrReply(c, expireArg, &expire, NULL) != C_OK)
return; return;
if (expire < 0) { if (expire < 0) {
addReplyError(c,"invalid expire time, must be >= 0"); addReplyError(c,"invalid expire time, must be >= 0");
return; return;
} }
if (unit == UNIT_SECONDS) { if (unit == UNIT_SECONDS) {
if (expire > (long long) EB_EXPIRE_TIME_MAX / 1000) { if (expire > (long long) HFE_MAX_ABS_TIME_MSEC / 1000) {
addReplyErrorExpireTime(c); addReplyErrorExpireTime(c);
return; return;
} }
...@@ -2997,7 +3093,7 @@ static void hexpireGenericCommand(client *c, const char *cmd, long long basetime ...@@ -2997,7 +3093,7 @@ static void hexpireGenericCommand(client *c, const char *cmd, long long basetime
} }
/* Ensure that the final absolute Unix timestamp does not exceed EB_EXPIRE_TIME_MAX. */ /* Ensure that the final absolute Unix timestamp does not exceed EB_EXPIRE_TIME_MAX. */
if (expire > (long long) EB_EXPIRE_TIME_MAX - basetime) { if (expire > (long long) HFE_MAX_ABS_TIME_MSEC - basetime) {
addReplyErrorExpireTime(c); addReplyErrorExpireTime(c);
return; return;
} }
...@@ -3026,27 +3122,57 @@ static void hexpireGenericCommand(client *c, const char *cmd, long long basetime ...@@ -3026,27 +3122,57 @@ static void hexpireGenericCommand(client *c, const char *cmd, long long basetime
return; return;
/* Verify `numFields` is consistent with number of arguments */ /* Verify `numFields` is consistent with number of arguments */
if (numFields > (c->argc - numFieldsAt - 1)) { if (numFields != (c->argc - numFieldsAt - 1)) {
addReplyError(c, "Parameter `numFields` is more than number of arguments"); addReplyError(c, "The `numfields` parameter must match the number of arguments");
return; return;
} }
HashTypeSetEx exCtx; /* Non-existing keys and empty hashes are the same thing. It also means
hashTypeSetExInit(keyArg, hashObj, c, c->db, cmd, * fields in the command don't exist in the hash key. */
FIELD_DONT_CREATE2, if (!hashObj) {
expireSetCond, addReplyArrayLen(c, numFields);
&exCtx); for (int i = 0; i < numFields; i++) {
addReplyLongLong(c, HSETEX_NO_FIELD);
}
return;
}
HashTypeSetEx exCtx;
hashTypeSetExInit(keyArg, hashObj, c, c->db, cmd, expireSetCond, &exCtx);
addReplyArrayLen(c, numFields); addReplyArrayLen(c, numFields);
for (int i = 0 ; i < numFields ; i++) { fieldAt = numFieldsAt + 1;
sds field = c->argv[numFieldsAt+i+1]->ptr; while (fieldAt < c->argc) {
SetExRes res = hashTypeSetEx(c->db, hashObj, field, NULL, expire, &exCtx); sds field = c->argv[fieldAt]->ptr;
SetExRes res = hashTypeSetEx(hashObj, field, expire, &exCtx);
if (unlikely(res != HSETEX_OK)) {
/* If the field was not set, prevent field propagation */
rewriteClientCommandArgument(c, fieldAt, NULL);
fieldsNotSet = 1;
} else {
++fieldAt;
}
addReplyLongLong(c,res); addReplyLongLong(c,res);
} }
hashTypeSetExDone(&exCtx); hashTypeSetExDone(&exCtx);
/* rewrite command for the replica sake */ /* Avoid propagating command if not even one field was updated (Either because
* the time is in the past, and corresponding HDELs were sent, or conditions
* not met) then it is useless and invalid to propagate command with no fields */
if (exCtx.fieldUpdated == 0) {
preventCommandPropagation(c);
return;
}
/* If some fields were dropped, rewrite the number of fields */
if (fieldsNotSet) {
robj *numFieldsObj = createStringObjectFromLongLong(exCtx.fieldUpdated);
rewriteClientCommandArgument(c, numFieldsAt, numFieldsObj);
decrRefCount(numFieldsObj);
}
/* Propagate as HPEXPIREAT millisecond-timestamp. Rewrite only if not already */ /* Propagate as HPEXPIREAT millisecond-timestamp. Rewrite only if not already */
if (c->cmd->proc != hpexpireatCommand) { if (c->cmd->proc != hpexpireatCommand) {
...@@ -3109,8 +3235,9 @@ void hpersistCommand(client *c) { ...@@ -3109,8 +3235,9 @@ void hpersistCommand(client *c) {
int changed = 0; /* Used to determine whether to send a notification. */ int changed = 0; /* Used to determine whether to send a notification. */
/* Read the hash object */ /* Read the hash object */
if ((hashObj = lookupKeyWriteOrReply(c, c->argv[1], shared.emptyarray)) == NULL || hashObj = lookupKeyWrite(c->db, c->argv[1]);
checkType(c, hashObj, OBJ_HASH)) return; if (checkType(c, hashObj, OBJ_HASH))
return;
if (strcasecmp(c->argv[numFieldsAt-1]->ptr, "FIELDS")) { if (strcasecmp(c->argv[numFieldsAt-1]->ptr, "FIELDS")) {
addReplyError(c, "Mandatory argument FIELDS is missing or not at the right position"); addReplyError(c, "Mandatory argument FIELDS is missing or not at the right position");
...@@ -3123,8 +3250,18 @@ void hpersistCommand(client *c) { ...@@ -3123,8 +3250,18 @@ void hpersistCommand(client *c) {
return; return;
/* Verify `numFields` is consistent with number of arguments */ /* Verify `numFields` is consistent with number of arguments */
if (numFields > (c->argc - numFieldsAt - 1)) { if (numFields != (c->argc - numFieldsAt - 1)) {
addReplyError(c, "Parameter `numFields` is more than number of arguments"); addReplyError(c, "The `numfields` parameter must match the number of arguments");
return;
}
/* Non-existing keys and empty hashes are the same thing. It also means
* fields in the command don't exist in the hash key. */
if (!hashObj) {
addReplyArrayLen(c, numFields);
for (int i = 0; i < numFields; i++) {
addReplyLongLong(c, HFE_PERSIST_NO_FIELD);
}
return; return;
} }
...@@ -3216,5 +3353,9 @@ void hpersistCommand(client *c) { ...@@ -3216,5 +3353,9 @@ void hpersistCommand(client *c) {
/* Generates a hpersist event if the expiry time associated with any field /* Generates a hpersist event if the expiry time associated with any field
* has been successfully deleted. */ * has been successfully deleted. */
if (changed) notifyKeyspaceEvent(NOTIFY_HASH,"hpersist",c->argv[1],c->db->id); if (changed) {
notifyKeyspaceEvent(NOTIFY_HASH, "hpersist", c->argv[1], c->db->id);
signalModifiedKey(c, c->db, c->argv[1]);
server.dirty++;
}
} }
#define REDIS_VERSION "7.3.240" #define REDIS_VERSION "7.3.241"
#define REDIS_VERSION_NUM 0x000703f0 #define REDIS_VERSION_NUM 0x000703f1
...@@ -429,8 +429,8 @@ start_server [list overrides [list "dir" $server_path]] { ...@@ -429,8 +429,8 @@ start_server [list overrides [list "dir" $server_path]] {
r HMSET key a 1 b 2 c 3 d 4 e 5 r HMSET key a 1 b 2 c 3 d 4 e 5
# expected to be expired long after restart # expected to be expired long after restart
r HEXPIREAT key 2524600800 FIELDS 1 a r HEXPIREAT key 2524600800 FIELDS 1 a
# expected long TTL value (6 bytes) is saved and loaded correctly # expected long TTL value (46 bits) is saved and loaded correctly
r HPEXPIREAT key 188900976391764 FIELDS 1 b r HPEXPIREAT key 65755674080852 FIELDS 1 b
# expected to be already expired after restart # expected to be already expired after restart
r HPEXPIRE key 80 FIELDS 1 d r HPEXPIRE key 80 FIELDS 1 d
# expected to be expired soon after restart # expected to be expired soon after restart
...@@ -443,7 +443,7 @@ start_server [list overrides [list "dir" $server_path]] { ...@@ -443,7 +443,7 @@ start_server [list overrides [list "dir" $server_path]] {
wait_done_loading r wait_done_loading r
assert_equal [lsort [r hgetall key]] "1 2 3 a b c" assert_equal [lsort [r hgetall key]] "1 2 3 a b c"
assert_equal [r hpexpiretime key FIELDS 3 a b c] {2524600800000 188900976391764 -1} assert_equal [r hpexpiretime key FIELDS 3 a b c] {2524600800000 65755674080852 -1}
assert_equal [s rdb_last_load_keys_loaded] 1 assert_equal [s rdb_last_load_keys_loaded] 1
# wait until expired_hash_fields equals 2 # wait until expired_hash_fields equals 2
......
...@@ -302,6 +302,21 @@ int propagateTestIncr(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -302,6 +302,21 @@ int propagateTestIncr(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
return REDISMODULE_OK; return REDISMODULE_OK;
} }
int propagateTestVerbatim(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc < 2){
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
long long replicate_num;
RedisModule_StringToLongLong(argv[1], &replicate_num);
/* Replicate the command verbatim for the specified number of times. */
for (long long i = 0; i < replicate_num; i++)
RedisModule_ReplicateVerbatim(ctx);
RedisModule_ReplyWithSimpleString(ctx,"OK");
return REDISMODULE_OK;
}
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) { int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv); REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc); REDISMODULE_NOT_USED(argc);
...@@ -368,6 +383,11 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -368,6 +383,11 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
propagateTestIncr, propagateTestIncr,
"write",1,1,1) == REDISMODULE_ERR) "write",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"propagate-test.verbatim",
propagateTestVerbatim,
"write",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK; return REDISMODULE_OK;
} }
......
...@@ -21,6 +21,45 @@ start_server {tags {"modules"}} { ...@@ -21,6 +21,45 @@ start_server {tags {"modules"}} {
r hgetall k r hgetall k
} {squirrel ofcourse banana no what nothing something nice} } {squirrel ofcourse banana no what nothing something nice}
test {Module hash - set (override) NX expired field successfully} {
r debug set-active-expire 0
r del H1 H2
r hash.set H1 "n" f1 v1
r hpexpire H1 1 FIELDS 1 f1
r hash.set H2 "n" f1 v1 f2 v2
r hpexpire H2 1 FIELDS 1 f1
after 5
assert_equal 0 [r hash.set H1 "n" f1 xx]
assert_equal "f1 xx" [r hgetall H1]
assert_equal 0 [r hash.set H2 "n" f1 yy]
assert_equal "f1 f2 v2 yy" [lsort [r hgetall H2]]
r debug set-active-expire 1
} {OK} {needs:debug}
test {Module hash - set XX of expired field gets failed as expected} {
r debug set-active-expire 0
r del H1 H2
r hash.set H1 "n" f1 v1
r hpexpire H1 1 FIELDS 1 f1
r hash.set H2 "n" f1 v1 f2 v2
r hpexpire H2 1 FIELDS 1 f1
after 5
# expected to fail on condition XX. hgetall should return empty list
r hash.set H1 "x" f1 xx
assert_equal "" [lsort [r hgetall H1]]
# But expired field was not lazy deleted
assert_equal 1 [r hlen H1]
# expected to fail on condition XX. hgetall should return list without expired f1
r hash.set H2 "x" f1 yy
assert_equal "f2 v2" [lsort [r hgetall H2]]
# But expired field was not lazy deleted
assert_equal 2 [r hlen H2]
r debug set-active-expire 1
} {OK} {needs:debug}
test "Unload the module - hash" { test "Unload the module - hash" {
assert_equal {OK} [r module unload hash] assert_equal {OK} [r module unload hash]
} }
......
...@@ -761,3 +761,41 @@ tags "modules aof" { ...@@ -761,3 +761,41 @@ tags "modules aof" {
} }
} }
} }
# This test does not really test module functionality, but rather uses a module
# command to test Redis replication mechanisms.
test {Replicas that was marked as CLIENT_CLOSE_ASAP should not keep the replication backlog from been trimmed} {
start_server [list overrides [list loadmodule "$testmodule"]] {
set replica [srv 0 client]
start_server [list overrides [list loadmodule "$testmodule"]] {
set master [srv 0 client]
set master_host [srv 0 host]
set master_port [srv 0 port]
$master config set client-output-buffer-limit "replica 10mb 5mb 0"
# Start the replication process...
$replica replicaof $master_host $master_port
wait_for_sync $replica
test {module propagates from timer} {
# Replicate large commands to make the replica disconnected.
$master write [format_command propagate-test.verbatim 100000 [string repeat "a" 1000]] ;# almost 100mb
# Execute this command together with module commands within the same
# event loop to prevent periodic cleanup of replication backlog.
$master write [format_command info memory]
$master flush
$master read ;# propagate-test.verbatim
set res [$master read] ;# info memory
# Wait for the replica to be disconnected.
wait_for_log_messages 0 {"*flags=S*scheduled to be closed ASAP for overcoming of output buffer limits*"} 0 1500 10
# Due to the replica reaching the soft limit (5MB), memory peaks should not significantly
# exceed the replica soft limit. Furthermore, as the replica release its reference to
# replication backlog, it should be properly trimmed, the memory usage of replication
# backlog should not significantly exceed repl-backlog-size (default 1MB). */
assert_lessthan [getInfoProperty $res used_memory_peak] 10000000;# less than 10mb
assert_lessthan [getInfoProperty $res mem_replication_backlog] 2000000;# less than 2mb
}
}
}
}
...@@ -25,12 +25,16 @@ start_server {tags {"modules"}} { ...@@ -25,12 +25,16 @@ start_server {tags {"modules"}} {
} {{f1 1}} } {{f1 1}}
test {Module scan hash listpack with hexpire} { test {Module scan hash listpack with hexpire} {
r hmset hh f1 v1 f2 v2 r debug set-active-expire 0
r hmset hh f1 v1 f2 v2 f3 v3
r hexpire hh 100000 fields 1 f1 r hexpire hh 100000 fields 1 f1
r hpexpire hh 1 fields 1 f3
after 10
assert_range [r httl hh fields 1 f1] 10000 100000 assert_range [r httl hh fields 1 f1] 10000 100000
assert_encoding listpackex hh assert_encoding listpackex hh
r debug set-active-expire 1
lsort [r scan.scan_key hh] lsort [r scan.scan_key hh]
} {{f1 v1} {f2 v2}} } {{f1 v1} {f2 v2}} {needs:debug}
test {Module scan hash dict} { test {Module scan hash dict} {
r config set hash-max-ziplist-entries 2 r config set hash-max-ziplist-entries 2
...@@ -44,10 +48,22 @@ start_server {tags {"modules"}} { ...@@ -44,10 +48,22 @@ start_server {tags {"modules"}} {
r del hh r del hh
r hmset hh f1 v1 f2 v2 f3 v3 r hmset hh f1 v1 f2 v2 f3 v3
r hexpire hh 100000 fields 1 f2 r hexpire hh 100000 fields 1 f2
r hpexpire hh 5 fields 1 f3
assert_range [r httl hh fields 1 f2] 10000 100000 assert_range [r httl hh fields 1 f2] 10000 100000
assert_encoding hashtable hh assert_encoding hashtable hh
after 10
lsort [r scan.scan_key hh] lsort [r scan.scan_key hh]
} {{f1 v1} {f2 v2} {f3 v3}} } {{f1 v1} {f2 v2}}
test {Module scan hash with hexpire can return no items} {
r del hh
r debug set-active-expire 0
r hmset hh f1 v1 f2 v2 f3 v3
r hpexpire hh 1 fields 3 f1 f2 f3
after 10
r debug set-active-expire 1
lsort [r scan.scan_key hh]
} {} {needs:debug}
test {Module scan zset listpack} { test {Module scan zset listpack} {
r zadd zz 1 f1 2 f2 r zadd zz 1 f1 2 f2
......
...@@ -356,16 +356,17 @@ start_server {tags {"pubsub network"}} { ...@@ -356,16 +356,17 @@ start_server {tags {"pubsub network"}} {
foreach {type max_lp_entries} {listpackex 512 hashtable 0} { foreach {type max_lp_entries} {listpackex 512 hashtable 0} {
test "Keyspace notifications: hash events test ($type)" { test "Keyspace notifications: hash events test ($type)" {
r config set hash-max-listpack-entries $max_lp_entries r config set hash-max-listpack-entries $max_lp_entries
r config set notify-keyspace-events Kh r config set notify-keyspace-events Khg
r del myhash r del myhash
set rd1 [redis_deferring_client] set rd1 [redis_deferring_client]
assert_equal {1} [psubscribe $rd1 *] assert_equal {1} [psubscribe $rd1 *]
r hmset myhash yes 1 no 0 r hmset myhash yes 1 no 0 f1 1 f2 2 f3_hdel 3
r hincrby myhash yes 10 r hincrby myhash yes 10
r hexpire myhash 999999 FIELDS 1 yes r hexpire myhash 999999 FIELDS 1 yes
r hexpireat myhash [expr {[clock seconds] + 999999}] NX FIELDS 1 no r hexpireat myhash [expr {[clock seconds] + 999999}] NX FIELDS 1 no
r hpexpire myhash 999999 FIELDS 1 yes r hpexpire myhash 999999 FIELDS 1 yes
r hpersist myhash FIELDS 1 yes r hpersist myhash FIELDS 1 yes
r hpexpire myhash 0 FIELDS 1 yes
assert_encoding $type myhash assert_encoding $type myhash
assert_equal "pmessage * __keyspace@${db}__:myhash hset" [$rd1 read] assert_equal "pmessage * __keyspace@${db}__:myhash hset" [$rd1 read]
assert_equal "pmessage * __keyspace@${db}__:myhash hincrby" [$rd1 read] assert_equal "pmessage * __keyspace@${db}__:myhash hincrby" [$rd1 read]
...@@ -373,8 +374,38 @@ start_server {tags {"pubsub network"}} { ...@@ -373,8 +374,38 @@ start_server {tags {"pubsub network"}} {
assert_equal "pmessage * __keyspace@${db}__:myhash hexpire" [$rd1 read] assert_equal "pmessage * __keyspace@${db}__:myhash hexpire" [$rd1 read]
assert_equal "pmessage * __keyspace@${db}__:myhash hexpire" [$rd1 read] assert_equal "pmessage * __keyspace@${db}__:myhash hexpire" [$rd1 read]
assert_equal "pmessage * __keyspace@${db}__:myhash hpersist" [$rd1 read] assert_equal "pmessage * __keyspace@${db}__:myhash hpersist" [$rd1 read]
assert_equal "pmessage * __keyspace@${db}__:myhash hexpired" [$rd1 read]
# Test that we will get `hexpired` notification when
# a hash field is removed by active expire.
r hpexpire myhash 10 FIELDS 1 no
after 100 ;# Wait for active expire
assert_equal "pmessage * __keyspace@${db}__:myhash hexpire" [$rd1 read]
assert_equal "pmessage * __keyspace@${db}__:myhash hexpired" [$rd1 read]
# Test that when a field with TTL is deleted by commands like hdel without
# updating the global DS, active expire will not send a notification.
r hpexpire myhash 100 FIELDS 1 f3_hdel
r hdel myhash f3_hdel
after 200 ;# Wait for active expire
assert_equal "pmessage * __keyspace@${db}__:myhash hexpire" [$rd1 read]
assert_equal "pmessage * __keyspace@${db}__:myhash hdel" [$rd1 read]
# Test that we will get `hexpired` notification when
# a hash field is removed by lazy expire.
r debug set-active-expire 0
r hpexpire myhash 10 FIELDS 2 f1 f2
after 20
r hmget myhash f1 f2 ;# Trigger lazy expire
assert_equal "pmessage * __keyspace@${db}__:myhash hexpire" [$rd1 read]
# We should get only one `hexpired` notification even two fields was expired.
assert_equal "pmessage * __keyspace@${db}__:myhash hexpired" [$rd1 read]
# We should get a `del` notification after all fields were expired.
assert_equal "pmessage * __keyspace@${db}__:myhash del" [$rd1 read]
r debug set-active-expire 1
$rd1 close $rd1 close
} } {0} {needs:debug}
} ;# foreach } ;# foreach
test "Keyspace notifications: stream events test" { test "Keyspace notifications: stream events test" {
......
...@@ -32,13 +32,6 @@ proc get_hashes_with_expiry_fields {r} { ...@@ -32,13 +32,6 @@ proc get_hashes_with_expiry_fields {r} {
return 0 return 0
} }
proc create_hash {key entries} {
r del $key
foreach entry $entries {
r hset $key [lindex $entry 0] [lindex $entry 1]
}
}
proc get_keys {l} { proc get_keys {l} {
set res {} set res {}
foreach entry $l { foreach entry $l {
...@@ -48,22 +41,6 @@ proc get_keys {l} { ...@@ -48,22 +41,6 @@ proc get_keys {l} {
return $res return $res
} }
proc cmp_hrandfield_result {hash_name expected_result} {
# Accumulate hrandfield results
unset -nocomplain myhash
array set myhash {}
for {set i 0} {$i < 100} {incr i} {
set key [r hrandfield $hash_name]
set myhash($key) 1
}
set res [lsort [array names myhash]]
if {$res eq $expected_result} {
return 1
} else {
return $res
}
}
proc dumpAllHashes {client} { proc dumpAllHashes {client} {
set keyAndFields(0,0) 0 set keyAndFields(0,0) 0
unset keyAndFields unset keyAndFields
...@@ -77,36 +54,6 @@ proc dumpAllHashes {client} { ...@@ -77,36 +54,6 @@ proc dumpAllHashes {client} {
return [array get keyAndFields] return [array get keyAndFields]
} }
proc hrandfieldTest {activeExpireConfig} {
r debug set-active-expire $activeExpireConfig
r del myhash
set contents {{field1 1} {field2 2} }
create_hash myhash $contents
set factorValgrind [expr {$::valgrind ? 2 : 1}]
# Set expiration time for field1 and field2 such that field1 expires first
r hpexpire myhash 1 NX FIELDS 1 field1
r hpexpire myhash 100 NX FIELDS 1 field2
# On call hrandfield command lazy expire deletes field1 first
wait_for_condition 8 10 {
[cmp_hrandfield_result myhash "field2"] == 1
} else {
fail "Expected field2 to be returned by HRANDFIELD."
}
# On call hrandfield command lazy expire deletes field2 as well
wait_for_condition 8 20 {
[cmp_hrandfield_result myhash "{}"] == 1
} else {
fail "Expected {} to be returned by HRANDFIELD."
}
# restore the default value
r debug set-active-expire 1
}
############################### TESTS ######################################### ############################### TESTS #########################################
start_server {tags {"external:skip needs:debug"}} { start_server {tags {"external:skip needs:debug"}} {
...@@ -117,15 +64,12 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -117,15 +64,12 @@ start_server {tags {"external:skip needs:debug"}} {
r config set hash-max-listpack-entries 512 r config set hash-max-listpack-entries 512
} }
test "HEXPIRE/HEXPIREAT/HPEXPIRE/HPEXPIREAT - Returns empty array if key does not exist" { test "HEXPIRE/HEXPIREAT/HPEXPIRE/HPEXPIREAT - Returns array if the key does not exist" {
r del myhash r del myhash
# Make sure we can distinguish between an empty array and a null response assert_equal [r HEXPIRE myhash 1000 FIELDS 1 a] [list $E_NO_FIELD]
r readraw 1 assert_equal [r HEXPIREAT myhash 1000 FIELDS 1 a] [list $E_NO_FIELD]
assert_equal {*0} [r HEXPIRE myhash 1000 FIELDS 1 a] assert_equal [r HPEXPIRE myhash 1000 FIELDS 2 a b] [list $E_NO_FIELD $E_NO_FIELD]
assert_equal {*0} [r HEXPIREAT myhash 1000 FIELDS 1 a] assert_equal [r HPEXPIREAT myhash 1000 FIELDS 2 a b] [list $E_NO_FIELD $E_NO_FIELD]
assert_equal {*0} [r HPEXPIRE myhash 1000 FIELDS 1 a]
assert_equal {*0} [r HPEXPIREAT myhash 1000 FIELDS 1 a]
r readraw 0
} }
test "HEXPIRE/HEXPIREAT/HPEXPIRE/HPEXPIREAT - Verify that the expire time does not overflow" { test "HEXPIRE/HEXPIREAT/HPEXPIRE/HPEXPIREAT - Verify that the expire time does not overflow" {
...@@ -205,15 +149,17 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -205,15 +149,17 @@ start_server {tags {"external:skip needs:debug"}} {
r del myhash r del myhash
r hset myhash f1 v1 r hset myhash f1 v1
assert_error {*Parameter `numFields` should be greater than 0} {r hpexpire myhash 1000 NX FIELDS 0 f1 f2 f3} assert_error {*Parameter `numFields` should be greater than 0} {r hpexpire myhash 1000 NX FIELDS 0 f1 f2 f3}
assert_error {*Parameter `numFields` is more than number of arguments} {r hpexpire myhash 1000 NX FIELDS 4 f1 f2 f3} # <count> not match with actual number of fields
assert_error {*parameter must match the number*} {r hpexpire myhash 1000 NX FIELDS 4 f1 f2 f3}
assert_error {*parameter must match the number*} {r hpexpire myhash 1000 NX FIELDS 2 f1 f2 f3}
} }
test "HPEXPIRE - parameter expire-time near limit of 2^48 ($type)" { test "HPEXPIRE - parameter expire-time near limit of 2^46 ($type)" {
r del myhash r del myhash
r hset myhash f1 v1 r hset myhash f1 v1
# below & above # below & above
assert_equal [r hpexpire myhash [expr (1<<48) - [clock milliseconds] - 1000 ] FIELDS 1 f1] [list $E_OK] assert_equal [r hpexpire myhash [expr (1<<46) - [clock milliseconds] - 1000 ] FIELDS 1 f1] [list $E_OK]
assert_error {*invalid expire time*} {r hpexpire myhash [expr (1<<48) - [clock milliseconds] + 100 ] FIELDS 1 f1} assert_error {*invalid expire time*} {r hpexpire myhash [expr (1<<46) - [clock milliseconds] + 100 ] FIELDS 1 f1}
} }
test "Lazy Expire - fields are lazy deleted ($type)" { test "Lazy Expire - fields are lazy deleted ($type)" {
...@@ -305,13 +251,10 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -305,13 +251,10 @@ start_server {tags {"external:skip needs:debug"}} {
r flushall async r flushall async
} }
test "HTTL/HPTTL - Returns empty array if key does not exist" { test "HTTL/HPTTL - Returns array if the key does not exist" {
r del myhash r del myhash
# Make sure we can distinguish between an empty array and a null response assert_equal [r HTTL myhash FIELDS 1 a] [list $T_NO_FIELD]
r readraw 1 assert_equal [r HPTTL myhash FIELDS 2 a b] [list $T_NO_FIELD $T_NO_FIELD]
assert_equal {*0} [r HTTL myhash FIELDS 1 a]
assert_equal {*0} [r HPTTL myhash FIELDS 1 a]
r readraw 0
} }
test "HTTL/HPTTL - Input validation gets failed on nonexists field or field without expire ($type)" { test "HTTL/HPTTL - Input validation gets failed on nonexists field or field without expire ($type)" {
...@@ -320,10 +263,10 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -320,10 +263,10 @@ start_server {tags {"external:skip needs:debug"}} {
r HPEXPIRE myhash 1000 NX FIELDS 1 field1 r HPEXPIRE myhash 1000 NX FIELDS 1 field1
foreach cmd {HTTL HPTTL} { foreach cmd {HTTL HPTTL} {
assert_equal [r $cmd non_exists_key FIELDS 1 f] {}
assert_equal [r $cmd myhash FIELDS 2 field2 non_exists_field] "$T_NO_EXPIRY $T_NO_FIELD" assert_equal [r $cmd myhash FIELDS 2 field2 non_exists_field] "$T_NO_EXPIRY $T_NO_FIELD"
# Set numFields less than actual number of fields. Fine. # <count> not match with actual number of fields
assert_equal [r $cmd myhash FIELDS 1 non_exists_field1 non_exists_field2] "$T_NO_FIELD" assert_error {*parameter must match the number*} {r $cmd myhash FIELDS 1 non_exists_field1 non_exists_field2}
assert_error {*parameter must match the number*} {r $cmd myhash FIELDS 3 non_exists_field1 non_exists_field2}
} }
} }
...@@ -337,13 +280,10 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -337,13 +280,10 @@ start_server {tags {"external:skip needs:debug"}} {
assert_range $ttl 1000 2000 assert_range $ttl 1000 2000
} }
test "HEXPIRETIME/HPEXPIRETIME - Returns empty array if key does not exist" { test "HEXPIRETIME/HPEXPIRETIME - Returns array if the key does not exist" {
r del myhash r del myhash
# Make sure we can distinguish between an empty array and a null response assert_equal [r HEXPIRETIME myhash FIELDS 1 a] [list $T_NO_FIELD]
r readraw 1 assert_equal [r HPEXPIRETIME myhash FIELDS 2 a b] [list $T_NO_FIELD $T_NO_FIELD]
assert_equal {*0} [r HEXPIRETIME myhash FIELDS 1 a]
assert_equal {*0} [r HPEXPIRETIME myhash FIELDS 1 a]
r readraw 0
} }
test "HEXPIRETIME - returns TTL in Unix timestamp ($type)" { test "HEXPIRETIME - returns TTL in Unix timestamp ($type)" {
...@@ -406,22 +346,33 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -406,22 +346,33 @@ start_server {tags {"external:skip needs:debug"}} {
r debug set-active-expire 1 r debug set-active-expire 1
} }
# OPEN: To decide if to delete expired fields at start of HRANDFIELD. test "Test HRANDFIELD deletes all expired fields ($type)" {
# test "Test HRANDFIELD does not return expired fields ($type)" {
# hrandfieldTest 0
# hrandfieldTest 1
# }
test "Test HRANDFIELD can return expired fields ($type)" {
r debug set-active-expire 0 r debug set-active-expire 0
r del myhash r flushall
r hset myhash f1 v1 f2 v2 f3 v3 f4 v4 f5 v5 r hset myhash f1 v1 f2 v2 f3 v3 f4 v4 f5 v5
r hpexpire myhash 1 NX FIELDS 4 f1 f2 f3 f4 r hpexpire myhash 1 FIELDS 2 f1 f2
after 5 after 5
set res [cmp_hrandfield_result myhash "f1 f2 f3 f4 f5"] assert_equal [lsort [r hrandfield myhash 5]] "f3 f4 f5"
assert {$res == 1} r hpexpire myhash 1 FIELDS 3 f3 f4 f5
r debug set-active-expire 1 after 5
assert_equal [lsort [r hrandfield myhash 5]] ""
assert_equal [r keys *] ""
r del myhash
r hset myhash f1 v1 f2 v2 f3 v3
r hpexpire myhash 1 FIELDS 1 f1
after 5
set res [r hrandfield myhash]
assert {$res == "f2" || $res == "f3"}
r hpexpire myhash 1 FIELDS 1 f2
after 5
assert_equal [lsort [r hrandfield myhash 5]] "f3"
r hpexpire myhash 1 FIELDS 1 f3
after 5
assert_equal [r hrandfield myhash] ""
assert_equal [r keys *] ""
r debug set-active-expire 1
} }
test "Lazy Expire - HLEN does count expired fields ($type)" { test "Lazy Expire - HLEN does count expired fields ($type)" {
...@@ -711,12 +662,10 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -711,12 +662,10 @@ start_server {tags {"external:skip needs:debug"}} {
r debug set-active-expire 1 r debug set-active-expire 1
} }
test "HPERSIST - Returns empty array if key does not exist ($type)" { test "HPERSIST - Returns array if the key does not exist ($type)" {
r del myhash r del myhash
# Make sure we can distinguish between an empty array and a null response assert_equal [r HPERSIST myhash FIELDS 1 a] [list $P_NO_FIELD]
r readraw 1 assert_equal [r HPERSIST myhash FIELDS 2 a b] [list $P_NO_FIELD $P_NO_FIELD]
assert_equal {*0} [r HPERSIST myhash FIELDS 1 a]
r readraw 0
} }
test "HPERSIST - input validation ($type)" { test "HPERSIST - input validation ($type)" {
...@@ -726,9 +675,11 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -726,9 +675,11 @@ start_server {tags {"external:skip needs:debug"}} {
r hexpire myhash 1000 NX FIELDS 1 f1 r hexpire myhash 1000 NX FIELDS 1 f1
assert_error {*wrong number of arguments*} {r hpersist myhash} assert_error {*wrong number of arguments*} {r hpersist myhash}
assert_error {*wrong number of arguments*} {r hpersist myhash FIELDS 1} assert_error {*wrong number of arguments*} {r hpersist myhash FIELDS 1}
assert_equal [r hpersist not-exists-key FIELDS 1 f1] {}
assert_equal [r hpersist myhash FIELDS 2 f1 not-exists-field] "$P_OK $P_NO_FIELD" assert_equal [r hpersist myhash FIELDS 2 f1 not-exists-field] "$P_OK $P_NO_FIELD"
assert_equal [r hpersist myhash FIELDS 1 f2] "$P_NO_EXPIRY" assert_equal [r hpersist myhash FIELDS 1 f2] "$P_NO_EXPIRY"
# <count> not match with actual number of fields
assert_error {*parameter must match the number*} {r hpersist myhash FIELDS 2 f1 f2 f3}
assert_error {*parameter must match the number*} {r hpersist myhash FIELDS 4 f1 f2 f3}
} }
test "HPERSIST - verify fields with TTL are persisted ($type)" { test "HPERSIST - verify fields with TTL are persisted ($type)" {
...@@ -969,14 +920,29 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -969,14 +920,29 @@ start_server {tags {"external:skip needs:debug"}} {
r config set hash-max-listpack-entries 512 r config set hash-max-listpack-entries 512
} }
test "Command rewrite and expired hash fields are propagated to replica ($type)" { test "Test Command propagated to replica as expected ($type)" {
start_server {overrides {appendonly {yes} appendfsync always} tags {external:skip}} { start_server {overrides {appendonly {yes} appendfsync always} tags {external:skip}} {
set aof [get_last_incr_aof_path r] set aof [get_last_incr_aof_path r]
# Time is in the past so it should propagate HDELs to replica
# and delete the fields
r hset h0 x1 y1 x2 y2
r hexpireat h0 1 fields 3 x1 x2 non_exists_field
r hset h1 f1 v1 f2 v2 r hset h1 f1 v1 f2 v2
# Next command won't be propagated to replica
# because XX condition not met or field not exists
r hexpire h1 10 XX FIELDS 3 f1 f2 non_exists_field
r hpexpire h1 20 FIELDS 1 f1 r hpexpire h1 20 FIELDS 1 f1
r hpexpire h1 30 FIELDS 1 f2
# Next command will be propagate with only field 'f2'
# because NX condition not met for field 'f1'
r hpexpire h1 30 NX FIELDS 2 f1 f2
# Non exists field should be ignored
r hpexpire h1 30 FIELDS 1 non_exists_field r hpexpire h1 30 FIELDS 1 non_exists_field
r hset h2 f1 v1 f2 v2 f3 v3 f4 v4 r hset h2 f1 v1 f2 v2 f3 v3 f4 v4
r hpexpire h2 40 FIELDS 2 f1 non_exists_field r hpexpire h2 40 FIELDS 2 f1 non_exists_field
...@@ -993,11 +959,16 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -993,11 +959,16 @@ start_server {tags {"external:skip needs:debug"}} {
# Assert that each TTL-related command are persisted with absolute timestamps in AOF # Assert that each TTL-related command are persisted with absolute timestamps in AOF
assert_aof_content $aof { assert_aof_content $aof {
{select *} {select *}
{hset h0 x1 y1 x2 y2}
{multi}
{hdel h0 x1}
{hdel h0 x2}
{exec}
{hset h1 f1 v1 f2 v2} {hset h1 f1 v1 f2 v2}
{hpexpireat h1 * FIELDS 1 f1} {hpexpireat h1 * FIELDS 1 f1}
{hpexpireat h1 * FIELDS 1 f2} {hpexpireat h1 * FIELDS 1 f2}
{hset h2 f1 v1 f2 v2 f3 v3 f4 v4} {hset h2 f1 v1 f2 v2 f3 v3 f4 v4}
{hpexpireat h2 * FIELDS 2 f1 non_exists_field} {hpexpireat h2 * FIELDS 1 f1}
{hpexpireat h2 * FIELDS 1 f2} {hpexpireat h2 * FIELDS 1 f2}
{hpexpireat h2 * FIELDS 1 f3} {hpexpireat h2 * FIELDS 1 f3}
{hpexpireat h2 * FIELDS 1 f4} {hpexpireat h2 * FIELDS 1 f4}
...@@ -1067,7 +1038,13 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -1067,7 +1038,13 @@ start_server {tags {"external:skip needs:debug"}} {
r hpexpireat h1 [expr [clock seconds]*1000+100000] NX FIELDS 1 f2 r hpexpireat h1 [expr [clock seconds]*1000+100000] NX FIELDS 1 f2
r hpexpire h1 100000 NX FIELDS 3 f3 f4 f5 r hpexpire h1 100000 NX FIELDS 3 f3 f4 f5
r hexpire h1 100000 FIELDS 1 f6 r hexpire h1 100000 FIELDS 1 f6
r hset h5 f1 v1
# Verify HRANDFIELD deletes expired fields and propagates it
r hset h2 f1 v1 f2 v2
r hpexpire h2 1 FIELDS 2 f1 f2
after 5
assert_equal [r hrandfield h4 2] ""
after 200
assert_aof_content $aof { assert_aof_content $aof {
{select *} {select *}
...@@ -1076,7 +1053,10 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -1076,7 +1053,10 @@ start_server {tags {"external:skip needs:debug"}} {
{hpexpireat h1 * FIELDS 1 f2} {hpexpireat h1 * FIELDS 1 f2}
{hpexpireat h1 * NX FIELDS 3 f3 f4 f5} {hpexpireat h1 * NX FIELDS 3 f3 f4 f5}
{hpexpireat h1 * FIELDS 1 f6} {hpexpireat h1 * FIELDS 1 f6}
{hset h5 f1 v1} {hset h2 f1 v1 f2 v2}
{hpexpireat h2 * FIELDS 2 f1 f2}
{hdel h2 *}
{hdel h2 *}
} }
array set keyAndFields1 [dumpAllHashes r] array set keyAndFields1 [dumpAllHashes r]
...@@ -1099,12 +1079,16 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -1099,12 +1079,16 @@ start_server {tags {"external:skip needs:debug"}} {
r flushall ; # Clean up keyspace to avoid interference by keys from other tests r flushall ; # Clean up keyspace to avoid interference by keys from other tests
set repl [attach_to_replication_stream] set repl [attach_to_replication_stream]
# HEXPIRE/HPEXPIRE should be translated into HPEXPIREAT
r hset h1 f1 v1 r hset h1 f1 v1
r hexpireat h1 [expr [clock seconds]+100] NX FIELDS 1 f1 r hexpireat h1 [expr [clock seconds]+100] NX FIELDS 1 f1
r hset h2 f2 v2 r hset h2 f2 v2
r hpexpireat h2 [expr [clock seconds]*1000+100000] NX FIELDS 1 f2 r hpexpireat h2 [expr [clock seconds]*1000+100000] NX FIELDS 1 f2
r hset h3 f3 v3 f4 v4 r hset h3 f3 v3 f4 v4 f5 v5
# hpersist does nothing here. Verify it is not propagated.
r hpersist h3 FIELDS 1 f5
r hexpire h3 100 FIELDS 3 f3 f4 non_exists_field r hexpire h3 100 FIELDS 3 f3 f4 non_exists_field
r hpersist h3 FIELDS 1 f3
assert_replication_stream $repl { assert_replication_stream $repl {
{select *} {select *}
...@@ -1112,12 +1096,55 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -1112,12 +1096,55 @@ start_server {tags {"external:skip needs:debug"}} {
{hpexpireat h1 * NX FIELDS 1 f1} {hpexpireat h1 * NX FIELDS 1 f1}
{hset h2 f2 v2} {hset h2 f2 v2}
{hpexpireat h2 * NX FIELDS 1 f2} {hpexpireat h2 * NX FIELDS 1 f2}
{hset h3 f3 v3 f4 v4} {hset h3 f3 v3 f4 v4 f5 v5}
{hpexpireat h3 * FIELDS 3 f3 f4 non_exists_field} {hpexpireat h3 * FIELDS 2 f3 f4}
{hpersist h3 FIELDS 1 f3}
} }
close_replication_stream $repl close_replication_stream $repl
} {} {needs:repl} } {} {needs:repl}
test {HRANDFIELD delete expired fields and propagate DELs to replica} {
r debug set-active-expire 0
r flushall
set repl [attach_to_replication_stream]
# HRANDFIELD delete expired fields and propagate MULTI-EXEC DELs. Reply none.
r hset h1 f1 v1 f2 v2
r hpexpire h1 1 FIELDS 2 f1 f2
after 5
assert_equal [r hrandfield h1 2] ""
# HRANDFIELD delete expired field and propagate DEL. Reply non-expired field.
r hset h2 f1 v1 f2 v2
r hpexpire h2 1 FIELDS 1 f1
after 5
assert_equal [r hrandfield h2 2] "f2"
# HRANDFIELD delete expired field and propagate DEL. Reply none.
r hset h3 f1 v1
r hpexpire h3 1 FIELDS 1 f1
after 5
assert_equal [r hrandfield h3 2] ""
assert_replication_stream $repl {
{select *}
{hset h1 f1 v1 f2 v2}
{hpexpireat h1 * FIELDS 2 f1 f2}
{multi}
{hdel h1 *}
{hdel h1 *}
{exec}
{hset h2 f1 v1 f2 v2}
{hpexpireat h2 * FIELDS 1 f1}
{hdel h2 f1}
{hset h3 f1 v1}
{hpexpireat h3 * FIELDS 1 f1}
{hdel h3 f1}
}
close_replication_stream $repl
r debug set-active-expire 1
} {OK} {needs:repl}
# Start another server to test replication of TTLs # Start another server to test replication of TTLs
start_server {tags {needs:repl external:skip}} { start_server {tags {needs:repl external:skip}} {
# Set the outer layer server as primary # Set the outer layer server as primary
...@@ -1161,4 +1188,3 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -1161,4 +1188,3 @@ start_server {tags {"external:skip needs:debug"}} {
} }
} }
} }
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