/* * Copyright (c) 2009-Present, Redis Ltd. * All rights reserved. * * Licensed under your choice of the Redis Source Available License 2.0 * (RSALv2) or the Server Side Public License v1 (SSPLv1). */ #include "server.h" #include "ebuckets.h" #include /* Threshold for HEXPIRE and HPERSIST to be considered whether it is worth to * update the expiration time of the hash object in global HFE DS. */ #define HASH_NEW_EXPIRE_DIFF_THRESHOLD max(4000, 1<> 2) typedef enum GetFieldRes { /* common (Used by hashTypeGet* value family) */ GETF_OK = 0, /* The field was found. */ GETF_NOT_FOUND, /* The field was not found. */ GETF_EXPIRED, /* Logically expired (Might be lazy deleted or not) */ GETF_EXPIRED_HASH, /* Delete hash since retrieved field was expired and * it was the last field in the hash. */ } 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 */ static ExpireAction onFieldExpire(eItem item, void *ctx); static ExpireMeta* hfieldGetExpireMeta(const eItem field); static ExpireMeta *hashGetExpireMeta(const eItem hash); static void hexpireGenericCommand(client *c, const char *cmd, long long basetime, int unit); 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 propagateHashFieldDeletion(redisDb *db, sds key, char *field, size_t fieldLen); /* hash dictType funcs */ static int dictHfieldKeyCompare(dict *d, const void *key1, const void *key2); static uint64_t dictMstrHash(const void *key); static void dictHfieldDestructor(dict *d, void *field); static size_t hashDictWithExpireMetadataBytes(dict *d); static void hashDictWithExpireOnRelease(dict *d); static robj* hashTypeLookupWriteOrCreate(client *c, robj *key); /*----------------------------------------------------------------------------- * Define dictType of hash * * - Stores fields as mstr strings with optional metadata to attach TTL * - Note that small hashes are represented with listpacks * - Once expiration is set for a field, the dict instance and corresponding * dictType are replaced with a dict containing metadata for Hash Field * Expiration (HFE) and using dictType `mstrHashDictTypeWithHFE` *----------------------------------------------------------------------------*/ dictType mstrHashDictType = { dictSdsHash, /* lookup hash function */ NULL, /* key dup */ NULL, /* val dup */ dictSdsMstrKeyCompare, /* lookup key compare */ dictHfieldDestructor, /* key destructor */ dictSdsDestructor, /* val destructor */ .storedHashFunction = dictMstrHash, /* stored hash function */ .storedKeyCompare = dictHfieldKeyCompare, /* stored key compare */ }; /* Define alternative dictType of hash with hash-field expiration (HFE) support */ dictType mstrHashDictTypeWithHFE = { dictSdsHash, /* lookup hash function */ NULL, /* key dup */ NULL, /* val dup */ dictSdsMstrKeyCompare, /* lookup key compare */ dictHfieldDestructor, /* key destructor */ dictSdsDestructor, /* val destructor */ .storedHashFunction = dictMstrHash, /* stored hash function */ .storedKeyCompare = dictHfieldKeyCompare, /* stored key compare */ .dictMetadataBytes = hashDictWithExpireMetadataBytes, .onDictRelease = hashDictWithExpireOnRelease, }; /*----------------------------------------------------------------------------- * Hash Field Expiration (HFE) Feature * * Each hash instance maintains its own set of hash field expiration within its * private ebuckets DS. In order to support HFE active expire cycle across hash * instances, hashes with associated HFE will be also registered in a global * ebuckets DS with expiration time value that reflects their next minimum * time to expire. The global HFE Active expiration will be triggered from * activeExpireCycle() function and will invoke "local" HFE Active expiration * for each hash instance that has expired fields. * * hashExpireBucketsType - ebuckets-type to be used at the global space * (db->hexpires) to register hashes that have one or more fields with time-Expiration. * The hashes will be registered in with the expiration time of the earliest field * in the hash. *----------------------------------------------------------------------------*/ EbucketsType hashExpireBucketsType = { .onDeleteItem = NULL, .getExpireMeta = hashGetExpireMeta, /* get ExpireMeta attached to each hash */ .itemsAddrAreOdd = 0, /* Addresses of dict are even */ }; /* dictExpireMetadata - ebuckets-type for hash fields with time-Expiration. ebuckets * instance Will be attached to each hash that has at least one field with expiry * time. */ EbucketsType hashFieldExpireBucketsType = { .onDeleteItem = NULL, .getExpireMeta = hfieldGetExpireMeta, /* get ExpireMeta attached to each field */ .itemsAddrAreOdd = 1, /* Addresses of hfield (mstr) are odd!! */ }; /* OnFieldExpireCtx passed to OnFieldExpire() */ typedef struct OnFieldExpireCtx { robj *hashObj; redisDb *db; } OnFieldExpireCtx; /* The implementation of hashes by dict was modified from storing fields as sds * strings to store "mstr" (Immutable string with metadata) in order to be able to * attach TTL (ExpireMeta) to the hash-field. This usage of mstr opens up the * opportunity for future features to attach additional metadata by need to the * fields. * * The following defines new hfield kind of mstr */ typedef enum HfieldMetaFlags { HFIELD_META_EXPIRE = 0, } HfieldMetaFlags; mstrKind mstrFieldKind = { .name = "hField", /* Taking care that all metaSize[*] values are even ensures that all * addresses of hfield instances will be odd. */ .metaSize[HFIELD_META_EXPIRE] = sizeof(ExpireMeta), }; static_assert(sizeof(struct ExpireMeta ) % 2 == 0, "must be even!"); /* Used by hpersistCommand() */ typedef enum SetPersistRes { HFE_PERSIST_NO_FIELD = -2, /* No such hash-field */ HFE_PERSIST_NO_TTL = -1, /* No TTL attached to the field */ HFE_PERSIST_OK = 1 } SetPersistRes; static inline int isDictWithMetaHFE(dict *d) { return d->type == &mstrHashDictTypeWithHFE; } /*----------------------------------------------------------------------------- * setex* - Set field's expiration * * Setting expiration time to fields might be time-consuming and complex since * each update of expiration time, not only updates `ebuckets` of corresponding * hash, but also might update `ebuckets` of global HFE DS. It is required to opt * sequence of field updates with expirartion for a given hash, such that only * once done, the global HFE DS will get updated. * * To do so, follow the scheme: * 1. Call hashTypeSetExInit() to initialize the HashTypeSetEx struct. * 2. Call hashTypeSetEx() one time or more, for each field/expiration update. * 3. Call hashTypeSetExDone() for notification and update of global HFE. *----------------------------------------------------------------------------*/ /* Returned value of hashTypeSetEx() */ typedef enum SetExRes { HSETEX_OK = 1, /* Expiration time set/updated as expected */ HSETEX_NO_FIELD = -2, /* No such hash-field */ 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 */ } SetExRes; /* Used by httlGenericCommand() */ typedef enum GetExpireTimeRes { HFE_GET_NO_FIELD = -2, /* No such hash-field */ HFE_GET_NO_TTL = -1, /* No TTL attached to the field */ } GetExpireTimeRes; typedef enum ExpireSetCond { HFE_NX = 1<<0, HFE_XX = 1<<1, HFE_GT = 1<<2, HFE_LT = 1<<3 } ExpireSetCond; /* Used by hashTypeSetEx() for setting fields or their expiry */ typedef struct HashTypeSetEx { /*** config ***/ ExpireSetCond expireSetCond; /* [XX | NX | GT | LT] */ /*** metadata ***/ uint64_t minExpire; /* if uninit EB_EXPIRE_TIME_INVALID */ redisDb *db; robj *key, *hashObj; uint64_t minExpireFields; /* Trace updated fields and their previous/new * minimum expiration time. If minimum recorded * is above minExpire of the hash, then we don't * have to update global HFE DS */ int fieldDeleted; /* Number of fields deleted */ int fieldUpdated; /* Number of fields updated */ /* Optionally provide client for notification */ client *c; const char *cmd; } HashTypeSetEx; int hashTypeSetExInit(robj *key, robj *o, client *c, redisDb *db, const char *cmd, ExpireSetCond expireSetCond, HashTypeSetEx *ex); SetExRes hashTypeSetEx(robj *o, sds field, uint64_t expireAt, HashTypeSetEx *exInfo); void hashTypeSetExDone(HashTypeSetEx *e); /*----------------------------------------------------------------------------- * Accessor functions for dictType of hash *----------------------------------------------------------------------------*/ static int dictHfieldKeyCompare(dict *d, const void *key1, const void *key2) { int l1,l2; UNUSED(d); l1 = hfieldlen((hfield)key1); l2 = hfieldlen((hfield)key2); if (l1 != l2) return 0; return memcmp(key1, key2, l1) == 0; } static uint64_t dictMstrHash(const void *key) { return dictGenHashFunction((unsigned char*)key, mstrlen((char*)key)); } static void dictHfieldDestructor(dict *d, void *field) { /* If attached TTL to the field, then remove it from hash's private ebuckets. */ if (hfieldGetExpireTime(field) != EB_EXPIRE_TIME_INVALID) { dictExpireMetadata *dictExpireMeta = (dictExpireMetadata *) dictMetadata(d); ebRemove(&dictExpireMeta->hfe, &hashFieldExpireBucketsType, field); } hfieldFree(field); /* Don't have to update global HFE DS. It's unnecessary. Implementing this * would introduce significant complexity and overhead for an operation that * isn't critical. In the worst case scenario, the hash will be efficiently * updated later by an active-expire operation, or it will be removed by the * hash's dbGenericDelete() function. */ } static size_t hashDictWithExpireMetadataBytes(dict *d) { UNUSED(d); /* expireMeta of the hash, ref to ebuckets and pointer to hash's key */ return sizeof(dictExpireMetadata); } static void hashDictWithExpireOnRelease(dict *d) { /* for sure allocated with metadata. Otherwise, this func won't be registered */ dictExpireMetadata *dictExpireMeta = (dictExpireMetadata *) dictMetadata(d); ebDestroy(&dictExpireMeta->hfe, &hashFieldExpireBucketsType, NULL); } /*----------------------------------------------------------------------------- * listpackEx functions *----------------------------------------------------------------------------*/ /* * If any of hash field expiration command is called on a listpack hash object * for the first time, we convert it to OBJ_ENCODING_LISTPACK_EX encoding. * We allocate "struct listpackEx" which holds listpack pointer and metadata to * register key to the global DS. In the listpack, we append another TTL entry * for each field-value pair. From now on, listpack will have triplets in it: * field-value-ttl. If TTL is not set for a field, we store 'zero' as the TTL * value. 'zero' is encoded as two bytes in the listpack. Memory overhead of a * non-existing TTL will be two bytes per field. * * Fields in the listpack will be ordered by TTL. Field with the smallest expiry * time will be the first item. Fields without TTL will be at the end of the * listpack. This way, it is easier/faster to find expired items. */ #define HASH_LP_NO_TTL 0 struct listpackEx *listpackExCreate(void) { listpackEx *lpt = zcalloc(sizeof(*lpt)); lpt->meta.trash = 1; lpt->lp = NULL; lpt->key = NULL; return lpt; } static void listpackExFree(listpackEx *lpt) { lpFree(lpt->lp); zfree(lpt); } struct lpFingArgs { uint64_t max_to_search; /* [in] Max number of tuples to search */ uint64_t expire_time; /* [in] Find the tuple that has a TTL larger than expire_time */ unsigned char *p; /* [out] First item of the tuple that has a TTL larger than expire_time */ int expired; /* [out] Number of tuples that have TTLs less than expire_time */ int index; /* Internally used */ unsigned char *fptr; /* Internally used, temp ptr */ }; /* Callback for lpFindCb(). Used to find number of expired fields as part of * active expiry or when trying to find the position for the new field according * to its expiry time.*/ static int cbFindInListpack(const unsigned char *lp, unsigned char *p, void *user, unsigned char *s, long long slen) { (void) lp; struct lpFingArgs *r = user; r->index++; if (r->max_to_search == 0) return 0; /* Break the loop and return */ if (r->index % 3 == 1) { r->fptr = p; /* First item of the tuple. */ } else if (r->index % 3 == 0) { serverAssert(!s); /* Third item of a tuple is expiry time */ if (slen == HASH_LP_NO_TTL || (uint64_t) slen >= r->expire_time) { r->p = r->fptr; return 0; /* Break the loop and return */ } r->expired++; r->max_to_search--; } return 1; } /* Returns number of expired fields. */ static uint64_t listpackExExpireDryRun(const robj *o) { serverAssert(o->encoding == OBJ_ENCODING_LISTPACK_EX); listpackEx *lpt = o->ptr; struct lpFingArgs r = { .max_to_search = UINT64_MAX, .expire_time = commandTimeSnapshot(), }; lpFindCb(lpt->lp, NULL, &r, cbFindInListpack, 0); return r.expired; } /* Returns the expiration time of the item with the nearest expiration. */ static uint64_t listpackExGetMinExpire(robj *o) { serverAssert(o->encoding == OBJ_ENCODING_LISTPACK_EX); long long expireAt; unsigned char *fptr; listpackEx *lpt = o->ptr; /* As fields are ordered by expire time, first field will have the smallest * expiry time. Third element is the expiry time of the first field */ fptr = lpSeek(lpt->lp, 2); if (fptr != NULL) { serverAssert(lpGetIntegerValue(fptr, &expireAt)); /* Check if this is a non-volatile field. */ if (expireAt != HASH_LP_NO_TTL) return expireAt; } return EB_EXPIRE_TIME_INVALID; } /* Walk over fields and delete the expired ones. */ void listpackExExpire(redisDb *db, robj *o, ExpireInfo *info) { serverAssert(o->encoding == OBJ_ENCODING_LISTPACK_EX); uint64_t expired = 0, min = EB_EXPIRE_TIME_INVALID; unsigned char *ptr; listpackEx *lpt = o->ptr; ptr = lpFirst(lpt->lp); while (ptr != NULL && (info->itemsExpired < info->maxToExpire)) { long long val; int64_t flen; unsigned char intbuf[LP_INTBUF_SIZE], *fref; fref = lpGet(ptr, &flen, intbuf); ptr = lpNext(lpt->lp, ptr); serverAssert(ptr); ptr = lpNext(lpt->lp, ptr); serverAssert(ptr && lpGetIntegerValue(ptr, &val)); /* Fields are ordered by expiry time. If we reached to a non-expired * or a non-volatile field, we know rest is not yet expired. */ if (val == HASH_LP_NO_TTL || (uint64_t) val > info->now) break; propagateHashFieldDeletion(db, ((listpackEx *) o->ptr)->key, (char *)((fref) ? fref : intbuf), flen); server.stat_expired_subkeys++; ptr = lpNext(lpt->lp, ptr); info->itemsExpired++; expired++; } if (expired) lpt->lp = lpDeleteRange(lpt->lp, 0, expired * 3); min = hashTypeGetMinExpire(o, 1 /*accurate*/); info->nextExpireTime = min; } static void listpackExAddInternal(robj *o, listpackEntry ent[3]) { listpackEx *lpt = o->ptr; /* Shortcut, just append at the end if this is a non-volatile field. */ if (ent[2].lval == HASH_LP_NO_TTL) { lpt->lp = lpBatchAppend(lpt->lp, ent, 3); return; } struct lpFingArgs r = { .max_to_search = UINT64_MAX, .expire_time = ent[2].lval, }; /* Check if there is a field with a larger TTL. */ lpFindCb(lpt->lp, NULL, &r, cbFindInListpack, 0); /* If list is empty or there is no field with a larger TTL, result will be * NULL. Otherwise, just insert before the found item.*/ if (r.p) lpt->lp = lpBatchInsert(lpt->lp, r.p, LP_BEFORE, ent, 3, NULL); else lpt->lp = lpBatchAppend(lpt->lp, ent, 3); } /* Add new field ordered by expire time. */ void listpackExAddNew(robj *o, char *field, size_t flen, char *value, size_t vlen, uint64_t expireAt) { listpackEntry ent[3] = { {.sval = (unsigned char*) field, .slen = flen}, {.sval = (unsigned char*) value, .slen = vlen}, {.lval = expireAt} }; listpackExAddInternal(o, ent); } /* If expiry time is changed, this function will place field into the correct * position. First, it deletes the field and re-inserts to the listpack ordered * by expiry time. */ static void listpackExUpdateExpiry(robj *o, sds field, unsigned char *fptr, unsigned char *vptr, uint64_t expire_at) { unsigned int slen = 0; long long val = 0; unsigned char tmp[512] = {0}; unsigned char *valstr; sds tmpval = NULL; listpackEx *lpt = o->ptr; /* Copy value */ valstr = lpGetValue(vptr, &slen, &val); if (valstr) { /* Normally, item length in the listpack is limited by * 'hash-max-listpack-value' config. It is unlikely, but it might be * larger than sizeof(tmp). */ if (slen > sizeof(tmp)) tmpval = sdsnewlen(valstr, slen); else memcpy(tmp, valstr, slen); } /* Delete field name, value and expiry time */ lpt->lp = lpDeleteRangeWithEntry(lpt->lp, &fptr, 3); listpackEntry ent[3] = {{0}}; ent[0].sval = (unsigned char*) field; ent[0].slen = sdslen(field); if (valstr) { ent[1].sval = tmpval ? (unsigned char *) tmpval : tmp; ent[1].slen = slen; } else { ent[1].lval = val; } ent[2].lval = expire_at; listpackExAddInternal(o, ent); sdsfree(tmpval); } /* Update field expire time. */ SetExRes hashTypeSetExpiryListpack(HashTypeSetEx *ex, sds field, unsigned char *fptr, unsigned char *vptr, unsigned char *tptr, uint64_t expireAt) { long long expireTime; uint64_t prevExpire = EB_EXPIRE_TIME_INVALID; serverAssert(lpGetIntegerValue(tptr, &expireTime)); if (expireTime != HASH_LP_NO_TTL) { prevExpire = (uint64_t) expireTime; } if (prevExpire == EB_EXPIRE_TIME_INVALID) { /* For fields without expiry, LT condition is considered valid */ if (ex->expireSetCond & (HFE_XX | HFE_GT)) return HSETEX_NO_CONDITION_MET; } else { if (((ex->expireSetCond == HFE_GT) && (prevExpire >= expireAt)) || ((ex->expireSetCond == HFE_LT) && (prevExpire <= expireAt)) || (ex->expireSetCond == HFE_NX) ) return HSETEX_NO_CONDITION_MET; /* Track of minimum expiration time (only later update global HFE DS) */ if (ex->minExpireFields > prevExpire) ex->minExpireFields = prevExpire; } /* If expired, then delete the field and propagate the deletion. * If replica, continue like the field is valid */ if (unlikely(checkAlreadyExpired(expireAt))) { propagateHashFieldDeletion(ex->db, ex->key->ptr, field, sdslen(field)); hashTypeDelete(ex->hashObj, field, 1); server.stat_expired_subkeys++; ex->fieldDeleted++; return HSETEX_DELETED; } if (ex->minExpireFields > expireAt) ex->minExpireFields = expireAt; listpackExUpdateExpiry(ex->hashObj, field, fptr, vptr, expireAt); ex->fieldUpdated++; return HSETEX_OK; } /* Returns 1 if expired */ int hashTypeIsExpired(const robj *o, uint64_t expireAt) { if (o->encoding == OBJ_ENCODING_LISTPACK_EX) { if (expireAt == HASH_LP_NO_TTL) return 0; } else if (o->encoding == OBJ_ENCODING_HT) { if (expireAt == EB_EXPIRE_TIME_INVALID) return 0; } else { serverPanic("Unknown encoding: %d", o->encoding); } return (mstime_t) expireAt < commandTimeSnapshot(); } /* Returns listpack pointer of the object. */ unsigned char *hashTypeListpackGetLp(robj *o) { if (o->encoding == OBJ_ENCODING_LISTPACK) return o->ptr; else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) return ((listpackEx*)o->ptr)->lp; serverPanic("Unknown encoding: %d", o->encoding); } /*----------------------------------------------------------------------------- * Hash type API *----------------------------------------------------------------------------*/ /* Check the length of a number of objects to see if we need to convert a * listpack to a real hash. Note that we only check string encoded objects * as their string length can be queried in constant time. */ void hashTypeTryConversion(redisDb *db, robj *o, robj **argv, int start, int end) { int i; size_t sum = 0; if (o->encoding != OBJ_ENCODING_LISTPACK && o->encoding != OBJ_ENCODING_LISTPACK_EX) return; /* We guess that most of the values in the input are unique, so * if there are enough arguments we create a pre-sized hash, which * might over allocate memory if there are duplicates. */ size_t new_fields = (end - start + 1) / 2; if (new_fields > server.hash_max_listpack_entries) { hashTypeConvert(o, OBJ_ENCODING_HT, &db->hexpires); dictExpand(o->ptr, new_fields); return; } for (i = start; i <= end; i++) { if (!sdsEncodedObject(argv[i])) continue; size_t len = sdslen(argv[i]->ptr); if (len > server.hash_max_listpack_value) { hashTypeConvert(o, OBJ_ENCODING_HT, &db->hexpires); return; } sum += len; } if (!lpSafeToAdd(hashTypeListpackGetLp(o), sum)) hashTypeConvert(o, OBJ_ENCODING_HT, &db->hexpires); } /* Get the value from a listpack encoded hash, identified by field. */ GetFieldRes hashTypeGetFromListpack(robj *o, sds field, unsigned char **vstr, unsigned int *vlen, long long *vll, uint64_t *expiredAt) { *expiredAt = EB_EXPIRE_TIME_INVALID; unsigned char *zl, *fptr = NULL, *vptr = NULL; if (o->encoding == OBJ_ENCODING_LISTPACK) { 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); } } } else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) { long long expire; unsigned char *h; listpackEx *lpt = o->ptr; fptr = lpFirst(lpt->lp); if (fptr != NULL) { fptr = lpFind(lpt->lp, fptr, (unsigned char*)field, sdslen(field), 2); if (fptr != NULL) { vptr = lpNext(lpt->lp, fptr); serverAssert(vptr != NULL); h = lpNext(lpt->lp, vptr); serverAssert(h && lpGetIntegerValue(h, &expire)); if (expire != HASH_LP_NO_TTL) *expiredAt = expire; } } } else { serverPanic("Unknown hash encoding: %d", o->encoding); } if (vptr != NULL) { *vstr = lpGetValue(vptr, vlen, vll); return GETF_OK; } return GETF_NOT_FOUND; } /* Get the value from a hash table encoded hash, identified by field. * Returns NULL when the field cannot be found, otherwise the SDS value * is returned. */ GetFieldRes hashTypeGetFromHashTable(robj *o, sds field, sds *value, uint64_t *expiredAt) { dictEntry *de; *expiredAt = EB_EXPIRE_TIME_INVALID; serverAssert(o->encoding == OBJ_ENCODING_HT); de = dictFind(o->ptr, field); if (de == NULL) return GETF_NOT_FOUND; *expiredAt = hfieldGetExpireTime(dictGetKey(de)); *value = (sds) dictGetVal(de); return GETF_OK; } /* Higher level function of hashTypeGet*() that returns the hash value * associated with the specified field. * Arguments: * hfeFlags - Lookup for HFE_LAZY_* flags * * Returned: * GetFieldRes - Result of get operation * vstr, vlen - if string, ref in either *vstr and *vlen if it's * returned in string form, * 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 * always check the function return by checking the return value * for GETF_OK and checking if vll (or vstr) is NULL. * */ GetFieldRes hashTypeGetValue(redisDb *db, robj *o, sds field, unsigned char **vstr, unsigned int *vlen, long long *vll, int hfeFlags) { uint64_t expiredAt; sds key; GetFieldRes res; if (o->encoding == OBJ_ENCODING_LISTPACK || o->encoding == OBJ_ENCODING_LISTPACK_EX) { *vstr = NULL; res = hashTypeGetFromListpack(o, field, vstr, vlen, vll, &expiredAt); if (res == GETF_NOT_FOUND) return GETF_NOT_FOUND; } else if (o->encoding == OBJ_ENCODING_HT) { sds value = NULL; res = hashTypeGetFromHashTable(o, field, &value, &expiredAt); if (res == GETF_NOT_FOUND) return GETF_NOT_FOUND; *vstr = (unsigned char*) value; *vlen = sdslen(value); } else { serverPanic("Unknown hash encoding"); } if (expiredAt >= (uint64_t) commandTimeSnapshot()) return GETF_OK; 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) key = ((listpackEx *) o->ptr)->key; else key = ((dictExpireMetadata *) dictMetadata((dict*)o->ptr))->key; /* delete the field and propagate the deletion */ serverAssert(hashTypeDelete(o, field, 1) == 1); propagateHashFieldDeletion(db, key, field, sdslen(field)); server.stat_expired_subkeys++; /* If the field is the last one in the hash, then the hash will be deleted */ res = GETF_EXPIRED; robj *keyObj = createStringObject(key, sdslen(key)); 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); res = GETF_EXPIRED_HASH; } signalModifiedKey(NULL, db, keyObj); decrRefCount(keyObj); return res; } /* Like hashTypeGetValue() but returns a Redis object, which is useful for * interaction with the hash type outside t_hash.c. * The function returns NULL if the field is not found in the hash. Otherwise * 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 * in the hash, then the hash will as well be deleted. In this case, * isHashDeleted will be set to 1. */ robj *hashTypeGetValueObject(redisDb *db, robj *o, sds field, int hfeFlags, int *isHashDeleted) { unsigned char *vstr; unsigned int vlen; long long vll; if (isHashDeleted) *isHashDeleted = 0; GetFieldRes res = hashTypeGetValue(db,o,field,&vstr,&vlen,&vll, hfeFlags); if (res == GETF_OK) { if (vstr) return createStringObject((char*)vstr,vlen); else return createStringObjectFromLongLong(vll); } if ((res == GETF_EXPIRED_HASH) && (isHashDeleted)) *isHashDeleted = 1; /* GETF_EXPIRED_HASH, GETF_EXPIRED, GETF_NOT_FOUND */ return NULL; } /* Test if the specified field exists in the given hash. If the field is * expired (HFE), then it will be lazy deleted * * hfeFlags - Lookup HFE_LAZY_* flags * 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, * 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 hfeFlags, int *isHashDeleted) { unsigned char *vstr = NULL; unsigned int vlen = UINT_MAX; long long vll = LLONG_MAX; GetFieldRes res = hashTypeGetValue(db, o, field, &vstr, &vlen, &vll, hfeFlags); if (isHashDeleted) *isHashDeleted = (res == GETF_EXPIRED_HASH) ? 1 : 0; return (res == GETF_OK) ? 1 : 0; } /* Add a new field, overwrite the old with the new value if it already exists. * Return 0 on insert and 1 on update. * * By default, the key and value SDS strings are copied if needed, so the * caller retains ownership of the strings passed. However this behavior * can be effected by passing appropriate flags (possibly bitwise OR-ed): * * 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_KEEP_TTL -- keep original TTL if field already exists * * 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 * entry or to free the SDS string before returning to the caller. * * HASH_SET_COPY corresponds to no flags passed, and means the default * semantics of copying the values if needed. * */ #define HASH_SET_TAKE_FIELD (1<<0) #define HASH_SET_TAKE_VALUE (1<<1) #define HASH_SET_KEEP_TTL (1<<2) #define HASH_SET_COPY 0 int hashTypeSet(redisDb *db, robj *o, sds field, sds value, int flags) { int update = 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 (sdslen(field) > server.hash_max_listpack_value || sdslen(value) > server.hash_max_listpack_value) hashTypeConvert(o, OBJ_ENCODING_HT, &db->hexpires); } if (o->encoding == OBJ_ENCODING_LISTPACK) { unsigned char *zl, *fptr, *vptr; 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); /* Replace value */ zl = lpReplace(zl, &vptr, (unsigned char*)value, sdslen(value)); update = 1; } } 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); de = dictAddRaw(ht, newField, &existing); dictUseStoredKeyApi(ht, 0); /* If field already exists, then update "field". "Value" will be set afterward */ if (de == NULL) { 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; } if (flags & HASH_SET_TAKE_VALUE) { dictSetVal(ht, de, value); flags &= ~HASH_SET_TAKE_VALUE; } else { dictSetVal(ht, de, sdsdup(value)); } } else { serverPanic("Unknown hash encoding"); } /* 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; } SetExRes hashTypeSetExpiryHT(HashTypeSetEx *exInfo, sds field, uint64_t expireAt) { dict *ht = exInfo->hashObj->ptr; dictEntry *existingEntry = NULL; /* New field with expiration metadata */ hfield hfNew = hfieldNew(field, sdslen(field), 1 /*withExpireMeta*/); if ((existingEntry = dictFind(ht, field)) == NULL) { hfieldFree(hfNew); return HSETEX_NO_FIELD; } hfield hfOld = dictGetKey(existingEntry); /* If field doesn't have expiry metadata attached */ if (!hfieldIsExpireAttached(hfOld)) { /* For fields without expiry, LT condition is considered valid */ if (exInfo->expireSetCond & (HFE_XX | HFE_GT)) { hfieldFree(hfNew); return HSETEX_NO_CONDITION_MET; } /* Delete old field. Below goanna dictSetKey(..,hfNew) */ hfieldFree(hfOld); } else { /* field has ExpireMeta struct attached */ /* No need for hfNew (Just modify expire-time of existing field) */ hfieldFree(hfNew); uint64_t prevExpire = hfieldGetExpireTime(hfOld); /* If field has valid expiration time, then check GT|LT|NX */ 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; /* remove old expiry time from hash's private ebuckets */ dictExpireMetadata *dm = (dictExpireMetadata *) dictMetadata(ht); ebRemove(&dm->hfe, &hashFieldExpireBucketsType, hfOld); /* Track of minimum expiration time (only later update global HFE DS) */ if (exInfo->minExpireFields > prevExpire) exInfo->minExpireFields = prevExpire; } else { /* field has invalid expiry. No need to ebRemove() */ /* Check XX|LT|GT */ if (exInfo->expireSetCond & (HFE_XX | HFE_GT)) return HSETEX_NO_CONDITION_MET; } /* Reuse hfOld as hfNew and rewrite its expiry with ebAdd() */ hfNew = hfOld; } 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))) { /* replicas should not initiate deletion of fields */ propagateHashFieldDeletion(exInfo->db, exInfo->key->ptr, field, sdslen(field)); hashTypeDelete(exInfo->hashObj, field, 1); server.stat_expired_subkeys++; exInfo->fieldDeleted++; return HSETEX_DELETED; } if (exInfo->minExpireFields > expireAt) exInfo->minExpireFields = expireAt; dictExpireMetadata *dm = (dictExpireMetadata *) dictMetadata(ht); ebAdd(&dm->hfe, &hashFieldExpireBucketsType, hfNew, expireAt); exInfo->fieldUpdated++; return HSETEX_OK; } /* * Set field expiration * * Take care to call first hashTypeSetExInit() and then call this function. * Finally, call hashTypeSetExDone() to notify and update global HFE DS. */ SetExRes hashTypeSetEx(robj *o, sds field, uint64_t expireAt, HashTypeSetEx *exInfo) { 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; if ((fptr = lpFirst(lpt->lp)) == NULL) return HSETEX_NO_FIELD; fptr = lpFind(lpt->lp, fptr, (unsigned char*)field, sdslen(field), 2); if (!fptr) return HSETEX_NO_FIELD; /* Grab pointer to the value (fptr points to the field) */ vptr = lpNext(lpt->lp, fptr); serverAssert(vptr != NULL); tptr = lpNext(lpt->lp, vptr); serverAssert(tptr && lpGetIntegerValue(tptr, &expireTime)); /* update TTL */ return hashTypeSetExpiryListpack(exInfo, field, fptr, vptr, tptr, expireAt); } else if (o->encoding == OBJ_ENCODING_HT) { /* If needed to set the field along with expiry */ return hashTypeSetExpiryHT(exInfo, field, expireAt); } else { serverPanic("Unknown hash encoding"); } return HSETEX_OK; /* never reach here */ } void initDictExpireMetadata(sds key, robj *o) { dict *ht = o->ptr; dictExpireMetadata *m = (dictExpireMetadata *) dictMetadata(ht); m->key = key; m->hfe = ebCreate(); /* Allocate HFE DS */ m->expireMeta.trash = 1; /* mark as trash (as long it wasn't ebAdd()) */ } /* Init HashTypeSetEx struct before calling hashTypeSetEx() */ int hashTypeSetExInit(robj *key, robj *o, client *c, redisDb *db, const char *cmd, ExpireSetCond expireSetCond, HashTypeSetEx *ex) { dict *ht = o->ptr; ex->expireSetCond = expireSetCond; ex->minExpire = EB_EXPIRE_TIME_INVALID; ex->c = c; ex->cmd = cmd; ex->db = db; ex->key = key; ex->hashObj = o; ex->fieldDeleted = 0; ex->fieldUpdated = 0; ex->minExpireFields = EB_EXPIRE_TIME_INVALID; /* Take care that HASH support expiration */ if (o->encoding == OBJ_ENCODING_LISTPACK) { hashTypeConvert(o, OBJ_ENCODING_LISTPACK_EX, &c->db->hexpires); listpackEx *lpt = o->ptr; dictEntry *de = dbFind(c->db, key->ptr); serverAssert(de != NULL); lpt->key = dictGetKey(de); } else if (o->encoding == OBJ_ENCODING_HT) { /* Take care dict has HFE metadata */ if (!isDictWithMetaHFE(ht)) { /* Realloc (only header of dict) with metadata for hash-field expiration */ dictTypeAddMeta(&ht, &mstrHashDictTypeWithHFE); dictExpireMetadata *m = (dictExpireMetadata *) dictMetadata(ht); o->ptr = ht; /* Find the key in the keyspace. Need to keep reference to the key for * notifications or even removal of the hash */ dictEntry *de = dbFind(db, key->ptr); serverAssert(de != NULL); /* Fillup dict HFE metadata */ m->key = dictGetKey(de); /* reference key in keyspace */ m->hfe = ebCreate(); /* Allocate HFE DS */ m->expireMeta.trash = 1; /* mark as trash (as long it wasn't ebAdd()) */ } } /* Read minExpire from attached ExpireMeta to the hash */ ex->minExpire = hashTypeGetMinExpire(o, 0); return C_OK; } /* * After calling hashTypeSetEx() for setting fields or their expiry, call this * function to notify and update global HFE DS. */ void hashTypeSetExDone(HashTypeSetEx *ex) { /* Notify keyspace event, update dirty count and update global HFE DS */ if (ex->fieldDeleted + ex->fieldUpdated > 0) { server.dirty += ex->fieldDeleted + ex->fieldUpdated; if (ex->fieldDeleted && hashTypeLength(ex->hashObj, 0) == 0) { dbDelete(ex->db,ex->key); signalModifiedKey(ex->c, ex->db, ex->key); notifyKeyspaceEvent(NOTIFY_HASH, "hdel", ex->key, ex->db->id); notifyKeyspaceEvent(NOTIFY_GENERIC,"del",ex->key, ex->db->id); } else { signalModifiedKey(ex->c, ex->db, ex->key); notifyKeyspaceEvent(NOTIFY_HASH, ex->fieldDeleted ? "hdel" : "hexpire", ex->key, ex->db->id); /* 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 * than expiration time provided in the command, then the minimum * HFE of the hash won't change following this command. */ if ((ex->minExpire < ex->minExpireFields)) return; /* Retrieve new expired time. It might have changed. */ uint64_t newMinExpire = hashTypeGetMinExpire(ex->hashObj, 1 /*accurate*/); /* 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 * case fields of hash will be active-expired up to few seconds later. * * In any case, active-expire operation will know to update global * HFE DS more efficiently than here for a single item. */ uint64_t diff = (ex->minExpire > newMinExpire) ? (ex->minExpire - newMinExpire) : (newMinExpire - ex->minExpire); if (diff < HASH_NEW_EXPIRE_DIFF_THRESHOLD) return; if (ex->minExpire != EB_EXPIRE_TIME_INVALID) ebRemove(&ex->db->hexpires, &hashExpireBucketsType, ex->hashObj); if (newMinExpire != EB_EXPIRE_TIME_INVALID) ebAdd(&ex->db->hexpires, &hashExpireBucketsType, ex->hashObj, newMinExpire); } } } /* Delete an element from a hash. * * Return 1 on deleted and 0 on not found. * isSdsField - 1 if the field is sds, 0 if it is hfield */ int hashTypeDelete(robj *o, void *field, int isSdsField) { int deleted = 0; int fieldLen = (isSdsField) ? sdslen((sds)field) : hfieldlen((hfield)field); if (o->encoding == OBJ_ENCODING_LISTPACK) { unsigned char *zl, *fptr; zl = o->ptr; fptr = lpFirst(zl); if (fptr != NULL) { fptr = lpFind(zl, fptr, (unsigned char*)field, fieldLen, 1); if (fptr != NULL) { /* Delete both of the key and the value. */ zl = lpDeleteRangeWithEntry(zl,&fptr,2); o->ptr = zl; deleted = 1; } } } else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) { unsigned char *fptr; listpackEx *lpt = o->ptr; fptr = lpFirst(lpt->lp); if (fptr != NULL) { fptr = lpFind(lpt->lp, fptr, (unsigned char*)field, fieldLen, 2); if (fptr != NULL) { /* Delete field, value and ttl */ lpt->lp = lpDeleteRangeWithEntry(lpt->lp, &fptr, 3); deleted = 1; } } } else if (o->encoding == OBJ_ENCODING_HT) { /* dictDelete() will call dictHfieldDestructor() */ dictUseStoredKeyApi((dict*)o->ptr, isSdsField ? 0 : 1); if (dictDelete((dict*)o->ptr, field) == C_OK) { deleted = 1; } dictUseStoredKeyApi((dict*)o->ptr, 0); } else { serverPanic("Unknown hash encoding"); } return deleted; } /* Return the number of elements in a hash. * * Note, subtractExpiredFields=1 might be pricy in case there are many HFEs */ unsigned long hashTypeLength(const robj *o, int subtractExpiredFields) { unsigned long length = ULONG_MAX; if (o->encoding == OBJ_ENCODING_LISTPACK) { length = lpLength(o->ptr) / 2; } else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) { listpackEx *lpt = o->ptr; length = lpLength(lpt->lp) / 3; if (subtractExpiredFields && lpt->meta.trash == 0) length -= listpackExExpireDryRun(o); } else if (o->encoding == OBJ_ENCODING_HT) { uint64_t expiredItems = 0; dict *d = (dict*)o->ptr; if (subtractExpiredFields && isDictWithMetaHFE(d)) { dictExpireMetadata *meta = (dictExpireMetadata *) dictMetadata(d); /* If dict registered in global HFE DS */ if (meta->expireMeta.trash == 0) expiredItems = ebExpireDryRun(meta->hfe, &hashFieldExpireBucketsType, commandTimeSnapshot()); } length = dictSize(d) - expiredItems; } else { serverPanic("Unknown hash encoding"); } return length; } hashTypeIterator *hashTypeInitIterator(robj *subject) { hashTypeIterator *hi = zmalloc(sizeof(hashTypeIterator)); hi->subject = subject; hi->encoding = subject->encoding; if (hi->encoding == OBJ_ENCODING_LISTPACK || hi->encoding == OBJ_ENCODING_LISTPACK_EX) { hi->fptr = NULL; hi->vptr = NULL; hi->tptr = NULL; hi->expire_time = EB_EXPIRE_TIME_INVALID; } else if (hi->encoding == OBJ_ENCODING_HT) { hi->di = dictGetIterator(subject->ptr); } else { serverPanic("Unknown hash encoding"); } return hi; } void hashTypeReleaseIterator(hashTypeIterator *hi) { if (hi->encoding == OBJ_ENCODING_HT) dictReleaseIterator(hi->di); zfree(hi); } /* Move to the next entry in the hash. Return C_OK when the next entry * could be found and C_ERR when the iterator reaches the end. */ int hashTypeNext(hashTypeIterator *hi, int skipExpiredFields) { hi->expire_time = EB_EXPIRE_TIME_INVALID; if (hi->encoding == OBJ_ENCODING_LISTPACK) { unsigned char *zl; unsigned char *fptr, *vptr; zl = hi->subject->ptr; fptr = hi->fptr; vptr = hi->vptr; if (fptr == NULL) { /* Initialize cursor */ serverAssert(vptr == NULL); fptr = lpFirst(zl); } else { /* Advance cursor */ serverAssert(vptr != NULL); fptr = lpNext(zl, vptr); } if (fptr == NULL) return C_ERR; /* Grab pointer to the value (fptr points to the field) */ vptr = lpNext(zl, fptr); serverAssert(vptr != NULL); /* fptr, vptr now point to the first or next pair */ hi->fptr = fptr; hi->vptr = vptr; } else if (hi->encoding == OBJ_ENCODING_LISTPACK_EX) { long long expire_time; unsigned char *zl = hashTypeListpackGetLp(hi->subject); unsigned char *fptr, *vptr, *tptr; fptr = hi->fptr; vptr = hi->vptr; tptr = hi->tptr; if (fptr == NULL) { /* Initialize cursor */ serverAssert(vptr == NULL); fptr = lpFirst(zl); } else { /* Advance cursor */ serverAssert(tptr != NULL); fptr = lpNext(zl, tptr); } if (fptr == NULL) return C_ERR; while (fptr != NULL) { /* Grab pointer to the value (fptr points to the field) */ vptr = lpNext(zl, fptr); serverAssert(vptr != NULL); tptr = lpNext(zl, vptr); serverAssert(tptr && lpGetIntegerValue(tptr, &expire_time)); if (!skipExpiredFields || !hashTypeIsExpired(hi->subject, expire_time)) break; fptr = lpNext(zl, tptr); } if (fptr == NULL) return C_ERR; /* fptr, vptr now point to the first or next pair */ hi->fptr = fptr; hi->vptr = vptr; hi->tptr = tptr; hi->expire_time = (expire_time != HASH_LP_NO_TTL) ? (uint64_t) expire_time : EB_EXPIRE_TIME_INVALID; } else if (hi->encoding == OBJ_ENCODING_HT) { while ((hi->de = dictNext(hi->di)) != NULL) { hi->expire_time = hfieldGetExpireTime(dictGetKey(hi->de)); /* this condition still valid if expire_time equals EB_EXPIRE_TIME_INVALID */ if (skipExpiredFields && ((mstime_t)hi->expire_time < commandTimeSnapshot())) continue; return C_OK; } return C_ERR; } else { serverPanic("Unknown hash encoding"); } return C_OK; } /* Get the field or value at iterator cursor, for an iterator on a hash value * encoded as a listpack. Prototype is similar to `hashTypeGetFromListpack`. */ void hashTypeCurrentFromListpack(hashTypeIterator *hi, int what, unsigned char **vstr, unsigned int *vlen, long long *vll, uint64_t *expireTime) { serverAssert(hi->encoding == OBJ_ENCODING_LISTPACK || hi->encoding == OBJ_ENCODING_LISTPACK_EX); if (what & OBJ_HASH_KEY) { *vstr = lpGetValue(hi->fptr, vlen, vll); } else { *vstr = lpGetValue(hi->vptr, vlen, vll); } if (expireTime) *expireTime = hi->expire_time; } /* Get the field or value at iterator cursor, for an iterator on a hash value * encoded as a hash table. Prototype is similar to * `hashTypeGetFromHashTable`. * * expireTime - If parameter is not null, then the function will return the expire * time of the field. If expiry not set, return EB_EXPIRE_TIME_INVALID */ void hashTypeCurrentFromHashTable(hashTypeIterator *hi, int what, char **str, size_t *len, uint64_t *expireTime) { serverAssert(hi->encoding == OBJ_ENCODING_HT); hfield key = NULL; if (what & OBJ_HASH_KEY) { key = dictGetKey(hi->de); *str = key; *len = hfieldlen(key); } else { sds val = dictGetVal(hi->de); *str = val; *len = sdslen(val); } if (expireTime) *expireTime = hi->expire_time; } /* Higher level function of hashTypeCurrent*() that returns the hash value * at current iterator position. * * The returned element is returned by reference in either *vstr and *vlen if * it's returned in string form, or stored in *vll if it's returned as * a number. * * If *vll is populated *vstr is set to NULL, so the caller * can always check the function return by checking the return value * type checking if vstr == NULL. */ void hashTypeCurrentObject(hashTypeIterator *hi, int what, unsigned char **vstr, unsigned int *vlen, long long *vll, uint64_t *expireTime) { if (hi->encoding == OBJ_ENCODING_LISTPACK || hi->encoding == OBJ_ENCODING_LISTPACK_EX) { *vstr = NULL; hashTypeCurrentFromListpack(hi, what, vstr, vlen, vll, expireTime); } else if (hi->encoding == OBJ_ENCODING_HT) { char *ele; size_t eleLen; hashTypeCurrentFromHashTable(hi, what, &ele, &eleLen, expireTime); *vstr = (unsigned char*) ele; *vlen = eleLen; } else { serverPanic("Unknown hash encoding"); } } /* Return the key or value at the current iterator position as a new * SDS string. */ sds hashTypeCurrentObjectNewSds(hashTypeIterator *hi, int what) { unsigned char *vstr; unsigned int vlen; long long vll; hashTypeCurrentObject(hi,what,&vstr,&vlen,&vll, NULL); if (vstr) return sdsnewlen(vstr,vlen); return sdsfromlonglong(vll); } /* Return the key at the current iterator position as a new hfield string. */ hfield hashTypeCurrentObjectNewHfield(hashTypeIterator *hi) { char buf[LONG_STR_SIZE]; unsigned char *vstr; unsigned int vlen; long long vll; uint64_t expireTime; hfield hf; hashTypeCurrentObject(hi,OBJ_HASH_KEY,&vstr,&vlen,&vll, &expireTime); if (!vstr) { vlen = ll2string(buf, sizeof(buf), vll); vstr = (unsigned char *) buf; } hf = hfieldNew(vstr,vlen, expireTime != EB_EXPIRE_TIME_INVALID); return hf; } static robj *hashTypeLookupWriteOrCreate(client *c, robj *key) { robj *o = lookupKeyWrite(c->db,key); if (checkType(c,o,OBJ_HASH)) return NULL; if (o == NULL) { o = createHashObject(); dbAdd(c->db,key,o); } return o; } void hashTypeConvertListpack(robj *o, int enc) { serverAssert(o->encoding == OBJ_ENCODING_LISTPACK); if (enc == OBJ_ENCODING_LISTPACK) { /* Nothing to do... */ } else if (enc == OBJ_ENCODING_LISTPACK_EX) { unsigned char *p; /* Append HASH_LP_NO_TTL to each field name - value pair. */ p = lpFirst(o->ptr); while (p != NULL) { p = lpNext(o->ptr, p); serverAssert(p); o->ptr = lpInsertInteger(o->ptr, HASH_LP_NO_TTL, p, LP_AFTER, &p); p = lpNext(o->ptr, p); } listpackEx *lpt = listpackExCreate(); lpt->lp = o->ptr; o->encoding = OBJ_ENCODING_LISTPACK_EX; o->ptr = lpt; } else if (enc == OBJ_ENCODING_HT) { hashTypeIterator *hi; dict *dict; int ret; hi = hashTypeInitIterator(o); dict = dictCreate(&mstrHashDictType); /* Presize the dict to avoid rehashing */ dictExpand(dict,hashTypeLength(o, 0)); while (hashTypeNext(hi, 0) != C_ERR) { hfield key = hashTypeCurrentObjectNewHfield(hi); sds value = hashTypeCurrentObjectNewSds(hi,OBJ_HASH_VALUE); dictUseStoredKeyApi(dict, 1); ret = dictAdd(dict, key, value); dictUseStoredKeyApi(dict, 0); if (ret != DICT_OK) { hfieldFree(key); sdsfree(value); /* Needed for gcc ASAN */ hashTypeReleaseIterator(hi); /* Needed for gcc ASAN */ serverLogHexDump(LL_WARNING,"listpack with dup elements dump", o->ptr,lpBytes(o->ptr)); serverPanic("Listpack corruption detected"); } } hashTypeReleaseIterator(hi); zfree(o->ptr); o->encoding = OBJ_ENCODING_HT; o->ptr = dict; } else { serverPanic("Unknown hash encoding"); } } void hashTypeConvertListpackEx(robj *o, int enc, ebuckets *hexpires) { serverAssert(o->encoding == OBJ_ENCODING_LISTPACK_EX); if (enc == OBJ_ENCODING_LISTPACK_EX) { return; } else if (enc == OBJ_ENCODING_HT) { int ret; hashTypeIterator *hi; dict *dict; dictExpireMetadata *dictExpireMeta; listpackEx *lpt = o->ptr; uint64_t minExpire = hashTypeGetMinExpire(o, 0); if (hexpires && lpt->meta.trash != 1) ebRemove(hexpires, &hashExpireBucketsType, o); dict = dictCreate(&mstrHashDictTypeWithHFE); dictExpand(dict,hashTypeLength(o, 0)); dictExpireMeta = (dictExpireMetadata *) dictMetadata(dict); /* Fillup dict HFE metadata */ dictExpireMeta->key = lpt->key; /* reference key in keyspace */ dictExpireMeta->hfe = ebCreate(); /* Allocate HFE DS */ dictExpireMeta->expireMeta.trash = 1; /* mark as trash (as long it wasn't ebAdd()) */ hi = hashTypeInitIterator(o); while (hashTypeNext(hi, 0) != C_ERR) { hfield key = hashTypeCurrentObjectNewHfield(hi); sds value = hashTypeCurrentObjectNewSds(hi,OBJ_HASH_VALUE); dictUseStoredKeyApi(dict, 1); ret = dictAdd(dict, key, value); dictUseStoredKeyApi(dict, 0); if (ret != DICT_OK) { hfieldFree(key); sdsfree(value); /* Needed for gcc ASAN */ hashTypeReleaseIterator(hi); /* Needed for gcc ASAN */ serverLogHexDump(LL_WARNING,"listpack with dup elements dump", lpt->lp,lpBytes(lpt->lp)); serverPanic("Listpack corruption detected"); } if (hi->expire_time != EB_EXPIRE_TIME_INVALID) ebAdd(&dictExpireMeta->hfe, &hashFieldExpireBucketsType, key, hi->expire_time); } hashTypeReleaseIterator(hi); listpackExFree(lpt); o->encoding = OBJ_ENCODING_HT; o->ptr = dict; if (hexpires && minExpire != EB_EXPIRE_TIME_INVALID) ebAdd(hexpires, &hashExpireBucketsType, o, minExpire); } else { serverPanic("Unknown hash encoding: %d", enc); } } /* NOTE: hexpires can be NULL (Won't register in global HFE DS) */ void hashTypeConvert(robj *o, int enc, ebuckets *hexpires) { if (o->encoding == OBJ_ENCODING_LISTPACK) { hashTypeConvertListpack(o, enc); } else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) { hashTypeConvertListpackEx(o, enc, hexpires); } else if (o->encoding == OBJ_ENCODING_HT) { serverPanic("Not implemented"); } else { serverPanic("Unknown hash encoding"); } } /* This is a helper function for the COPY command. * Duplicate a hash object, with the guarantee that the returned object * has the same encoding as the original one. * * The resulting object always has refcount set to 1 */ robj *hashTypeDup(robj *o, sds newkey, uint64_t *minHashExpire) { robj *hobj; hashTypeIterator *hi; serverAssert(o->type == OBJ_HASH); if(o->encoding == OBJ_ENCODING_LISTPACK) { unsigned char *zl = o->ptr; size_t sz = lpBytes(zl); unsigned char *new_zl = zmalloc(sz); memcpy(new_zl, zl, sz); hobj = createObject(OBJ_HASH, new_zl); hobj->encoding = OBJ_ENCODING_LISTPACK; } else if(o->encoding == OBJ_ENCODING_LISTPACK_EX) { listpackEx *lpt = o->ptr; if (lpt->meta.trash == 0) *minHashExpire = ebGetMetaExpTime(&lpt->meta); listpackEx *dup = listpackExCreate(); dup->key = newkey; size_t sz = lpBytes(lpt->lp); dup->lp = lpNew(sz); memcpy(dup->lp, lpt->lp, sz); hobj = createObject(OBJ_HASH, dup); hobj->encoding = OBJ_ENCODING_LISTPACK_EX; } else if(o->encoding == OBJ_ENCODING_HT) { dictExpireMetadata *dictExpireMetaSrc, *dictExpireMetaDst = NULL; dict *d; /* If dict doesn't have HFE metadata, then create a new dict without it */ if (!isDictWithMetaHFE(o->ptr)) { d = dictCreate(&mstrHashDictType); } else { /* Create a new dict with HFE metadata */ d = dictCreate(&mstrHashDictTypeWithHFE); dictExpireMetaSrc = (dictExpireMetadata *) dictMetadata((dict *) o->ptr); dictExpireMetaDst = (dictExpireMetadata *) dictMetadata(d); dictExpireMetaDst->key = newkey; /* reference key in keyspace */ dictExpireMetaDst->hfe = ebCreate(); /* Allocate HFE DS */ dictExpireMetaDst->expireMeta.trash = 1; /* mark as trash (as long it wasn't ebAdd()) */ /* Extract the minimum expire time of the source hash (Will be used by caller * to register the new hash in the global ebuckets, i.e db->hexpires) */ if (dictExpireMetaSrc->expireMeta.trash == 0) *minHashExpire = ebGetMetaExpTime(&dictExpireMetaSrc->expireMeta); } dictExpand(d, dictSize((const dict*)o->ptr)); hi = hashTypeInitIterator(o); while (hashTypeNext(hi, 0) != C_ERR) { uint64_t expireTime; sds newfield, newvalue; /* Extract a field-value pair from an original hash object.*/ char *field, *value; size_t fieldLen, valueLen; hashTypeCurrentFromHashTable(hi, OBJ_HASH_KEY, &field, &fieldLen, &expireTime); if (expireTime == EB_EXPIRE_TIME_INVALID) { newfield = hfieldNew(field, fieldLen, 0); } else { newfield = hfieldNew(field, fieldLen, 1); ebAdd(&dictExpireMetaDst->hfe, &hashFieldExpireBucketsType, newfield, expireTime); } hashTypeCurrentFromHashTable(hi, OBJ_HASH_VALUE, &value, &valueLen, NULL); newvalue = sdsnewlen(value, valueLen); /* Add a field-value pair to a new hash object. */ dictUseStoredKeyApi(d, 1); dictAdd(d,newfield,newvalue); dictUseStoredKeyApi(d, 0); } hashTypeReleaseIterator(hi); hobj = createObject(OBJ_HASH, d); hobj->encoding = OBJ_ENCODING_HT; } else { serverPanic("Unknown hash encoding"); } return hobj; } /* Create a new sds string from the listpack entry. */ sds hashSdsFromListpackEntry(listpackEntry *e) { return e->sval ? sdsnewlen(e->sval, e->slen) : sdsfromlonglong(e->lval); } /* Reply with bulk string from the listpack entry. */ void hashReplyFromListpackEntry(client *c, listpackEntry *e) { if (e->sval) addReplyBulkCBuffer(c, e->sval, e->slen); else addReplyBulkLongLong(c, e->lval); } /* Return random element from a non empty hash. * 'key' and 'val' will be set to hold the element. * 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. */ void hashTypeRandomElement(robj *hashobj, unsigned long hashsize, CommonEntry *key, CommonEntry *val) { if (hashobj->encoding == OBJ_ENCODING_HT) { dictEntry *de = dictGetFairRandomKey(hashobj->ptr); hfield field = dictGetKey(de); key->sval = (unsigned char*)field; key->slen = hfieldlen(field); if (val) { sds s = dictGetVal(de); val->sval = (unsigned char*)s; val->slen = sdslen(s); } } else if (hashobj->encoding == OBJ_ENCODING_LISTPACK) { lpRandomPair(hashobj->ptr, hashsize, (listpackEntry *) key, (listpackEntry *) val, 2); } else if (hashobj->encoding == OBJ_ENCODING_LISTPACK_EX) { lpRandomPair(hashTypeListpackGetLp(hashobj), hashsize, (listpackEntry *) key, (listpackEntry *) val, 3); } else { serverPanic("Unknown hash encoding"); } } /* * Active expiration of fields in hash * * Called by hashTypeDbActiveExpire() for each hash registered in the HFE DB * (db->hexpires) with an expiration-time less than or equal current time. * * This callback performs the following actions for each hash: * - Delete expired fields as by calling ebExpire(hash) * - If afterward there are future fields to expire, it will update the hash in * HFE DB with the next hash-field minimum expiration time by returning * ACT_UPDATE_EXP_ITEM. * - If the hash has no more fields to expire, it is removed from the HFE DB * by returning ACT_REMOVE_EXP_ITEM. * - If hash has no more fields afterward, it will remove the hash from keyspace. */ static ExpireAction hashTypeActiveExpire(eItem item, void *ctx) { ExpireCtx *expireCtx = ctx; /* If no more quota left for this callback, stop */ if (expireCtx->fieldsToExpireQuota == 0) return ACT_STOP_ACTIVE_EXP; uint64_t nextExpTime = hashTypeExpire((robj *) item, expireCtx, 0); /* 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(), .itemsExpired = 0}; listpackExExpire(db, o, &info); keystr = ((listpackEx*)o->ptr)->key; } else { serverAssert(o->encoding == OBJ_ENCODING_HT); dict *d = o->ptr; dictExpireMetadata *dictExpireMeta = (dictExpireMetadata *) dictMetadata(d); OnFieldExpireCtx onFieldExpireCtx = { .hashObj = o, .db = db }; info = (ExpireInfo){ .maxToExpire = expireCtx->fieldsToExpireQuota, .onExpireItem = onFieldExpire, .ctx = &onFieldExpireCtx, .now = commandTimeSnapshot() }; ebExpire(&dictExpireMeta->hfe, &hashFieldExpireBucketsType, &info); keystr = dictExpireMeta->key; } /* Update quota left */ expireCtx->fieldsToExpireQuota -= info.itemsExpired; /* In some cases, a field might have been deleted without updating the global DS. * As a result, active-expire might not expire any fields, in such cases, * we don't need to send notifications or perform other operations for this key. */ if (info.itemsExpired) { robj *key = createStringObject(keystr, sdslen(keystr)); notifyKeyspaceEvent(NOTIFY_HASH, "hexpired", key, db->id); if (updateGlobalHFE) 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); } signalModifiedKey(NULL, db, key); decrRefCount(key); } /* 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; } /* Delete all expired fields in hash if needed (Currently used only by HRANDFIELD) * * Return 1 if the entire hash was deleted, 0 otherwise. * This function might be pricy in case there are many expired fields. */ static int hashTypeExpireIfNeeded(redisDb *db, robj *o) { uint64_t nextExpireTime; uint64_t minExpire = hashTypeGetMinExpire(o, 1 /*accurate*/); /* Nothing to expire */ if ((mstime_t) minExpire >= commandTimeSnapshot()) return 0; /* Follow expireIfNeeded() conditions of when not lazy-expire */ 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. * accurate=1 - Return the exact time by looking into the object DS. * accurate=0 - Return the minimum expiration time maintained in expireMeta * (Verify it is not trash before using it) 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; 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) { return EB_EXPIRE_TIME_INVALID; } else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) { return listpackExGetMinExpire(o); } else { serverAssert(o->encoding == OBJ_ENCODING_HT); dict *d = o->ptr; if (!isDictWithMetaHFE(d)) return EB_EXPIRE_TIME_INVALID; dictExpireMetadata *expireMeta = (dictExpireMetadata *) dictMetadata(d); return ebGetNextTimeToExpire(expireMeta->hfe, &hashFieldExpireBucketsType); } } uint64_t hashTypeRemoveFromExpires(ebuckets *hexpires, robj *o) { if (o->encoding == OBJ_ENCODING_LISTPACK) { return EB_EXPIRE_TIME_INVALID; } else if (o->encoding == OBJ_ENCODING_HT) { /* If dict doesn't holds HFE metadata */ if (!isDictWithMetaHFE(o->ptr)) return EB_EXPIRE_TIME_INVALID; } uint64_t expireTime = ebGetExpireTime(&hashExpireBucketsType, o); /* If registered in global HFE DS then remove it (not trash) */ if (expireTime != EB_EXPIRE_TIME_INVALID) ebRemove(hexpires, &hashExpireBucketsType, o); return expireTime; } int hashTypeIsFieldsWithExpire(robj *o) { if (o->encoding == OBJ_ENCODING_LISTPACK) { return 0; } else if (o->encoding == OBJ_ENCODING_LISTPACK_EX) { return EB_EXPIRE_TIME_INVALID != listpackExGetMinExpire(o); } else { /* o->encoding == OBJ_ENCODING_HT */ dict *d = o->ptr; /* If dict doesn't holds HFE metadata */ if (!isDictWithMetaHFE(d)) return 0; dictExpireMetadata *meta = (dictExpireMetadata *) dictMetadata(d); return ebGetTotalItems(meta->hfe, &hashFieldExpireBucketsType) != 0; } } /* Add hash to global HFE DS and update key for notifications. * * key - must be the same key instance that is persisted in db->dict * expireTime - expiration in msec. * If eq. 0 then the hash will be added to the global HFE DS with * the minimum expiration time that is already written in advance * to attached metadata (which considered as trash as long as it is * not attached to global HFE DS). * * Precondition: It is a hash of type listpackex or HT with HFE metadata. */ void hashTypeAddToExpires(redisDb *db, sds key, robj *hashObj, uint64_t expireTime) { if (expireTime > EB_EXPIRE_TIME_MAX) return; if (hashObj->encoding == OBJ_ENCODING_LISTPACK_EX) { listpackEx *lpt = hashObj->ptr; lpt->key = key; expireTime = (expireTime) ? expireTime : ebGetMetaExpTime(&lpt->meta); ebAdd(&db->hexpires, &hashExpireBucketsType, hashObj, expireTime); } else if (hashObj->encoding == OBJ_ENCODING_HT) { dict *d = hashObj->ptr; if (isDictWithMetaHFE(d)) { dictExpireMetadata *meta = (dictExpireMetadata *) dictMetadata(d); expireTime = (expireTime) ? expireTime : ebGetMetaExpTime(&meta->expireMeta); meta->key = key; ebAdd(&db->hexpires, &hashExpireBucketsType, hashObj, expireTime); } } } /* DB active expire and update hashes with time-expiration on fields. * * The callback function hashTypeActiveExpire() is invoked for each hash registered * in the HFE DB (db->expires) with an expiration-time less than or equal to the * current time. This callback performs the following actions for each hash: * - If the hash has one or more fields to expire, it will delete those fields. * - If there are more fields to expire, it will update the hash with the next * expiration time in HFE DB. * - If the hash has no more fields to expire, it is removed from the HFE DB. * - If the hash has no more fields, it is removed from the main DB. * * Returns number of fields active-expired. */ uint64_t hashTypeDbActiveExpire(redisDb *db, uint32_t maxFieldsToExpire) { ExpireCtx ctx = { .db = db, .fieldsToExpireQuota = maxFieldsToExpire }; ExpireInfo info = { .maxToExpire = UINT64_MAX, /* Only maxFieldsToExpire play a role */ .onExpireItem = hashTypeActiveExpire, .ctx = &ctx, .now = commandTimeSnapshot(), .itemsExpired = 0}; ebExpire(&db->hexpires, &hashExpireBucketsType, &info); /* Return number of fields active-expired */ return maxFieldsToExpire - ctx.fieldsToExpireQuota; } void hashTypeFree(robj *o) { switch (o->encoding) { case OBJ_ENCODING_HT: /* Verify hash is not registered in global HFE ds */ if (isDictWithMetaHFE((dict*)o->ptr)) { dictExpireMetadata *m = (dictExpireMetadata *)dictMetadata((dict*)o->ptr); serverAssert(m->expireMeta.trash == 1); } dictRelease((dict*) o->ptr); break; case OBJ_ENCODING_LISTPACK: lpFree(o->ptr); break; case OBJ_ENCODING_LISTPACK_EX: /* Verify hash is not registered in global HFE ds */ serverAssert(((listpackEx *) o->ptr)->meta.trash == 1); listpackExFree(o->ptr); break; default: serverPanic("Unknown hash encoding type"); break; } } /* Attempts to update the reference to the new key. Now it's only used in defrag. */ void hashTypeUpdateKeyRef(robj *o, sds newkey) { if (o->encoding == OBJ_ENCODING_LISTPACK_EX) { listpackEx *lpt = o->ptr; lpt->key = newkey; } else if (o->encoding == OBJ_ENCODING_HT && isDictWithMetaHFE(o->ptr)) { dictExpireMetadata *dictExpireMeta = (dictExpireMetadata *)dictMetadata((dict*)o->ptr); dictExpireMeta->key = newkey; } else { /* Nothing to do. */ } } ebuckets *hashTypeGetDictMetaHFE(dict *d) { dictExpireMetadata *dictExpireMeta = (dictExpireMetadata *) dictMetadata(d); return &dictExpireMeta->hfe; } /*----------------------------------------------------------------------------- * Hash type commands *----------------------------------------------------------------------------*/ void hsetnxCommand(client *c) { int isHashDeleted; robj *o; if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return; if (hashTypeExists(c->db, o, c->argv[2]->ptr, HFE_LAZY_EXPIRE, &isHashDeleted)) { addReply(c, shared.czero); return; } /* Field expired and in turn hash deleted. Create new one! */ if (isHashDeleted) { o = createHashObject(); dbAdd(c->db,c->argv[1],o); } hashTypeTryConversion(c->db, o,c->argv,2,3); hashTypeSet(c->db, o,c->argv[2]->ptr,c->argv[3]->ptr,HASH_SET_COPY); addReply(c, shared.cone); signalModifiedKey(c,c->db,c->argv[1]); notifyKeyspaceEvent(NOTIFY_HASH,"hset",c->argv[1],c->db->id); server.dirty++; } void hsetCommand(client *c) { int i, created = 0; robj *o; if ((c->argc % 2) == 1) { addReplyErrorArity(c); return; } if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return; hashTypeTryConversion(c->db,o,c->argv,2,c->argc-1); for (i = 2; i < c->argc; i += 2) created += !hashTypeSet(c->db, o,c->argv[i]->ptr,c->argv[i+1]->ptr,HASH_SET_COPY); /* HMSET (deprecated) and HSET return value is different. */ char *cmdname = c->argv[0]->ptr; if (cmdname[1] == 's' || cmdname[1] == 'S') { /* HSET */ addReplyLongLong(c, created); } else { /* HMSET */ addReply(c, shared.ok); } signalModifiedKey(c,c->db,c->argv[1]); notifyKeyspaceEvent(NOTIFY_HASH,"hset",c->argv[1],c->db->id); server.dirty += (c->argc - 2)/2; } void hincrbyCommand(client *c) { long long value, incr, oldvalue; robj *o; sds new; unsigned char *vstr; unsigned int vlen; if (getLongLongFromObjectOrReply(c,c->argv[3],&incr,NULL) != C_OK) return; if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return; GetFieldRes res = hashTypeGetValue(c->db,o,c->argv[2]->ptr,&vstr,&vlen,&value, HFE_LAZY_EXPIRE); if (res == GETF_OK) { if (vstr) { if (string2ll((char*)vstr,vlen,&value) == 0) { addReplyError(c,"hash value is not an integer"); return; } } /* Else hashTypeGetValue() already stored it into &value */ } else if ((res == GETF_NOT_FOUND) || (res == GETF_EXPIRED)) { value = 0; } else { /* Field expired and in turn hash deleted. Create new one! */ o = createHashObject(); dbAdd(c->db,c->argv[1],o); value = 0; } oldvalue = value; if ((incr < 0 && oldvalue < 0 && incr < (LLONG_MIN-oldvalue)) || (incr > 0 && oldvalue > 0 && incr > (LLONG_MAX-oldvalue))) { addReplyError(c,"increment or decrement would overflow"); return; } value += incr; new = sdsfromlonglong(value); hashTypeSet(c->db, o,c->argv[2]->ptr,new,HASH_SET_TAKE_VALUE | HASH_SET_KEEP_TTL); addReplyLongLong(c,value); signalModifiedKey(c,c->db,c->argv[1]); notifyKeyspaceEvent(NOTIFY_HASH,"hincrby",c->argv[1],c->db->id); server.dirty++; } void hincrbyfloatCommand(client *c) { long double value, incr; long long ll; robj *o; sds new; unsigned char *vstr; unsigned int vlen; if (getLongDoubleFromObjectOrReply(c,c->argv[3],&incr,NULL) != C_OK) return; if (isnan(incr) || isinf(incr)) { addReplyError(c,"value is NaN or Infinity"); return; } if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return; GetFieldRes res = hashTypeGetValue(c->db, o,c->argv[2]->ptr,&vstr,&vlen,&ll, HFE_LAZY_EXPIRE); if (res == GETF_OK) { if (vstr) { if (string2ld((char*)vstr,vlen,&value) == 0) { addReplyError(c,"hash value is not a float"); return; } } else { value = (long double)ll; } } else if ((res == GETF_NOT_FOUND) || (res == GETF_EXPIRED)) { value = 0; } else { /* Field expired and in turn hash deleted. Create new one! */ o = createHashObject(); dbAdd(c->db,c->argv[1],o); value = 0; } value += incr; if (isnan(value) || isinf(value)) { addReplyError(c,"increment would produce NaN or Infinity"); return; } char buf[MAX_LONG_DOUBLE_CHARS]; int len = ld2string(buf,sizeof(buf),value,LD_STR_HUMAN); new = sdsnewlen(buf,len); hashTypeSet(c->db, o,c->argv[2]->ptr,new,HASH_SET_TAKE_VALUE | HASH_SET_KEEP_TTL); addReplyBulkCBuffer(c,buf,len); signalModifiedKey(c,c->db,c->argv[1]); notifyKeyspaceEvent(NOTIFY_HASH,"hincrbyfloat",c->argv[1],c->db->id); server.dirty++; /* Always replicate HINCRBYFLOAT as an HSET command with the final value * in order to make sure that differences in float precision or formatting * will not create differences in replicas or after an AOF restart. */ robj *newobj; newobj = createRawStringObject(buf,len); rewriteClientCommandArgument(c,0,shared.hset); rewriteClientCommandArgument(c,3,newobj); decrRefCount(newobj); } static GetFieldRes addHashFieldToReply(client *c, robj *o, sds field, int hfeFlags) { if (o == NULL) { addReplyNull(c); return GETF_NOT_FOUND; } unsigned char *vstr = NULL; unsigned int vlen = UINT_MAX; long long vll = LLONG_MAX; GetFieldRes res = hashTypeGetValue(c->db, o, field, &vstr, &vlen, &vll, hfeFlags); if (res == GETF_OK) { if (vstr) { addReplyBulkCBuffer(c, vstr, vlen); } else { addReplyBulkLongLong(c, vll); } } else { addReplyNull(c); } return res; } void hgetCommand(client *c) { robj *o; if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) == NULL || checkType(c,o,OBJ_HASH)) return; addHashFieldToReply(c, o, c->argv[2]->ptr, HFE_LAZY_EXPIRE); } void hmgetCommand(client *c) { GetFieldRes res = GETF_OK; robj *o; int i; int expired = 0, deleted = 0; /* 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. */ o = lookupKeyRead(c->db, c->argv[1]); if (checkType(c,o,OBJ_HASH)) return; addReplyArrayLen(c, c->argc-2); for (i = 2; i < c->argc ; i++) { if (!deleted) { res = addHashFieldToReply(c, o, c->argv[i]->ptr, HFE_LAZY_NO_NOTIFICATION); expired += (res == GETF_EXPIRED); deleted += (res == GETF_EXPIRED_HASH); } else { /* If hash got lazy expired since all fields are expired (o is invalid), * then fill the rest with trivial nulls and return. */ addReplyNull(c); } } 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) { robj *o; int j, deleted = 0, keyremoved = 0; if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL || checkType(c,o,OBJ_HASH)) return; /* Hash field expiration is optimized to avoid frequent update global HFE DS for * each field deletion. Eventually active-expiration will run and update or remove * the hash from global HFE DS gracefully. Nevertheless, statistic "subexpiry" * might reflect wrong number of hashes with HFE to the user if it is the last * field with expiration. The following logic checks if this is indeed the last * field with expiration and removes it from global HFE DS. */ int isHFE = hashTypeIsFieldsWithExpire(o); for (j = 2; j < c->argc; j++) { if (hashTypeDelete(o,c->argv[j]->ptr,1)) { deleted++; if (hashTypeLength(o, 0) == 0) { dbDelete(c->db,c->argv[1]); keyremoved = 1; break; } } } if (deleted) { signalModifiedKey(c,c->db,c->argv[1]); notifyKeyspaceEvent(NOTIFY_HASH,"hdel",c->argv[1],c->db->id); if (keyremoved) { notifyKeyspaceEvent(NOTIFY_GENERIC, "del", c->argv[1], c->db->id); } else { if (isHFE && (hashTypeIsFieldsWithExpire(o) == 0)) /* is it last HFE */ ebRemove(&c->db->hexpires, &hashExpireBucketsType, o); } server.dirty += deleted; } addReplyLongLong(c,deleted); } void hlenCommand(client *c) { robj *o; if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL || checkType(c,o,OBJ_HASH)) return; addReplyLongLong(c,hashTypeLength(o, 0)); } void hstrlenCommand(client *c) { robj *o; unsigned char *vstr = NULL; unsigned int vlen = UINT_MAX; long long vll = LLONG_MAX; if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL || checkType(c,o,OBJ_HASH)) return; 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) { addReply(c, shared.czero); return; } size_t len = vstr ? vlen : sdigits10(vll); addReplyLongLong(c,len); } static void addHashIteratorCursorToReply(client *c, hashTypeIterator *hi, int what) { if (hi->encoding == OBJ_ENCODING_LISTPACK || hi->encoding == OBJ_ENCODING_LISTPACK_EX) { unsigned char *vstr = NULL; unsigned int vlen = UINT_MAX; long long vll = LLONG_MAX; hashTypeCurrentFromListpack(hi, what, &vstr, &vlen, &vll, NULL); if (vstr) addReplyBulkCBuffer(c, vstr, vlen); else addReplyBulkLongLong(c, vll); } else if (hi->encoding == OBJ_ENCODING_HT) { char *value; size_t len; hashTypeCurrentFromHashTable(hi, what, &value, &len, NULL); addReplyBulkCBuffer(c, value, len); } else { serverPanic("Unknown hash encoding"); } } void genericHgetallCommand(client *c, int flags) { robj *o; hashTypeIterator *hi; int length, count = 0; robj *emptyResp = (flags & OBJ_HASH_KEY && flags & OBJ_HASH_VALUE) ? shared.emptymap[c->resp] : shared.emptyarray; if ((o = lookupKeyReadOrReply(c,c->argv[1],emptyResp)) == NULL || checkType(c,o,OBJ_HASH)) return; /* We return a map if the user requested keys and values, like in the * HGETALL case. Otherwise to use a flat array makes more sense. */ if ((length = hashTypeLength(o, 1 /*subtractExpiredFields*/)) == 0) { addReply(c, emptyResp); return; } if (flags & OBJ_HASH_KEY && flags & OBJ_HASH_VALUE) { addReplyMapLen(c, length); } else { addReplyArrayLen(c, length); } hi = hashTypeInitIterator(o); while (hashTypeNext(hi, 1 /*skipExpiredFields*/) != C_ERR) { if (flags & OBJ_HASH_KEY) { addHashIteratorCursorToReply(c, hi, OBJ_HASH_KEY); count++; } if (flags & OBJ_HASH_VALUE) { addHashIteratorCursorToReply(c, hi, OBJ_HASH_VALUE); count++; } } hashTypeReleaseIterator(hi); /* Make sure we returned the right number of elements. */ if (flags & OBJ_HASH_KEY && flags & OBJ_HASH_VALUE) count /= 2; serverAssert(count == length); } void hkeysCommand(client *c) { genericHgetallCommand(c,OBJ_HASH_KEY); } void hvalsCommand(client *c) { genericHgetallCommand(c,OBJ_HASH_VALUE); } void hgetallCommand(client *c) { genericHgetallCommand(c,OBJ_HASH_KEY|OBJ_HASH_VALUE); } void hexistsCommand(client *c) { robj *o; if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL || checkType(c,o,OBJ_HASH)) return; addReply(c,hashTypeExists(c->db,o,c->argv[2]->ptr,HFE_LAZY_EXPIRE, NULL) ? shared.cone : shared.czero); } void hscanCommand(client *c) { robj *o; unsigned long long cursor; if (parseScanCursorOrReply(c,c->argv[2],&cursor) == C_ERR) return; if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptyscan)) == NULL || checkType(c,o,OBJ_HASH)) return; scanGenericCommand(c,o,cursor); } static void hrandfieldReplyWithListpack(client *c, unsigned int count, listpackEntry *keys, listpackEntry *vals) { for (unsigned long i = 0; i < count; i++) { if (vals && c->resp > 2) addReplyArrayLen(c,2); if (keys[i].sval) addReplyBulkCBuffer(c, keys[i].sval, keys[i].slen); else addReplyBulkLongLong(c, keys[i].lval); if (vals) { if (vals[i].sval) addReplyBulkCBuffer(c, vals[i].sval, vals[i].slen); else addReplyBulkLongLong(c, vals[i].lval); } } } /* How many times bigger should be the hash compared to the requested size * for us to not use the "remove elements" strategy? Read later in the * implementation for more info. */ #define HRANDFIELD_SUB_STRATEGY_MUL 3 /* If client is trying to ask for a very large number of random elements, * queuing may consume an unlimited amount of memory, so we want to limit * the number of randoms per time. */ #define HRANDFIELD_RANDOM_SAMPLE_LIMIT 1000 void hrandfieldWithCountCommand(client *c, long l, int withvalues) { unsigned long count, size; int uniq = 1; robj *hash; if ((hash = lookupKeyReadOrReply(c,c->argv[1],shared.emptyarray)) == NULL || checkType(c,hash,OBJ_HASH)) return; if(l >= 0) { count = (unsigned long) l; } else { count = -l; 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 == 0) { addReply(c,shared.emptyarray); return; } /* CASE 1: The count was negative, so the extraction method is just: * "return N random elements" sampling the whole set every time. * This case is trivial and can be served without auxiliary data * structures. This case is the only one that also needs to return the * elements in random order. */ if (!uniq || count == 1) { if (withvalues && c->resp == 2) addReplyArrayLen(c, count*2); else addReplyArrayLen(c, count); if (hash->encoding == OBJ_ENCODING_HT) { while (count--) { dictEntry *de = dictGetFairRandomKey(hash->ptr); hfield field = dictGetKey(de); sds value = dictGetVal(de); if (withvalues && c->resp > 2) addReplyArrayLen(c,2); addReplyBulkCBuffer(c, field, hfieldlen(field)); if (withvalues) addReplyBulkCBuffer(c, value, sdslen(value)); if (c->flags & CLIENT_CLOSE_ASAP) break; } } else if (hash->encoding == OBJ_ENCODING_LISTPACK || hash->encoding == OBJ_ENCODING_LISTPACK_EX) { listpackEntry *keys, *vals = NULL; unsigned long limit, sample_count; unsigned char *lp = hashTypeListpackGetLp(hash); int tuple_len = hash->encoding == OBJ_ENCODING_LISTPACK ? 2 : 3; limit = count > HRANDFIELD_RANDOM_SAMPLE_LIMIT ? HRANDFIELD_RANDOM_SAMPLE_LIMIT : count; keys = zmalloc(sizeof(listpackEntry)*limit); if (withvalues) vals = zmalloc(sizeof(listpackEntry)*limit); while (count) { sample_count = count > limit ? limit : count; count -= sample_count; lpRandomPairs(lp, sample_count, keys, vals, tuple_len); hrandfieldReplyWithListpack(c, sample_count, keys, vals); if (c->flags & CLIENT_CLOSE_ASAP) break; } zfree(keys); zfree(vals); } return; } /* Initiate reply count, RESP3 responds with nested array, RESP2 with flat one. */ long reply_size = count < size ? count : size; if (withvalues && c->resp == 2) addReplyArrayLen(c, reply_size*2); else addReplyArrayLen(c, reply_size); /* CASE 2: * The number of requested elements is greater than the number of * elements inside the hash: simply return the whole hash. */ if(count >= size) { hashTypeIterator *hi = hashTypeInitIterator(hash); while (hashTypeNext(hi, 0) != C_ERR) { if (withvalues && c->resp > 2) addReplyArrayLen(c,2); addHashIteratorCursorToReply(c, hi, OBJ_HASH_KEY); if (withvalues) addHashIteratorCursorToReply(c, hi, OBJ_HASH_VALUE); } hashTypeReleaseIterator(hi); return; } /* CASE 2.5 listpack only. Sampling unique elements, in non-random order. * Listpack encoded hashes are meant to be relatively small, so * HRANDFIELD_SUB_STRATEGY_MUL isn't necessary and we rather not make * copies of the entries. Instead, we emit them directly to the output * buffer. * * And it is inefficient to repeatedly pick one random element from a * listpack in CASE 4. So we use this instead. */ if (hash->encoding == OBJ_ENCODING_LISTPACK || hash->encoding == OBJ_ENCODING_LISTPACK_EX) { unsigned char *lp = hashTypeListpackGetLp(hash); int tuple_len = hash->encoding == OBJ_ENCODING_LISTPACK ? 2 : 3; listpackEntry *keys, *vals = NULL; keys = zmalloc(sizeof(listpackEntry)*count); if (withvalues) vals = zmalloc(sizeof(listpackEntry)*count); serverAssert(lpRandomPairsUnique(lp, count, keys, vals, tuple_len) == count); hrandfieldReplyWithListpack(c, count, keys, vals); zfree(keys); zfree(vals); return; } /* CASE 3: * The number of elements inside the hash of type dict is not greater than * HRANDFIELD_SUB_STRATEGY_MUL times the number of requested elements. * In this case we create an array of dictEntry pointers from the original hash, * and subtract random elements to reach the requested number of elements. * * 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 * used into CASE 4 is highly inefficient. */ if (count*HRANDFIELD_SUB_STRATEGY_MUL > size) { /* Hashtable encoding (generic implementation) */ dict *ht = hash->ptr; dictIterator *di; dictEntry *de; unsigned long idx = 0; /* Allocate a temporary array of pointers to stored key-values in dict and * assist it to remove random elements to reach the right count. */ struct FieldValPair { hfield field; sds value; } *pairs = zmalloc(sizeof(struct FieldValPair) * size); /* Add all the elements into the temporary array. */ di = dictGetIterator(ht); while((de = dictNext(di)) != NULL) pairs[idx++] = (struct FieldValPair) {dictGetKey(de), dictGetVal(de)}; dictReleaseIterator(di); /* Remove random elements to reach the right count. */ while (size > count) { unsigned long toDiscardIdx = rand() % size; pairs[toDiscardIdx] = pairs[--size]; } /* Reply with what's in the array */ for (idx = 0; idx < size; idx++) { if (withvalues && c->resp > 2) addReplyArrayLen(c,2); addReplyBulkCBuffer(c, pairs[idx].field, hfieldlen(pairs[idx].field)); if (withvalues) addReplyBulkCBuffer(c, pairs[idx].value, sdslen(pairs[idx].value)); } zfree(pairs); } /* CASE 4: We have a big hash compared to the requested number of elements. * In this case we can simply get random elements from the hash and add * to the temporary hash, trying to eventually get enough unique elements * to reach the specified count. */ 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) */ unsigned long added = 0; while(added < count) { 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 * free it, otherwise increment the number of objects we have * in the result dictionary. */ if (dictAdd(dictUnique, field, value) != DICT_OK) continue; added++; /* We can reply right away, so that we don't need to store the value in the dict. */ if (withvalues && c->resp > 2) addReplyArrayLen(c,2); addReplyBulkCBuffer(c, field, hfieldlen(field)); if (withvalues) addReplyBulkCBuffer(c, value, sdslen(value)); } /* Release memory */ dictRelease(dictUnique); } } /* * HRANDFIELD - Return a random field from the hash value stored at key. * CLI usage: HRANDFIELD key [ [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) { long l; int withvalues = 0; robj *hash; CommonEntry ele; if (c->argc >= 3) { if (getRangeLongFromObjectOrReply(c,c->argv[2],-LONG_MAX,LONG_MAX,&l,NULL) != C_OK) return; if (c->argc > 4 || (c->argc == 4 && strcasecmp(c->argv[3]->ptr,"withvalues"))) { addReplyErrorObject(c,shared.syntaxerr); return; } else if (c->argc == 4) { withvalues = 1; if (l < -LONG_MAX/2 || l > LONG_MAX/2) { addReplyError(c,"value is out of range"); return; } } hrandfieldWithCountCommand(c, l, withvalues); return; } /* Handle variant without argument. Reply with simple bulk string */ if ((hash = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp]))== NULL || checkType(c,hash,OBJ_HASH)) { 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); if (ele.sval) addReplyBulkCBuffer(c, ele.sval, ele.slen); else addReplyBulkLongLong(c, ele.lval); } /*----------------------------------------------------------------------------- * Hash Field with optional expiry (based on mstr) *----------------------------------------------------------------------------*/ static hfield _hfieldNew(const void *field, size_t fieldlen, int withExpireMeta, int trymalloc) { if (!withExpireMeta) return mstrNew(field, fieldlen, trymalloc); hfield hf = mstrNewWithMeta(&mstrFieldKind, field, fieldlen, (mstrFlags) 1 << HFIELD_META_EXPIRE, trymalloc); if (!hf) return NULL; ExpireMeta *expireMeta = mstrMetaRef(hf, &mstrFieldKind, HFIELD_META_EXPIRE); /* as long as it is not inside ebuckets, it is considered trash */ expireMeta->trash = 1; return hf; } /* if expireAt is 0, then expireAt is ignored and no metadata is attached */ hfield hfieldNew(const void *field, size_t fieldlen, int withExpireMeta) { return _hfieldNew(field, fieldlen, withExpireMeta, 0); } hfield hfieldTryNew(const void *field, size_t fieldlen, int withExpireMeta) { return _hfieldNew(field, fieldlen, withExpireMeta, 1); } int hfieldIsExpireAttached(hfield field) { return mstrIsMetaAttached(field) && mstrGetFlag(field, (int) HFIELD_META_EXPIRE); } static ExpireMeta* hfieldGetExpireMeta(const eItem field) { /* extract the expireMeta from the field of type mstr */ return mstrMetaRef(field, &mstrFieldKind, (int) HFIELD_META_EXPIRE); } /* returned value is unix time in milliseconds */ uint64_t hfieldGetExpireTime(hfield field) { if (!hfieldIsExpireAttached(field)) return EB_EXPIRE_TIME_INVALID; ExpireMeta *expireMeta = mstrMetaRef(field, &mstrFieldKind, (int) HFIELD_META_EXPIRE); if (expireMeta->trash) return EB_EXPIRE_TIME_INVALID; return ebGetMetaExpTime(expireMeta); } /* Remove TTL from the field. Assumed ExpireMeta is attached and has valid value */ static void hfieldPersist(robj *hashObj, hfield field) { uint64_t fieldExpireTime = hfieldGetExpireTime(field); if (fieldExpireTime == EB_EXPIRE_TIME_INVALID) return; /* if field is set with expire, then dict must has HFE metadata attached */ dict *d = hashObj->ptr; dictExpireMetadata *dictExpireMeta = (dictExpireMetadata *)dictMetadata(d); /* If field has valid expiry then dict must have valid metadata as well */ serverAssert(dictExpireMeta->expireMeta.trash == 0); /* Remove field from private HFE DS */ ebRemove(&dictExpireMeta->hfe, &hashFieldExpireBucketsType, field); /* Don't have to update global HFE DS. It's unnecessary. Implementing this * would introduce significant complexity and overhead for an operation that * isn't critical. In the worst case scenario, the hash will be efficiently * updated later by an active-expire operation, or it will be removed by the * hash's dbGenericDelete() function. */ } int hfieldIsExpired(hfield field) { /* Condition remains valid even if hfieldGetExpireTime() returns EB_EXPIRE_TIME_INVALID, * as the constant is equivalent to (EB_EXPIRE_TIME_MAX + 1). */ return ( (mstime_t)hfieldGetExpireTime(field) < commandTimeSnapshot()); } /*----------------------------------------------------------------------------- * Hash Field Expiration (HFE) *----------------------------------------------------------------------------*/ /* Can be called either by active-expire cron job or query from the client */ static void propagateHashFieldDeletion(redisDb *db, sds key, char *field, size_t fieldLen) { robj *argv[] = { shared.hdel, createStringObject((char*) key, sdslen(key)), createStringObject(field, fieldLen) }; enterExecutionUnit(1, 0); int prev_replication_allowed = server.replication_allowed; server.replication_allowed = 1; alsoPropagate(db->id,argv, 3, PROPAGATE_AOF|PROPAGATE_REPL); server.replication_allowed = prev_replication_allowed; exitExecutionUnit(); /* Propagate the HDEL command */ postExecutionUnitOperations(); decrRefCount(argv[1]); decrRefCount(argv[2]); } /* Called during active expiration of hash-fields. Propagate to replica & Delete. */ static ExpireAction onFieldExpire(eItem item, void *ctx) { OnFieldExpireCtx *expCtx = ctx; hfield hf = item; dict *d = expCtx->hashObj->ptr; dictExpireMetadata *dictExpireMeta = (dictExpireMetadata *) dictMetadata(d); propagateHashFieldDeletion(expCtx->db, dictExpireMeta->key, hf, hfieldlen(hf)); serverAssert(hashTypeDelete(expCtx->hashObj, hf, 0) == 1); server.stat_expired_subkeys++; return ACT_REMOVE_EXP_ITEM; } /* Retrieve the ExpireMeta associated with the hash. * The caller is responsible for ensuring that it is indeed attached. */ static ExpireMeta *hashGetExpireMeta(const eItem hash) { robj *hashObj = (robj *)hash; if (hashObj->encoding == OBJ_ENCODING_LISTPACK_EX) { listpackEx *lpt = hashObj->ptr; return &lpt->meta; } else if (hashObj->encoding == OBJ_ENCODING_HT) { dict *d = hashObj->ptr; dictExpireMetadata *dictExpireMeta = (dictExpireMetadata *) dictMetadata(d); return &dictExpireMeta->expireMeta; } else { serverPanic("Unknown encoding: %d", hashObj->encoding); } } /* HTTL key */ static void httlGenericCommand(client *c, const char *cmd, long long basetime, int unit) { UNUSED(cmd); robj *hashObj; long numFields = 0, numFieldsAt = 3; /* Read the hash object */ hashObj = lookupKeyRead(c->db, c->argv[1]); if (checkType(c, hashObj, OBJ_HASH)) return; if (strcasecmp(c->argv[numFieldsAt-1]->ptr, "FIELDS")) { addReplyError(c, "Mandatory argument FIELDS is missing or not at the right position"); return; } /* Read number of fields */ if (getRangeLongFromObjectOrReply(c, c->argv[numFieldsAt], 1, LONG_MAX, &numFields, "Number of fields must be a positive integer") != C_OK) return; /* Verify `numFields` is consistent with number of arguments */ if (numFields != (c->argc - numFieldsAt - 1)) { 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; } if (hashObj->encoding == OBJ_ENCODING_LISTPACK) { void *lp = hashObj->ptr; addReplyArrayLen(c, numFields); for (int i = 0 ; i < numFields ; i++) { sds field = c->argv[numFieldsAt+1+i]->ptr; void *fptr = lpFirst(lp); if (fptr != NULL) fptr = lpFind(lp, fptr, (unsigned char *) field, sdslen(field), 1); if (!fptr) addReplyLongLong(c, HFE_GET_NO_FIELD); else addReplyLongLong(c, HFE_GET_NO_TTL); } return; } else if (hashObj->encoding == OBJ_ENCODING_LISTPACK_EX) { listpackEx *lpt = hashObj->ptr; addReplyArrayLen(c, numFields); for (int i = 0 ; i < numFields ; i++) { long long expire; sds field = c->argv[numFieldsAt+1+i]->ptr; void *fptr = lpFirst(lpt->lp); if (fptr != NULL) fptr = lpFind(lpt->lp, fptr, (unsigned char *) field, sdslen(field), 2); if (!fptr) { addReplyLongLong(c, HFE_GET_NO_FIELD); continue; } fptr = lpNext(lpt->lp, fptr); serverAssert(fptr); fptr = lpNext(lpt->lp, fptr); serverAssert(fptr && lpGetIntegerValue(fptr, &expire)); if (expire == HASH_LP_NO_TTL) { addReplyLongLong(c, HFE_GET_NO_TTL); continue; } if (expire <= commandTimeSnapshot()) { addReplyLongLong(c, HFE_GET_NO_FIELD); continue; } if (unit == UNIT_SECONDS) addReplyLongLong(c, (expire + 999 - basetime) / 1000); else addReplyLongLong(c, (expire - basetime)); } return; } else if (hashObj->encoding == OBJ_ENCODING_HT) { dict *d = hashObj->ptr; addReplyArrayLen(c, numFields); for (int i = 0 ; i < numFields ; i++) { sds field = c->argv[numFieldsAt+1+i]->ptr; dictEntry *de = dictFind(d, field); if (de == NULL) { addReplyLongLong(c, HFE_GET_NO_FIELD); continue; } hfield hf = dictGetKey(de); uint64_t expire = hfieldGetExpireTime(hf); if (expire == EB_EXPIRE_TIME_INVALID) { addReplyLongLong(c, HFE_GET_NO_TTL); /* no ttl */ continue; } if ( (long long) expire < commandTimeSnapshot()) { addReplyLongLong(c, HFE_GET_NO_FIELD); continue; } if (unit == UNIT_SECONDS) addReplyLongLong(c, (expire + 999 - basetime) / 1000); else addReplyLongLong(c, (expire - basetime)); } return; } else { serverPanic("Unknown encoding: %d", hashObj->encoding); } } /* This is the generic command implementation for HEXPIRE, HPEXPIRE, HEXPIREAT * and HPEXPIREAT. Because the command second argument may be relative or absolute * the "basetime" argument is used to signal what the base time is (either 0 * for *AT variants of the command, or the current time for relative expires). * * unit is either UNIT_SECONDS or UNIT_MILLISECONDS, and is only used for * the argv[2] parameter. The basetime is always specified in milliseconds. * * 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) { long numFields = 0, numFieldsAt = 4; long long expire; /* unix time in msec */ int fieldAt, fieldsNotSet = 0, expireSetCond = 0; robj *hashObj, *keyArg = c->argv[1], *expireArg = c->argv[2]; /* Read the hash object */ hashObj = lookupKeyWrite(c->db, keyArg); if (checkType(c, hashObj, OBJ_HASH)) return; /* Read the expiry time from command */ if (getLongLongFromObjectOrReply(c, expireArg, &expire, NULL) != C_OK) return; if (expire < 0) { addReplyError(c,"invalid expire time, must be >= 0"); return; } if (unit == UNIT_SECONDS) { if (expire > (long long) HFE_MAX_ABS_TIME_MSEC / 1000) { addReplyErrorExpireTime(c); return; } expire *= 1000; } /* Ensure that the final absolute Unix timestamp does not exceed EB_EXPIRE_TIME_MAX. */ if (expire > (long long) HFE_MAX_ABS_TIME_MSEC - basetime) { addReplyErrorExpireTime(c); return; } expire += basetime; /* Read optional expireSetCond [NX|XX|GT|LT] */ char *optArg = c->argv[3]->ptr; if (!strcasecmp(optArg, "nx")) { expireSetCond = HFE_NX; ++numFieldsAt; } else if (!strcasecmp(optArg, "xx")) { expireSetCond = HFE_XX; ++numFieldsAt; } else if (!strcasecmp(optArg, "gt")) { expireSetCond = HFE_GT; ++numFieldsAt; } else if (!strcasecmp(optArg, "lt")) { expireSetCond = HFE_LT; ++numFieldsAt; } if (strcasecmp(c->argv[numFieldsAt-1]->ptr, "FIELDS")) { addReplyError(c, "Mandatory argument FIELDS is missing or not at the right position"); return; } /* Read number of fields */ if (getRangeLongFromObjectOrReply(c, c->argv[numFieldsAt], 1, LONG_MAX, &numFields, "Parameter `numFields` should be greater than 0") != C_OK) return; /* Verify `numFields` is consistent with number of arguments */ if (numFields != (c->argc - numFieldsAt - 1)) { 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, HSETEX_NO_FIELD); } return; } HashTypeSetEx exCtx; hashTypeSetExInit(keyArg, hashObj, c, c->db, cmd, expireSetCond, &exCtx); addReplyArrayLen(c, numFields); fieldAt = numFieldsAt + 1; while (fieldAt < c->argc) { 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); } hashTypeSetExDone(&exCtx); /* 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 */ if (c->cmd->proc != hpexpireatCommand) { rewriteClientCommandArgument(c,0,shared.hpexpireat); } /* rewrite expiration time to unix time in msec */ if (basetime != 0 || unit == UNIT_SECONDS) { robj *expireObj = createStringObjectFromLongLong(expire); rewriteClientCommandArgument(c, 2, expireObj); decrRefCount(expireObj); } } /* HPEXPIRE key milliseconds [ NX | XX | GT | LT] numfields */ void hpexpireCommand(client *c) { hexpireGenericCommand(c,"hpexpire", commandTimeSnapshot(),UNIT_MILLISECONDS); } /* HEXPIRE key seconds [NX | XX | GT | LT] numfields */ void hexpireCommand(client *c) { hexpireGenericCommand(c,"hexpire", commandTimeSnapshot(),UNIT_SECONDS); } /* HEXPIREAT key unix-time-seconds [NX | XX | GT | LT] numfields */ void hexpireatCommand(client *c) { hexpireGenericCommand(c,"hexpireat", 0,UNIT_SECONDS); } /* HPEXPIREAT key unix-time-milliseconds [NX | XX | GT | LT] numfields */ void hpexpireatCommand(client *c) { hexpireGenericCommand(c,"hpexpireat", 0,UNIT_MILLISECONDS); } /* for each specified field: get the remaining time to live in seconds*/ /* HTTL key numfields */ void httlCommand(client *c) { httlGenericCommand(c, "httl", commandTimeSnapshot(), UNIT_SECONDS); } /* HPTTL key numfields */ void hpttlCommand(client *c) { httlGenericCommand(c, "hpttl", commandTimeSnapshot(), UNIT_MILLISECONDS); } /* HEXPIRETIME key numFields */ void hexpiretimeCommand(client *c) { httlGenericCommand(c, "hexpiretime", 0, UNIT_SECONDS); } /* HPEXPIRETIME key numFields */ void hpexpiretimeCommand(client *c) { httlGenericCommand(c, "hexpiretime", 0, UNIT_MILLISECONDS); } /* HPERSIST key */ void hpersistCommand(client *c) { robj *hashObj; long numFields = 0, numFieldsAt = 3; int changed = 0; /* Used to determine whether to send a notification. */ /* Read the hash object */ hashObj = lookupKeyWrite(c->db, c->argv[1]); if (checkType(c, hashObj, OBJ_HASH)) return; if (strcasecmp(c->argv[numFieldsAt-1]->ptr, "FIELDS")) { addReplyError(c, "Mandatory argument FIELDS is missing or not at the right position"); return; } /* Read number of fields */ if (getRangeLongFromObjectOrReply(c, c->argv[numFieldsAt], 1, LONG_MAX, &numFields, "Number of fields must be a positive integer") != C_OK) return; /* Verify `numFields` is consistent with number of arguments */ if (numFields != (c->argc - numFieldsAt - 1)) { 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; } if (hashObj->encoding == OBJ_ENCODING_LISTPACK) { addReplyArrayLen(c, numFields); for (int i = 0 ; i < numFields ; i++) { sds field = c->argv[numFieldsAt + 1 + i]->ptr; unsigned char *fptr, *zl = hashObj->ptr; fptr = lpFirst(zl); if (fptr != NULL) fptr = lpFind(zl, fptr, (unsigned char *) field, sdslen(field), 1); if (!fptr) addReplyLongLong(c, HFE_PERSIST_NO_FIELD); else addReplyLongLong(c, HFE_PERSIST_NO_TTL); } return; } else if (hashObj->encoding == OBJ_ENCODING_LISTPACK_EX) { long long prevExpire; unsigned char *fptr, *vptr, *tptr; listpackEx *lpt = hashObj->ptr; addReplyArrayLen(c, numFields); for (int i = 0 ; i < numFields ; i++) { sds field = c->argv[numFieldsAt + 1 + i]->ptr; fptr = lpFirst(lpt->lp); if (fptr != NULL) fptr = lpFind(lpt->lp, fptr, (unsigned char*)field, sdslen(field), 2); if (!fptr) { addReplyLongLong(c, HFE_PERSIST_NO_FIELD); continue; } vptr = lpNext(lpt->lp, fptr); serverAssert(vptr); tptr = lpNext(lpt->lp, vptr); serverAssert(tptr && lpGetIntegerValue(tptr, &prevExpire)); if (prevExpire == HASH_LP_NO_TTL) { addReplyLongLong(c, HFE_PERSIST_NO_TTL); continue; } if (prevExpire < commandTimeSnapshot()) { addReplyLongLong(c, HFE_PERSIST_NO_FIELD); continue; } listpackExUpdateExpiry(hashObj, field, fptr, vptr, HASH_LP_NO_TTL); addReplyLongLong(c, HFE_PERSIST_OK); changed = 1; } } else if (hashObj->encoding == OBJ_ENCODING_HT) { dict *d = hashObj->ptr; addReplyArrayLen(c, numFields); for (int i = 0 ; i < numFields ; i++) { sds field = c->argv[numFieldsAt + 1 + i]->ptr; dictEntry *de = dictFind(d, field); if (de == NULL) { addReplyLongLong(c, HFE_PERSIST_NO_FIELD); continue; } hfield hf = dictGetKey(de); uint64_t expire = hfieldGetExpireTime(hf); if (expire == EB_EXPIRE_TIME_INVALID) { addReplyLongLong(c, HFE_PERSIST_NO_TTL); continue; } /* Already expired. Pretend there is no such field */ if ( (long long) expire < commandTimeSnapshot()) { addReplyLongLong(c, HFE_PERSIST_NO_FIELD); continue; } hfieldPersist(hashObj, hf); addReplyLongLong(c, HFE_PERSIST_OK); changed = 1; } } else { serverPanic("Unknown encoding: %d", hashObj->encoding); } /* Generates a hpersist event if the expiry time associated with any field * has been successfully deleted. */ if (changed) { notifyKeyspaceEvent(NOTIFY_HASH, "hpersist", c->argv[1], c->db->id); signalModifiedKey(c, c->db, c->argv[1]); server.dirty++; } }