gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

Use metadata to handle the reference relationship between kvstore and dict (#966)

Browse Source 
Feature `one-dict-per-slot` refactors the database, and part of it
involved splitting the rehashing list from the global level back to the
database level, or more specifically, the kvstore level. This change is
fine, and it also simplifies the process of swapping databases, which is
good. And it should not have a major impact on the efficiency of
incremental rehashing.

To implement the kvstore-level rehashing list, each `dict` under the
`kvstore` needs to know which `kvstore` it belongs. However, kvstore did
not insert the reference relationship into the `dict` itself, instead,
it placed it in the `dictType`. In my view, this is a somewhat odd way.
Theoretically, `dictType` is just a collection of function handles, a
kind of virtual type that can be referenced globally, not an entity. But
now the `dictType` is instantiated, with each `kvstore` owning an actual
`dictType`, which in turn holds a reverse reference to the `kvstore`'s
resource pointer. This design is somewhat uncomfortable for me.

I think the `dictType` should not be instantiated. The references
between actual resources (`kvstore` and `dict`) should occur between
specific objects, rather than force materializing the `dictType`, which
is supposed to be virtual.

---------

Signed-off-by: zhaozhao.zz <zhaozhao.zz@alibaba-inc.com>
This commit is contained in:
zhaozhao.zz 2024-09-02 22:35:24 +08:00 committed by GitHub
parent 70624ea63d
commit 32116d09bb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 53 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/db.c
View File

@ -584,8 +584,8 @@ serverDb *initTempDb(void) {
serverDb *tempDb = zcalloc(sizeof(serverDb) * server.dbnum);
for (int i = 0; i < server.dbnum; i++) {
tempDb[i].id = i;
tempDb[i].keys = kvstoreCreate(&dbDictType, slot_count_bits, flags);
tempDb[i].expires = kvstoreCreate(&dbExpiresDictType, slot_count_bits, flags);
tempDb[i].keys = kvstoreCreate(&kvstoreKeysDictType, slot_count_bits, flags);
tempDb[i].expires = kvstoreCreate(&kvstoreExpiresDictType, slot_count_bits, flags);
}
return tempDb;

4
src/dict.h
View File

@ -70,10 +70,6 @@ typedef struct dictType {
* computing the length of the key + header when buf is NULL. */
size_t (*embedKey)(unsigned char *buf, size_t buf_len, const void *key, unsigned char *header_size);
/* Data */
void *userdata;
/* Flags */
/* The 'no_value' flag, if set, indicates that values are not used, i.e. the
* dict is a set. When this flag is set, it's not possible to access the

32
src/kvstore.c
View File

@ -52,7 +52,7 @@ static dict *kvstoreIteratorNextDict(kvstoreIterator *kvs_it);
struct _kvstore {
int flags;
dictType dtype;
dictType *dtype;
dict **dicts;
long long num_dicts;
long long num_dicts_bits;
@ -86,6 +86,7 @@ struct _kvstoreDictIterator {
/* Dict metadata for database, used for record the position in rehashing list. */
typedef struct {
listNode *rehashing_node; /* list node in rehashing list */
kvstore *kvs;
} kvstoreDictMetadata;
/**********************************/
@ -166,7 +167,9 @@ static dict *createDictIfNeeded(kvstore *kvs, int didx) {
dict *d = kvstoreGetDict(kvs, didx);
if (d) return d;
kvs->dicts[didx] = dictCreate(&kvs->dtype);
kvs->dicts[didx] = dictCreate(kvs->dtype);
kvstoreDictMetadata *metadata = (kvstoreDictMetadata *)dictMetadata(kvs->dicts[didx]);
metadata->kvs = kvs;
kvs->allocated_dicts++;
return kvs->dicts[didx];
}
@ -197,9 +200,9 @@ static void freeDictIfNeeded(kvstore *kvs, int didx) {
* If there are multiple dicts, updates the bucket count for the given dictionary
* in a DB, bucket count incremented with the new ht size during the rehashing phase.
* If there's one dict, bucket count can be retrieved directly from single dict bucket. */
static void kvstoreDictRehashingStarted(dict *d) {
kvstore *kvs = d->type->userdata;
void kvstoreDictRehashingStarted(dict *d) {
kvstoreDictMetadata *metadata = (kvstoreDictMetadata *)dictMetadata(d);
kvstore *kvs = metadata->kvs;
listAddNodeTail(kvs->rehashing, d);
metadata->rehashing_node = listLast(kvs->rehashing);
@ -214,9 +217,9 @@ static void kvstoreDictRehashingStarted(dict *d) {
*
* Updates the bucket count for the given dictionary in a DB. It removes
* the old ht size of the dictionary from the total sum of buckets for a DB. */
static void kvstoreDictRehashingCompleted(dict *d) {
kvstore *kvs = d->type->userdata;
void kvstoreDictRehashingCompleted(dict *d) {
kvstoreDictMetadata *metadata = (kvstoreDictMetadata *)dictMetadata(d);
kvstore *kvs = metadata->kvs;
if (metadata->rehashing_node) {
listDelNode(kvs->rehashing, metadata->rehashing_node);
metadata->rehashing_node = NULL;
@ -230,7 +233,7 @@ static void kvstoreDictRehashingCompleted(dict *d) {
}
/* Returns the size of the DB dict metadata in bytes. */
static size_t kvstoreDictMetadataSize(dict *d) {
size_t kvstoreDictMetadataSize(dict *d) {
UNUSED(d);
return sizeof(kvstoreDictMetadata);
}
@ -253,20 +256,9 @@ kvstore *kvstoreCreate(dictType *type, int num_dicts_bits, int flags) {
assert(num_dicts_bits <= 16);
kvstore *kvs = zcalloc(sizeof(*kvs));
memcpy(&kvs->dtype, type, sizeof(kvs->dtype));
kvs->dtype = type;
kvs->flags = flags;
/* kvstore must be the one to set these callbacks, so we make sure the
* caller didn't do it */
assert(!type->userdata);
assert(!type->dictMetadataBytes);
assert(!type->rehashingStarted);
assert(!type->rehashingCompleted);
kvs->dtype.userdata = kvs;
kvs->dtype.dictMetadataBytes = kvstoreDictMetadataSize;
kvs->dtype.rehashingStarted = kvstoreDictRehashingStarted;
kvs->dtype.rehashingCompleted = kvstoreDictRehashingCompleted;
kvs->num_dicts_bits = num_dicts_bits;
kvs->num_dicts = 1 << kvs->num_dicts_bits;
kvs->dicts = zcalloc(sizeof(dict *) * kvs->num_dicts);
@ -760,7 +752,7 @@ void kvstoreDictLUTDefrag(kvstore *kvs, kvstoreDictLUTDefragFunction *defragfn)
}
uint64_t kvstoreGetHash(kvstore *kvs, const void *key) {
return kvs->dtype.hashFunction(key);
return kvs->dtype->hashFunction(key);
}
void *kvstoreDictFetchValue(kvstore *kvs, int didx, const void *key) {

4
src/kvstore.h
View File

@ -37,6 +37,10 @@ int kvstoreNumAllocatedDicts(kvstore *kvs);
int kvstoreNumDicts(kvstore *kvs);
uint64_t kvstoreGetHash(kvstore *kvs, const void *key);
void kvstoreDictRehashingStarted(dict *d);
void kvstoreDictRehashingCompleted(dict *d);
size_t kvstoreDictMetadataSize(dict *d);
/* kvstore iterator specific functions */
kvstoreIterator *kvstoreIteratorInit(kvstore *kvs);
void kvstoreIteratorRelease(kvstoreIterator *kvs_it);

4
src/lazyfree.c
View File

@ -192,8 +192,8 @@ void emptyDbAsync(serverDb *db) {
flags |= KVSTORE_FREE_EMPTY_DICTS;
}
kvstore *oldkeys = db->keys, *oldexpires = db->expires;
db->keys = kvstoreCreate(&dbDictType, slot_count_bits, flags);
db->expires = kvstoreCreate(&dbExpiresDictType, slot_count_bits, flags);
db->keys = kvstoreCreate(&kvstoreKeysDictType, slot_count_bits, flags);
db->expires = kvstoreCreate(&kvstoreExpiresDictType, slot_count_bits, flags);
atomic_fetch_add_explicit(&lazyfree_objects, kvstoreSize(oldkeys), memory_order_relaxed);
bioCreateLazyFreeJob(lazyfreeFreeDatabase, 2, oldkeys, oldexpires);
}

40
src/server.c
View File

@ -474,26 +474,32 @@ dictType zsetDictType = {
NULL, /* allow to expand */
};
/* Db->dict, keys are sds strings, vals are Objects. */
dictType dbDictType = {
/* Kvstore->keys, keys are sds strings, vals are Objects. */
dictType kvstoreKeysDictType = {
dictSdsHash, /* hash function */
NULL, /* key dup */
dictSdsKeyCompare, /* key compare */
NULL, /* key is embedded in the dictEntry and freed internally */
dictObjectDestructor, /* val destructor */
dictResizeAllowed, /* allow to resize */
kvstoreDictRehashingStarted,
kvstoreDictRehashingCompleted,
kvstoreDictMetadataSize,
.embedKey = dictSdsEmbedKey,
.embedded_entry = 1,
};
/* Db->expires */
dictType dbExpiresDictType = {
/* Kvstore->expires */
dictType kvstoreExpiresDictType = {
dictSdsHash, /* hash function */
NULL, /* key dup */
dictSdsKeyCompare, /* key compare */
NULL, /* key destructor */
NULL, /* val destructor */
dictResizeAllowed, /* allow to resize */
kvstoreDictRehashingStarted,
kvstoreDictRehashingCompleted,
kvstoreDictMetadataSize,
};
/* Command table. sds string -> command struct pointer. */
@ -540,7 +546,7 @@ dictType keylistDictType = {
};
/* KeyDict hash table type has unencoded Objects as keys and
* dicts as values. It's used for PUBSUB command to track clients subscribing the channels. */
* dicts as values. It's used for PUBSUB command to track clients subscribing the patterns. */
dictType objToDictDictType = {
dictObjHash, /* hash function */
NULL, /* key dup */
@ -550,6 +556,20 @@ dictType objToDictDictType = {
NULL /* allow to expand */
};
/* Same as objToDictDictType, added some kvstore callbacks, it's used
* for PUBSUB command to track clients subscribing the channels. */
dictType kvstoreChannelDictType = {
dictObjHash, /* hash function */
NULL, /* key dup */
dictObjKeyCompare, /* key compare */
dictObjectDestructor, /* key destructor */
dictDictDestructor, /* val destructor */
NULL, /* allow to expand */
kvstoreDictRehashingStarted,
kvstoreDictRehashingCompleted,
kvstoreDictMetadataSize,
};
/* Modules system dictionary type. Keys are module name,
* values are pointer to ValkeyModule struct. */
dictType modulesDictType = {
@ -2626,8 +2646,8 @@ void initServer(void) {
flags |= KVSTORE_FREE_EMPTY_DICTS;
}
for (j = 0; j < server.dbnum; j++) {
server.db[j].keys = kvstoreCreate(&dbDictType, slot_count_bits, flags);
server.db[j].expires = kvstoreCreate(&dbExpiresDictType, slot_count_bits, flags);
server.db[j].keys = kvstoreCreate(&kvstoreKeysDictType, slot_count_bits, flags);
server.db[j].expires = kvstoreCreate(&kvstoreExpiresDictType, slot_count_bits, flags);
server.db[j].expires_cursor = 0;
server.db[j].blocking_keys = dictCreate(&keylistDictType);
server.db[j].blocking_keys_unblock_on_nokey = dictCreate(&objectKeyPointerValueDictType);
@ -2642,10 +2662,10 @@ void initServer(void) {
/* Note that server.pubsub_channels was chosen to be a kvstore (with only one dict, which
* seems odd) just to make the code cleaner by making it be the same type as server.pubsubshard_channels
* (which has to be kvstore), see pubsubtype.serverPubSubChannels */
server.pubsub_channels = kvstoreCreate(&objToDictDictType, 0, KVSTORE_ALLOCATE_DICTS_ON_DEMAND);
server.pubsub_channels = kvstoreCreate(&kvstoreChannelDictType, 0, KVSTORE_ALLOCATE_DICTS_ON_DEMAND);
server.pubsub_patterns = dictCreate(&objToDictDictType);
server.pubsubshard_channels =
kvstoreCreate(&objToDictDictType, slot_count_bits, KVSTORE_ALLOCATE_DICTS_ON_DEMAND | KVSTORE_FREE_EMPTY_DICTS);
server.pubsubshard_channels = kvstoreCreate(&kvstoreChannelDictType, slot_count_bits,
KVSTORE_ALLOCATE_DICTS_ON_DEMAND | KVSTORE_FREE_EMPTY_DICTS);
server.pubsub_clients = 0;
server.watching_clients = 0;
server.cronloops = 0;

5
src/server.h
View File

@ -2637,7 +2637,8 @@ extern dictType objectKeyHeapPointerValueDictType;
extern dictType setDictType;
extern dictType BenchmarkDictType;
extern dictType zsetDictType;
extern dictType dbDictType;
extern dictType kvstoreKeysDictType;
extern dictType kvstoreExpiresDictType;
extern double R_Zero, R_PosInf, R_NegInf, R_Nan;
extern dictType hashDictType;
extern dictType stringSetDictType;
@ -2645,7 +2646,7 @@ extern dictType externalStringType;
extern dictType sdsHashDictType;
extern dictType clientDictType;
extern dictType objToDictDictType;
extern dictType dbExpiresDictType;
extern dictType kvstoreChannelDictType;
extern dictType modulesDictType;
extern dictType sdsReplyDictType;
extern dictType keylistDictType;