Redis/redis-For-Windows
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

[6.0] src/defrag.c

Browse Source
This commit is contained in:
tporadowski
2021-01-25 18:30:44 +01:00
parent 8390505c46
commit 7c238d573e
1 changed files with 117 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/defrag.c
+117 -69
View File
@@ -5,8 +5,8 @@
* We do that by scanning the keyspace and for each pointer we have, we can try to
* ask the allocator if moving it to a new address will help reduce fragmentation.
*
* Copyright (c) 2017, Oran Agra
* Copyright (c) 2017, Redis Labs, Inc
* Copyright (c) 2020, Oran Agra
* Copyright (c) 2020, Redis Labs, Inc
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -43,30 +43,23 @@
/* this method was added to jemalloc in order to help us understand which
* pointers are worthwhile moving and which aren't */
int je_get_defrag_hint(void* ptr, int *bin_util, int *run_util);
int je_get_defrag_hint(void* ptr);
/* forward declarations*/
void defragDictBucketCallback(void *privdata, dictEntry **bucketref);
dictEntry* replaceSateliteDictKeyPtrAndOrDefragDictEntry(dict *d, sds oldkey, sds newkey, uint64_t hash, PORT_LONG *defragged);
dictEntry* replaceSatelliteDictKeyPtrAndOrDefragDictEntry(dict *d, sds oldkey, sds newkey, uint64_t hash, PORT_LONG *defragged);
/* Defrag helper for generic allocations.
*
* returns NULL in case the allocatoin wasn't moved.
* returns NULL in case the allocation wasn't moved.
* when it returns a non-null value, the old pointer was already released
* and should NOT be accessed. */
void* activeDefragAlloc(void *ptr) {
int bin_util, run_util;
size_t size;
void *newptr;
if(!je_get_defrag_hint(ptr, &bin_util, &run_util)) {
server.stat_active_defrag_misses++;
return NULL;
}
/* if this run is more utilized than the average utilization in this bin
* (or it is full), skip it. This will eventually move all the allocations
* from relatively empty runs into relatively full runs. */
if (run_util > bin_util || run_util == 1<<16) {
if(!je_get_defrag_hint(ptr)) {
server.stat_active_defrag_misses++;
size = zmalloc_size(ptr);
return NULL;
}
/* move this allocation to a new allocation.
@@ -81,7 +74,7 @@ void* activeDefragAlloc(void *ptr) {
/*Defrag helper for sds strings
*
* returns NULL in case the allocatoin wasn't moved.
* returns NULL in case the allocation wasn't moved.
* when it returns a non-null value, the old pointer was already released
* and should NOT be accessed. */
sds activeDefragSds(sds sdsptr) {
@@ -97,7 +90,7 @@ sds activeDefragSds(sds sdsptr) {
/* Defrag helper for robj and/or string objects
*
* returns NULL in case the allocatoin wasn't moved.
* returns NULL in case the allocation wasn't moved.
* when it returns a non-null value, the old pointer was already released
* and should NOT be accessed. */
robj *activeDefragStringOb(robj* ob, PORT_LONG *defragged) {
@@ -137,11 +130,11 @@ robj *activeDefragStringOb(robj* ob, PORT_LONG *defragged) {
}
/* Defrag helper for dictEntries to be used during dict iteration (called on
* each step). Teturns a stat of how many pointers were moved. */
* each step). Returns a stat of how many pointers were moved. */
PORT_LONG dictIterDefragEntry(dictIterator *iter) {
/* This function is a little bit dirty since it messes with the internals
* of the dict and it's iterator, but the benefit is that it is very easy
* to use, and require no other chagnes in the dict. */
* to use, and require no other changes in the dict. */
PORT_LONG defragged = 0;
dictht *ht;
/* Handle the next entry (if there is one), and update the pointer in the
@@ -245,7 +238,7 @@ double *zslDefrag(zskiplist *zsl, double score, sds oldele, sds newele) {
return NULL;
}
/* Defrag helpler for sorted set.
/* Defrag helper for sorted set.
* Defrag a single dict entry key name, and corresponding skiplist struct */
PORT_LONG activeDefragZsetEntry(zset *zs, dictEntry *de) {
sds newsds;
@@ -356,7 +349,7 @@ PORT_LONG activeDefragSdsListAndDict(list *l, dict *d, int dict_val_type) {
if ((newsds = activeDefragSds(sdsele))) {
/* When defragging an sds value, we need to update the dict key */
uint64_t hash = dictGetHash(d, newsds);
replaceSateliteDictKeyPtrAndOrDefragDictEntry(d, sdsele, newsds, hash, &defragged);
replaceSatelliteDictKeyPtrAndOrDefragDictEntry(d, sdsele, newsds, hash, &defragged);
ln->value = newsds;
defragged++;
}
@@ -374,7 +367,7 @@ PORT_LONG activeDefragSdsListAndDict(list *l, dict *d, int dict_val_type) {
if ((newele = activeDefragStringOb(ele, &defragged)))
de->v.val = newele, defragged++;
} else if (dict_val_type == DEFRAG_SDS_DICT_VAL_VOID_PTR) {
void *newptr, *ptr = ln->value;
void *newptr, *ptr = dictGetVal(de);
if ((newptr = activeDefragAlloc(ptr)))
ln->value = newptr, defragged++;
}
@@ -392,7 +385,7 @@ PORT_LONG activeDefragSdsListAndDict(list *l, dict *d, int dict_val_type) {
* moved. Return value is the the dictEntry if found, or NULL if not found.
* NOTE: this is very ugly code, but it let's us avoid the complication of
* doing a scan on another dict. */
dictEntry* replaceSateliteDictKeyPtrAndOrDefragDictEntry(dict *d, sds oldkey, sds newkey, uint64_t hash, PORT_LONG *defragged) {
dictEntry* replaceSatelliteDictKeyPtrAndOrDefragDictEntry(dict *d, sds oldkey, sds newkey, uint64_t hash, PORT_LONG *defragged) {
dictEntry **deref = dictFindEntryRefByPtrAndHash(d, oldkey, hash);
if (deref) {
dictEntry *de = *deref;
@@ -408,44 +401,87 @@ dictEntry* replaceSateliteDictKeyPtrAndOrDefragDictEntry(dict *d, sds oldkey, sd
return NULL;
}
PORT_LONG activeDefragQuickListNodes(quicklist *ql) {
quicklistNode *node = ql->head, *newnode;
PORT_LONG activeDefragQuickListNode(quicklist *ql, quicklistNode **node_ref) {
quicklistNode *newnode, *node = *node_ref;
PORT_LONG defragged = 0;
unsigned char *newzl;
if ((newnode = activeDefragAlloc(node))) {
if (newnode->prev)
newnode->prev->next = newnode;
else
ql->head = newnode;
if (newnode->next)
newnode->next->prev = newnode;
else
ql->tail = newnode;
*node_ref = node = newnode;
defragged++;
}
if ((newzl = activeDefragAlloc(node->zl)))
defragged++, node->zl = newzl;
return defragged;
}
long activeDefragQuickListNodes(quicklist *ql) {
quicklistNode *node = ql->head;
PORT_LONG defragged = 0;
while (node) {
if ((newnode = activeDefragAlloc(node))) {
if (newnode->prev)
newnode->prev->next = newnode;
else
ql->head = newnode;
if (newnode->next)
newnode->next->prev = newnode;
else
ql->tail = newnode;
node = newnode;
defragged++;
}
if ((newzl = activeDefragAlloc(node->zl)))
defragged++, node->zl = newzl;
defragged += activeDefragQuickListNode(ql, &node);
node = node->next;
}
return defragged;
}
/* when the value has lots of elements, we want to handle it later and not as
* oart of the main dictionary scan. this is needed in order to prevent latency
* part of the main dictionary scan. this is needed in order to prevent latency
* spikes when handling large items */
void defragLater(redisDb *db, dictEntry *kde) {
sds key = sdsdup(dictGetKey(kde));
listAddNodeTail(db->defrag_later, key);
}
PORT_LONG scanLaterList(robj *ob) {
/* returns 0 if no more work needs to be been done, and 1 if time is up and more work is needed. */
PORT_LONG scanLaterList(robj *ob, PORT_ULONG *cursor, PORT_LONGLONG endtime, PORT_LONGLONG *defragged) {
quicklist *ql = ob->ptr;
quicklistNode *node;
PORT_LONG iterations = 0;
int bookmark_failed = 0;
if (ob->type != OBJ_LIST || ob->encoding != OBJ_ENCODING_QUICKLIST)
return 0;
server.stat_active_defrag_scanned+=ql->len;
return activeDefragQuickListNodes(ql);
if (*cursor == 0) {
/* if cursor is 0, we start new iteration */
node = ql->head;
} else {
node = quicklistBookmarkFind(ql, "_AD");
if (!node) {
/* if the bookmark was deleted, it means we reached the end. */
*cursor = 0;
return 0;
}
node = node->next;
}
(*cursor)++;
while (node) {
(*defragged) += activeDefragQuickListNode(ql, &node);
server.stat_active_defrag_scanned++;
if (++iterations > 128 && !bookmark_failed) {
if (ustime() > endtime) {
if (!quicklistBookmarkCreate(&ql, "_AD", node)) {
bookmark_failed = 1;
} else {
ob->ptr = ql; /* bookmark creation may have re-allocated the quicklist */
return 1;
}
}
iterations = 0;
}
node = node->next;
}
quicklistBookmarkDelete(ql, "_AD");
*cursor = 0;
return bookmark_failed? 1: 0;
}
typedef struct {
@@ -626,6 +662,7 @@ int scanLaterStraemListpacks(robj *ob, PORT_ULONG *cursor, PORT_LONGLONG endtime
/* if cursor is non-zero, we seek to the static 'last' */
if (!raxSeek(&ri,">", last, sizeof(last))) {
*cursor = 0;
raxStop(&ri);
return 0;
}
/* assign the iterator node callback after the seek, so that the
@@ -638,7 +675,8 @@ int scanLaterStraemListpacks(robj *ob, PORT_ULONG *cursor, PORT_LONGLONG endtime
void *newdata = activeDefragAlloc(ri.data);
if (newdata)
raxSetData(ri.node, ri.data=newdata), (*defragged)++;
if (++iterations > 16) {
server.stat_active_defrag_scanned++;
if (++iterations > 128) {
if (ustime() > endtime) {
serverAssert(ri.key_len==sizeof(last));
memcpy(last,ri.key,ri.key_len);
@@ -776,7 +814,7 @@ PORT_LONG defragKey(redisDb *db, dictEntry *de) {
* I can't search in db->expires for that key after i already released
* the pointer it holds it won't be able to do the string compare */
uint64_t hash = dictGetHash(db->dict, de->key);
replaceSateliteDictKeyPtrAndOrDefragDictEntry(db->expires, keysds, newsds, hash, &defragged);
replaceSatelliteDictKeyPtrAndOrDefragDictEntry(db->expires, keysds, newsds, hash, &defragged);
}
/* Try to defrag robj and / or string value. */
@@ -847,7 +885,7 @@ void defragScanCallback(void *privdata, const dictEntry *de) {
server.stat_active_defrag_scanned++;
}
/* Defrag scan callback for each hash table bicket,
/* Defrag scan callback for each hash table bucket,
* used in order to defrag the dictEntry allocations. */
void defragDictBucketCallback(void *privdata, dictEntry **bucketref) {
UNUSED(privdata); /* NOTE: this function is also used by both activeDefragCycle and scanLaterHash, etc. don't use privdata */
@@ -881,7 +919,7 @@ float getAllocatorFragmentation(size_t *out_frag_bytes) {
return frag_pct;
}
/* We may need to defrag other globals, one small allcation can hold a full allocator run.
/* We may need to defrag other globals, one small allocation can hold a full allocator run.
* so although small, it is still important to defrag these */
PORT_LONG defragOtherGlobals() {
PORT_LONG defragged = 0;
@@ -900,8 +938,7 @@ int defragLaterItem(dictEntry *de, PORT_ULONG *cursor, PORT_LONGLONG endtime) {
if (de) {
robj *ob = dictGetVal(de);
if (ob->type == OBJ_LIST) {
server.stat_active_defrag_hits += scanLaterList(ob);
*cursor = 0; /* list has no scan, we must finish it in one go */
return scanLaterList(ob, cursor, endtime, &server.stat_active_defrag_hits);
} else if (ob->type == OBJ_SET) {
server.stat_active_defrag_hits += scanLaterSet(ob, cursor);
} else if (ob->type == OBJ_ZSET) {
@@ -919,10 +956,12 @@ int defragLaterItem(dictEntry *de, PORT_ULONG *cursor, PORT_LONGLONG endtime) {
return 0;
}
/* static variables serving defragLaterStep to continue scanning a key from were we stopped last time. */
static sds defrag_later_current_key = NULL;
static PORT_ULONG defrag_later_cursor = 0;
/* returns 0 if no more work needs to be been done, and 1 if time is up and more work is needed. */
int defragLaterStep(redisDb *db, PORT_LONGLONG endtime) {
static sds current_key = NULL;
static PORT_ULONG cursor = 0;
unsigned int iterations = 0;
PORT_ULONGLONG prev_defragged = server.stat_active_defrag_hits;
PORT_ULONGLONG prev_scanned = server.stat_active_defrag_scanned;
@@ -930,16 +969,15 @@ int defragLaterStep(redisDb *db, PORT_LONGLONG endtime) {
do {
/* if we're not continuing a scan from the last call or loop, start a new one */
if (!cursor) {
if (!defrag_later_cursor) {
listNode *head = listFirst(db->defrag_later);
/* Move on to next key */
if (current_key) {
serverAssert(current_key == head->value);
sdsfree(head->value);
if (defrag_later_current_key) {
serverAssert(defrag_later_current_key == head->value);
listDelNode(db->defrag_later, head);
cursor = 0;
current_key = NULL;
defrag_later_cursor = 0;
defrag_later_current_key = NULL;
}
/* stop if we reached the last one. */
@@ -948,23 +986,18 @@ int defragLaterStep(redisDb *db, PORT_LONGLONG endtime) {
return 0;
/* start a new key */
current_key = head->value;
cursor = 0;
defrag_later_current_key = head->value;
defrag_later_cursor = 0;
}
/* each time we enter this function we need to fetch the key from the dict again (if it still exists) */
dictEntry *de = dictFind(db->dict, current_key);
dictEntry *de = dictFind(db->dict, defrag_later_current_key);
key_defragged = server.stat_active_defrag_hits;
do {
int quit = 0;
if (defragLaterItem(de, &cursor, endtime))
if (defragLaterItem(de, &defrag_later_cursor, endtime))
quit = 1; /* time is up, we didn't finish all the work */
/* Don't start a new BIG key in this loop, this is because the
* next key can be a list, and scanLaterList must be done in once cycle */
if (!cursor)
quit = 1;
/* Once in 16 scan iterations, 512 pointer reallocations, or 64 fields
* (if we have a lot of pointers in one hash bucket, or rehashing),
* check if we reached the time limit. */
@@ -982,7 +1015,7 @@ int defragLaterStep(redisDb *db, PORT_LONGLONG endtime) {
prev_defragged = server.stat_active_defrag_hits;
prev_scanned = server.stat_active_defrag_scanned;
}
} while(cursor);
} while(defrag_later_cursor);
if(key_defragged != server.stat_active_defrag_hits)
server.stat_active_defrag_key_hits++;
else
@@ -1039,10 +1072,25 @@ void activeDefragCycle(void) {
mstime_t latency;
int quit = 0;
if (server.aof_child_pid!=-1 || server.rdb_child_pid!=-1)
if (!server.active_defrag_enabled) {
if (server.active_defrag_running) {
/* if active defrag was disabled mid-run, start from fresh next time. */
server.active_defrag_running = 0;
if (db)
listEmpty(db->defrag_later);
defrag_later_current_key = NULL;
defrag_later_cursor = 0;
current_db = -1;
cursor = 0;
db = NULL;
}
return;
}
if (hasActiveChildProcess())
return; /* Defragging memory while there's a fork will just do damage. */
/* Once a second, check if we the fragmentation justfies starting a scan
/* Once a second, check if the fragmentation justfies starting a scan
* or making it more aggressive. */
run_with_period(1000) {
computeDefragCycles();
@@ -1112,7 +1160,7 @@ void activeDefragCycle(void) {
* (if we have a lot of pointers in one hash bucket or rehasing),
* check if we reached the time limit.
* But regardless, don't start a new db in this loop, this is because after
* the last db we call defragOtherGlobals, which must be done in once cycle */
* the last db we call defragOtherGlobals, which must be done in one cycle */
if (!cursor || (++iterations > 16 ||
server.stat_active_defrag_hits - prev_defragged > 512 ||
server.stat_active_defrag_scanned - prev_scanned > 64)) {