diff --git a/src/defrag.c b/src/defrag.c index 0adf4f77..28cc8015 100644 --- a/src/defrag.c +++ b/src/defrag.c @@ -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)) {