Unverified Commit ce121b92 authored by Moti Cohen's avatar Moti Cohen Committed by GitHub
Browse files

HFE - Avoid lazy expire if called by modules + cleanup (#13326)

Need to be carefull if called by modules since modules API allow to open
and close key handler. We don't want to invalidate the handler
underneath.

* hashTypeExists(), hashTypeGetValueObject() - will return the logical
state of the field. A flag will indicate noExpire.
* RM_HashGet() - Will get NULL if the field expired. Fields won’t be
deleted.
* RM_ScanKey() - might return 0 items if all fields got expired. Fields
won’t be deleted.
* RM_HashSet() - If set, then override expired field. If delete, we can
either delete or leave it to active-expiration. XX/NX - logically
correct (Verify with tests).

Nice to have (not implemented):
* RedisModule_CloseKey() - We can local active-expire up-to 100 items. 

Note:
Length will be wrong to modules just like redis (Count expired fields).
parent 690ef363
...@@ -5271,10 +5271,21 @@ int RM_HashSet(RedisModuleKey *key, int flags, ...) { ...@@ -5271,10 +5271,21 @@ int RM_HashSet(RedisModuleKey *key, int flags, ...) {
   
/* Handle XX and NX */ /* Handle XX and NX */
if (flags & (REDISMODULE_HASH_XX|REDISMODULE_HASH_NX)) { if (flags & (REDISMODULE_HASH_XX|REDISMODULE_HASH_NX)) {
int isHashDeleted; int hfeFlags = HFE_LAZY_AVOID_HASH_DEL; /* Avoid invalidate the key */
int exists = hashTypeExists(key->db, key->value, field->ptr, &isHashDeleted);
/* hash-field-expiration is not exposed to modules */ /*
serverAssert(isHashDeleted == 0); * The hash might contain expired fields. If we lazily delete expired
* field and the command was sent with XX flag, the operation could
* fail and leave the hash empty, which the caller might not expect.
* To prevent unexpected behavior, we avoid lazy deletion in this case
* yet let the operation fail. Note that moduleDelKeyIfEmpty()
* below won't delete the hash if it left with single expired key
* because hash counts blindly expired fields as well.
*/
if (flags & REDISMODULE_HASH_XX)
hfeFlags |= HFE_LAZY_AVOID_FIELD_DEL;
int exists = hashTypeExists(key->db, key->value, field->ptr, hfeFlags, NULL);
if (((flags & REDISMODULE_HASH_XX) && !exists) || if (((flags & REDISMODULE_HASH_XX) && !exists) ||
((flags & REDISMODULE_HASH_NX) && exists)) ((flags & REDISMODULE_HASH_NX) && exists))
{ {
...@@ -5357,6 +5368,7 @@ int RM_HashSet(RedisModuleKey *key, int flags, ...) { ...@@ -5357,6 +5368,7 @@ int RM_HashSet(RedisModuleKey *key, int flags, ...) {
* RedisModule_FreeString(), or by enabling automatic memory management. * RedisModule_FreeString(), or by enabling automatic memory management.
*/ */
int RM_HashGet(RedisModuleKey *key, int flags, ...) { int RM_HashGet(RedisModuleKey *key, int flags, ...) {
int hfeFlags = HFE_LAZY_AVOID_FIELD_DEL | HFE_LAZY_AVOID_HASH_DEL;
va_list ap; va_list ap;
if (key->value && key->value->type != OBJ_HASH) return REDISMODULE_ERR; if (key->value && key->value->type != OBJ_HASH) return REDISMODULE_ERR;
   
...@@ -5378,21 +5390,16 @@ int RM_HashGet(RedisModuleKey *key, int flags, ...) { ...@@ -5378,21 +5390,16 @@ int RM_HashGet(RedisModuleKey *key, int flags, ...) {
if (flags & REDISMODULE_HASH_EXISTS) { if (flags & REDISMODULE_HASH_EXISTS) {
existsptr = va_arg(ap,int*); existsptr = va_arg(ap,int*);
if (key->value) { if (key->value) {
int isHashDeleted; *existsptr = hashTypeExists(key->db, key->value, field->ptr, hfeFlags, NULL);
*existsptr = hashTypeExists(key->db, key->value, field->ptr, &isHashDeleted);
/* hash-field-expiration is not exposed to modules */
serverAssert(isHashDeleted == 0);
} else { } else {
*existsptr = 0; *existsptr = 0;
} }
} else { } else {
int isHashDeleted;
valueptr = va_arg(ap,RedisModuleString**); valueptr = va_arg(ap,RedisModuleString**);
if (key->value) { if (key->value) {
*valueptr = hashTypeGetValueObject(key->db,key->value,field->ptr, &isHashDeleted); *valueptr = hashTypeGetValueObject(key->db, key->value, field->ptr,
hfeFlags, NULL);
   
/* Currently hash-field-expiration is not exposed to modules */
serverAssert(isHashDeleted == 0);
if (*valueptr) { if (*valueptr) {
robj *decoded = getDecodedObject(*valueptr); robj *decoded = getDecodedObject(*valueptr);
decrRefCount(*valueptr); decrRefCount(*valueptr);
...@@ -11080,6 +11087,11 @@ static void moduleScanKeyCallback(void *privdata, const dictEntry *de) { ...@@ -11080,6 +11087,11 @@ static void moduleScanKeyCallback(void *privdata, const dictEntry *de) {
value = NULL; value = NULL;
} else if (o->type == OBJ_HASH) { } else if (o->type == OBJ_HASH) {
sds val = dictGetVal(de); sds val = dictGetVal(de);
/* If field is expired, then ignore */
if (hfieldIsExpired(key))
return;
field = createStringObject(key, hfieldlen(key)); field = createStringObject(key, hfieldlen(key));
value = createStringObject(val, sdslen(val)); value = createStringObject(val, sdslen(val));
} else if (o->type == OBJ_ZSET) { } else if (o->type == OBJ_ZSET) {
...@@ -11189,9 +11201,8 @@ int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleSc ...@@ -11189,9 +11201,8 @@ int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleSc
ret = 0; ret = 0;
} else if (o->type == OBJ_ZSET || o->type == OBJ_HASH) { } else if (o->type == OBJ_ZSET || o->type == OBJ_HASH) {
unsigned char *lp, *p; unsigned char *lp, *p;
unsigned char *vstr; /* is hash with expiry on fields, then lp tuples are [field][value][expire] */
unsigned int vlen; int hfe = o->type == OBJ_HASH && o->encoding == OBJ_ENCODING_LISTPACK_EX;
long long vll;
   
if (o->type == OBJ_HASH) if (o->type == OBJ_HASH)
lp = hashTypeListpackGetLp(o); lp = hashTypeListpackGetLp(o);
...@@ -11200,19 +11211,32 @@ int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleSc ...@@ -11200,19 +11211,32 @@ int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleSc
   
p = lpSeek(lp,0); p = lpSeek(lp,0);
while(p) { while(p) {
vstr = lpGetValue(p,&vlen,&vll); long long vllField, vllValue, vllExpire;
robj *field = (vstr != NULL) ? unsigned int lenField, lenValue;
createStringObject((char*)vstr,vlen) : unsigned char *pField, *pValue;
createStringObjectFromLongLongWithSds(vll);
pField = lpGetValue(p,&lenField,&vllField);
p = lpNext(lp,p); p = lpNext(lp,p);
vstr = lpGetValue(p,&vlen,&vll); pValue = lpGetValue(p,&lenValue,&vllValue);
robj *value = (vstr != NULL) ?
createStringObject((char*)vstr,vlen) :
createStringObjectFromLongLongWithSds(vll);
fn(key, field, value, privdata);
p = lpNext(lp,p); p = lpNext(lp,p);
if (o->type == OBJ_HASH && o->encoding == OBJ_ENCODING_LISTPACK_EX)
p = lpNext(lp, p); /* Skip expire time */ if (hfe) {
serverAssert(lpGetIntegerValue(p, &vllExpire));
p = lpNext(lp, p);
/* Skip expired fields */
if (hashTypeIsExpired(o, vllExpire))
continue;
}
robj *value = (pValue != NULL) ?
createStringObject((char*)pValue,lenValue) :
createStringObjectFromLongLongWithSds(vllValue);
robj *field = (pField != NULL) ?
createStringObject((char*)pField,lenField) :
createStringObjectFromLongLongWithSds(vllField);
fn(key, field, value, privdata);
   
decrRefCount(field); decrRefCount(field);
decrRefCount(value); decrRefCount(value);
...@@ -11225,7 +11249,6 @@ int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleSc ...@@ -11225,7 +11249,6 @@ int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleSc
return ret; return ret;
} }
   
/* -------------------------------------------------------------------------- /* --------------------------------------------------------------------------
* ## Module fork API * ## Module fork API
* -------------------------------------------------------------------------- */ * -------------------------------------------------------------------------- */
......
...@@ -3191,9 +3191,14 @@ typedef struct dictExpireMetadata { ...@@ -3191,9 +3191,14 @@ 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 */
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,7 +3215,7 @@ void hashTypeCurrentObject(hashTypeIterator *hi, int what, unsigned char **vstr, ...@@ -3210,7 +3215,7 @@ 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);
......
...@@ -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;
......
This diff is collapsed.
...@@ -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]
} }
......
...@@ -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
......
...@@ -1103,8 +1103,11 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -1103,8 +1103,11 @@ start_server {tags {"external:skip needs:debug"}} {
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,8 +1115,9 @@ start_server {tags {"external:skip needs:debug"}} { ...@@ -1112,8 +1115,9 @@ 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 3 f3 f4 non_exists_field}
{hpersist h3 FIELDS 1 f3}
} }
close_replication_stream $repl close_replication_stream $repl
} {} {needs:repl} } {} {needs:repl}
......
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