redis4.0 内存碎片整理源码解析

int main(int argc, char **argv) {
    /* ... */  
    initServer();
    /* ... */  
}

void initServer(void) {
    /* ... */  
    /* Create the timer callback, this is our way to process many background
     * operations incrementally, like clients timeout, eviction of unaccessed
     * expired keys and so forth. */  
    if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
        serverPanic("Can't create event loop timers.");
        exit(1);
    }
    /* ... */  
}

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    /* ... */  
    databasesCron();
    /* ... */  
}

void databasesCron(void) {
    /* ... */  
    /* Defrag keys gradually. */  
    if (server.active_defrag_enabled)
        activeDefragCycle();
    /* ... */  
}

long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
        aeTimeProc *proc, void *clientData,
        aeEventFinalizerProc *finalizerProc)
{
    long long id = eventLoop->timeEventNextId++;
    aeTimeEvent *te;

    te = zmalloc(sizeof(*te));
    if (te == NULL) return AE_ERR;
    te->id = id;
    aeAddMillisecondsToNow(milliseconds,&te->when_sec,&te->when_ms);
    te->timeProc = proc;
    te->finalizerProc = finalizerProc;
    te->clientData = clientData;
    te->prev = NULL;
    te->next = eventLoop->timeEventHead;
    if (te->next)
        te->next->prev = te;
    eventLoop->timeEventHead = te;
    return id;
}

#define INTERPOLATE(x, x1, x2, y1, y2) ( (y1) + ((x)-(x1)) * ((y2)-(y1)) / ((x2)-(x1)) )
#define LIMIT(y, min, max) ((y)<(min)? min: ((y)>(max)? max: (y)))

void activeDefragCycle(void) {
    /* ... */  

    /* Once a second, check if we the fragmentation justfies starting a scan
     * or making it more aggressive. */  
    run_with_period(1000) {
        size_t frag_bytes;
        float frag_pct = getAllocatorFragmentation(&frag_bytes);
        /* If we're not already running, and below the threshold, exit. */  
        if (!server.active_defrag_running) {
            if(frag_pct < server.active_defrag_threshold_lower || frag_bytes < server.active_defrag_ignore_bytes)
                return;
        }

        /* 计算内存碎片整理所需要占用的主线程资源 */  
        /* Calculate the adaptive aggressiveness of the defrag */  
        int cpu_pct = INTERPOLATE(frag_pct,
                server.active_defrag_threshold_lower,
                server.active_defrag_threshold_upper,
                server.active_defrag_cycle_min,
                server.active_defrag_cycle_max);
        cpu_pct = LIMIT(cpu_pct,
                server.active_defrag_cycle_min,
                server.active_defrag_cycle_max);
        /* 限制占用资源范围 */  
         /* We allow increasing the aggressiveness during a scan, but don't
          * reduce it. */  
        if (!server.active_defrag_running ||
            cpu_pct > server.active_defrag_running)
        {
            server.active_defrag_running = cpu_pct;
            serverLog(LL_VERBOSE,
                "Starting active defrag, frag=%.0f%%, frag_bytes=%zu, cpu=%d%%",
                frag_pct, frag_bytes, cpu_pct);
        }
    }
    if (!server.active_defrag_running)
        return;

    timelimit = 1000000*server.active_defrag_running/server.hz/100;
    if (timelimit <= 0) timelimit = 1;

    /* ... */  
}

void activeDefragCycle(void) {
    /* ... */  

    start = ustime();
    do {
        cursor = dictScan(db->dict, cursor, defragScanCallback, defragDictBucketCallback, db);
        /* Once in 16 scan iterations, or 1000 pointer reallocations
         * (if we have a lot of pointers in one hash bucket), check if we
         * reached the tiem limit. */  
        if (cursor && (++iterations > 16 || server.stat_active_defrag_hits - defragged > 1000)) {
            if ((ustime() - start) > timelimit) {
                return;
            }
            iterations = 0;
            defragged = server.stat_active_defrag_hits;
        }
    } while(cursor);

    /* ... */  
}

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) {
        server.stat_active_defrag_misses++;
        return NULL;
    }
    /* move this allocation to a new allocation.
     * make sure not to use the thread cache. so that we don't get back the same
     * pointers we try to free */  
    size = zmalloc_size(ptr);
    newptr = zmalloc_no_tcache(size);
    memcpy(newptr, ptr, size);
    zfree_no_tcache(ptr);
    return newptr;
}

} else if (!strcasecmp(c->argv[1]->ptr,"purge") && c->argc == 2) {
#if defined(USE_JEMALLOC)
    char tmp[32];
    unsigned narenas = 0;
    size_t sz = sizeof(unsigned);
    if (!je_mallctl("arenas.narenas", &narenas, &sz, NULL, 0)) {
        sprintf(tmp, "arena.%d.purge", narenas);
        if (!je_mallctl(tmp, NULL, 0, NULL, 0)) {
            addReply(c, shared.ok);
            return;
        }
    }
    addReplyError(c, "Error purging dirty pages");
#else  
    addReply(c, shared.ok);
    /* Nothing to do for other allocators. */  
#endif  
}