log&& buffevent&&内存池 4

说起内存问题 就想起 buffer 和cache

• 其核心是：buffer和cache对读和写都会混存，只是对象不同，前者是针对块设备，后者是针对文件。

再就是大家都想 重写一下内存池 最后发现 自己写的是一坨屎

作者：韦易笑
链接：https://www.zhihu.com/question/25527491/answer/56571062
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

1. 实现教科书上的内存分配器：
做一个链表指向空闲内存，分配就是取出一块来，改写链表，返回，释放就是放回到链表里面，并做好归并。注意做好标记和保护，避免二次释放，还可以花点力气在如何查找最适合大小的内存快的搜索上，减少内存碎片，有空你了还可以把链表换成伙伴算法，写着玩嘛。

2. 实现固定内存分配器：
即实现一个 FreeList，每个 FreeList 用于分配固定大小的内存块，比如用于分配 32字节对象的固定内存分配器，之类的。每个固定内存分配器里面有两个链表，OpenList 用于存储未分配的空闲对象，CloseList用于存储已分配的内存对象，那么所谓的分配就是从 OpenList 中取出一个对象放到 CloseList 里并且返回给用户，释放又是从 CloseList 移回到 OpenList。分配时如果不够，那么就需要增长 OpenList：申请一个大一点的内存块，切割成比如 64 个相同大小的对象添加到 OpenList中。这个固定内存分配器回收的时候，统一把先前向系统申请的内存块全部还给系统。

3. 实现 FreeList 池：
在你实现了 FreeList的基础上，按照不同对象大小（8字节，16字节，32，64，128，256，512，1K。。。64K），构造十多个固定内存分配器，分配内存时根据内存大小查表，决定到底由哪个分配器负责，分配后要在头部的 header 处（ptr[-sizeof(char*)]处）写上 cookie，表示又哪个分配器分配的，这样释放时候你才能正确归还。如果大于64K，则直接用系统的 malloc作为分配，如此以浪费内存为代价你得到了一个分配时间近似O（1）的内存分配器，差不多实现了一个 memcached 的 slab 内存管理器了，但是先别得意。此 slab 非彼 slab（sunos/solaris/linux kernel 的 slab）。这说白了还是一个弱智的 freelist 无法归还内存给操作系统，某个 FreeList 如果高峰期占用了大量内存即使后面不用，也无法支援到其他内存不够的 FreeList，所以我们做的这个和 memcached 类似的分配器其实是比较残缺的，你还需要往下继续优化。

4. 实现正统的 slab （非memcached的伪 slab）代替 FreeList：
这时候你需要阅读一下 http://citeseer.ist.psu.edu/bonwick94slab.html 这篇论文了，现代内存分配技术的基础，如何管理 slab 上的对象，如何进行地址管理，如何管理不同 slab 的生命周期，如何将内存回收给系统。然后开始实现一个类似的东西，文章上传统的 slab 的各种基础概念虽然今天没有改变，但是所用到的数据结构和控制方法其实已经有很多更好的方法了，你可以边实现边思考下，实在不行还可以参考 kernel 源码嘛。但是有很多事情应用程序做不了，有很多实现你是不能照搬的，比如页面提供器，可以提供连续线性地址的页面，再比如说 kernel 本身记录着每个页面对应的 slab，你查找 slab 时，系统其实是根据线性地址移位得到页面编号，然后查表得到的，而你应用程序不可能这么干，你还得做一些额外的体系来解决这些问题，还需要写一些额外的 cookie 来做标记。做好内存收缩工作，内存不够时先收缩所有分配器的 slab，再尝试重新分配。再做好内存回收工作，多余的内存，一段时间不使用可以还给操作系统。

5. 实现混合分配策略：
你实现了上面很多常见的算法后，该具体阅读各种内存分配器的代码了，这些都是经过实践检验的，比如 libc 的内存分配器，或者参考有自带内存管理的各种开源项目，比如 python 源码，做点实验对比他们的优劣，然后根据分配对象的大小采用不同的分配策略，区别对待各种情况。试验的差不多了就得引入多线程支持了，将你的锁改小。注意很多系统层的线程安全策略你是没法弄的，比如操作系统可以关中断，短时间内禁止本cpu发生任务切换，这点应用程序就很麻烦了，还得用更小的锁来代替。当锁已经小到不能再小，也可以选择引入 STM 来代替各种链表的锁。

6. 实现 Per-CPU Cache：
现代内存分配器，在多核下的一个重要优化就是给多核增加 cache，为了进一步避免多线程锁竞争，需要引入 Per-CPU Cache 了。分配内存先找到对应线程所在的cpu，从该cpu上对应的 cache 里分配，cache 不够了就一次性从你底层的内存分配器里多分配几个对象进来填充 cache，释放时也是先放回 cache，cache里面如果对象太多，就做一次收缩，把内存换个底层分配器，让其他 cpu 的cache有机会利用。这样针对很多短生命周期的频繁的分配、释放，其实都是在 cache 里完成的，没有锁竞争，同时cache分配逻辑简单，速度更快。操作系统里面的代码经常是直接读取当前的cpu是哪个，而应用层实现你可以用 thread local storage 来代替，目前这些东西在 crt的 malloc 里还暂时支持不到位（不排除未来版本会增加），可以更多参考 tc/jemalloc。

7. 实现地址着色:
现代内存分配器必须多考虑总线压力，在很多机型上，如果内存访问集中在某条 cache line相同的偏移上，会给总线带来额外的负担和压力。比如你经常要分配一个 FILE 对象，而每个 FILE对象使用时会比较集中的访问 int FILE::flag; 这个成员变量，如果你的页面提供器提供的页面地址是按照 4K对齐的，那么很可能多个 FILE对象的 flag 成员所处的 cache line 偏移地址是相同的，大量访问这些相同的偏移地址会给总线带来很大负担，这时候你需要给每个对象额外增加一些偏移，让他们能够均匀的分布在线性地址对应的cache line 偏移上，消减总线冲突的开销。

8. 优化缓存竞争：
多核时代，很多单核时代的代码都需要针对性的优化改写，最基本的一条就是 cache 竞争，这是比前面锁竞争更恶劣的情况：如果两个cpu同时访问相同的 cache-line 或者物理页面，那么 cpu 之间为了保证内存一致性会做很多的通信工作，比如那个cpu0需要用到这段内存，发现cpu1也在用，那么需要通知cpu1，将cpu1 L1-L2缓存里面的数据写回该物理内存，并且释放控制权，这时cpu0取得了控制权才能继续操作，期间cpu0-cpu1之间的通信协议是比较复杂的，代价也是比较大的，cache竞争比锁竞争恶劣不少。为了避免 cache 竞争，需要比先前Per-CPU cache 更彻底的 Per-CPU Page 机制来解决，直接让不同的cpu使用不同的页面进行二次分配，彻底避免 cache 竞争。具体应用层的做法也是利用线性地址来判断所属页面（因为物理页面映射到进程地址也是4k对齐的），同时继续使用 thread local storage 或者用系统提供的 api 读取当前属于哪个 cpu 来实现。为了避免核太多每个核占据大量的页面带来的不必要的浪费，你可以参考下 Linux 最新的 slub 内存分配算法，但是 slub 也有未尽之处，好几个 linux 发行版在实践中发现 slub 还是存在一些问题的（非bug，而是机制），所以大部分发行版默认都是关闭 slub 的，虽然，你还是可以借鉴测试一下。

9. 调试和折腾：
继续参考各种现代内存分配器，取长补短，然后给你的分配器添加一些便于调试的机制，方便诊断各种问题。在你借鉴了很多开源项目，自己也做了一些所谓的优化，折腾了那么久以后，你或许以为你的分配器可以同各种开源分配器一战了，测试效果好像也挺好的，先别急，继续观察内存利用率，向操作系统申请/归还内存的频率等一系列容易被人忽视的指标是否相同。同时更换你的测试用例，看看更多的情况下，是否结果还和先前一样？这些都差不多的时候，你发现没有个一两年的大规模持续使用，你很难发现一些潜在的隐患和bug，可能你觉得没问题的代码，跑了两年后都会继续报bug，这很正常，多点耐心，兴许第三年以后就比较稳定了呢？

有卯用呢？

十多年前 libc 还不成熟的情况下，为了程序长时间运行的稳定性，大部分程序员都必须针对自己的应用来实现针对特定情况的内存分配器。当年如果不自己管理内存，很多客户端，如果计算密集频繁分配，才开始可能没什么区别，但跑个几个小时性能立马就下降下来了；服务器进程持续运行个10多天不重启，速度也会越来越慢，碎片多了嘛。如今 libc 的 malloc 也进步了很多，这样的情况比较少了，那你再做一个内存池的意义何在呢？

在你的玩具比较稳定的情况下，终于可以产生一些价值了，因为一些性能指标你无法兼得，标准的分配器往往提供了一个类似保守和中庸的做法，来针对大部分的情况，你可以做的第一步，就是打破这样的平衡，让你的分配器倾向于某些情况比如：

1. 现代计算机内存都很大，你是不是可以牺牲内存利用率为代价换取更高的内存归还/重用的效率？同时换取更快的分配速度？或许你会发现，你可以比 libc 的 malloc 平均浪费 30%内存的代价换来两倍以上的性能提升，在一些内存分配成为瓶颈的应用中起到积极的作用。

2. 比如你可以调整大小内存的比值，libc如果认为 8K以下是小内存，那么你可以不那么认为。

3. 比如如果你的系统就是一个单线程的东西，那么你是否能提供开关，完全以单线程的模式进行运作，完全绕过各种锁和针对多核进行的各种冗余操作呢？

4. 比如你的机器内存有限，你应用需要耗费大量的内存，那么你可以引入其他机制，以牺牲少量性能为代价，换取更好的内存回收效果和内存利用率。

5. 最近分配的对象尽量在线性地址上集中在一起，这样缓存命中高，也不易发生缺页。

6. 比如你程序里面某些对象需要被跟踪，你能否直接在分配器上实现对象跟踪机制，跟踪各种泄漏，越界问题？

7. 每个内存分配都在寻求最佳的公平，你在乎的公平是什么？

。。。。。。

写的挺经典的

来看下 memcache

 

初始化slabs_init

 //参数factor是扩容因子，默认值是1.25
void slabs_init(const size_t limit, const double factor, const bool prealloc) {
    int i = POWER_SMALLEST - 1;

    //settings.chunk_size默认值为48，可以在启动memcached的时候通过-n选项设置
    //size由两部分组成: item结构体本身和这个item对应的数据
    //这里的数据也就是set、add命令中的那个数据，后面的循环可以看到这个size变量
    //会根据扩容因子factor慢慢扩大，所以能存储的数据长度也会变大的

    unsigned int size = sizeof(item) + settings.chunk_size;

    //用户设置或默认的内存大小限制
    mem_limit = limit;

    //用户要求预分配一块的内存，以后需要内存，就向这块内存申请
    if (prealloc) { //默认false
        /* Allocate everything in a big chunk with malloc */
        mem_base = malloc(mem_limit);
        if (mem_base != NULL) {
            mem_current = mem_base;
            mem_avail = mem_limit;
        } else {
            fprintf(stderr, "Warning: Failed to allocate requested memory in"
                    " one large chunk.
Will allocate in smaller chunks
");
        }
    }

    //初始化数组，这个操作很重要，数组中所有元素的成员变量都为0了
    memset(slabclass, 0, sizeof(slabclass));

    //slabclass数组中的第一个元素并不使用
    //settings.item_size_max是memecached支持的最大item尺寸，默认为1M
    //也就是网上所说的memcahced存储的数据最大为1MB
    while (++i < POWER_LARGEST && size <= settings.item_size_max / factor) {
        /* Make sure items are always n-byte aligned */
        if (size % CHUNK_ALIGN_BYTES) //8字节对齐
            size += CHUNK_ALIGN_BYTES - (size % CHUNK_ALIGN_BYTES);

        //这个slabclass的slab分配器能分配的item的大小
        slabclass[i].size = size;
        //这个slabclass的slab分配器最多能分配多少个item(也决定了最多分配多少内存)
        slabclass[i].perslab = settings.item_size_max / slabclass[i].size;
        //扩容
        size *= factor;
        if (settings.verbose > 1) {
            fprintf(stderr, "slab class %3d: chunk size %9u perslab %7u
",
                    i, slabclass[i].size, slabclass[i].perslab);
        }
    }
    
//settings.item_size_max = 1024 * 1024=1M; /* The famous 1MB upper limit. */
//settings.item_size_max / factor    1048576/1.25=838860.8 这就是单page最大的chunk大小 字节
//slabclass[41]    {size=717184 perslab=1 slots=0x00000000 ...}    slabclass_t
//所以到了42就跳出循环了
//slabclass[42]    {size=1048576 perslab=1 slots=0x00000000 ...}    slabclass_t
//43就不分配了
//slabclass[43]    {size=0 perslab=0 slots=0x00000000 ...}    slabclass_t


    //最大的item
    power_largest = i;
    slabclass[power_largest].size = settings.item_size_max;
    slabclass[power_largest].perslab = 1;
    if (settings.verbose > 1) {
        fprintf(stderr, "slab class %3d: chunk size %9u perslab %7u
",
                i, slabclass[i].size, slabclass[i].perslab);
    }

    /* for the test suite:  faking of how much we've already malloc'd */
    {
        char *t_initial_malloc = getenv("T_MEMD_INITIAL_MALLOC");
        if (t_initial_malloc) {
            mem_malloced = (size_t)atol(t_initial_malloc);
        }

    }
    //预先分配内存
    if (prealloc) {
        slabs_preallocate(power_largest);
    }
}

第一次分配slab

static void process_command(conn *c, char *command) {

    token_t tokens[MAX_TOKENS];
    size_t ntokens;
    int comm;

    assert(c != NULL);

    MEMCACHED_PROCESS_COMMAND_START(c->sfd, c->rcurr, c->rbytes);

    if (settings.verbose > 1)
        fprintf(stderr, "<%d %s
", c->sfd, command);

    /*
     * for commands set/add/replace, we build an item and read the data
     * directly into it, then continue in nread_complete().
     */

    c->msgcurr = 0;
    c->msgused = 0;
    c->iovused = 0;
    if (add_msghdr(c) != 0) {
        out_of_memory(c, "SERVER_ERROR out of memory preparing response");
        return;
    }

    //将一条命令分割成一个个的token，并用tokens数组一一对应的指向
    //比如命令"set tt 3 0 10"，将被分割成"set"、"tt"、"3"、"0"、"10"
    //并用tokens数组的5个元素对应指向。token_t类型的value成员指向对应token
    //在command字符串中的位置，length则指明该token的长度
    //该函数返回token的数量，length则指明该token的长度
    //上面的set命令例子，tokensize_command会返回6。最后一个token是无意义的
    ntokens = tokenize_command(command, tokens, MAX_TOKENS);//将命令记号化

    //对于命令"get tk"，那么token[0].value等于指向"get"的开始位置
    //tokens[1].value则指向"tk"的开始位置
    ----------------------

        process_update_command(c, tokens, ntokens, comm, false);
        
        
}

//process_update_command读取第一行，complete_nread_ascii读取完数据后处理
static void process_update_command(conn *c, token_t *tokens, const size_t ntokens, int comm, bool handle_cas) {
    char *key; //键值
    size_t nkey; //键值长度
    unsigned int flags; //item的flags
    int32_t exptime_int = 0; 
    time_t exptime;//item的超时
    int vlen;
    uint64_t req_cas_id=0;
    item *it;

    assert(c != NULL);

    //服务器不需要回复信息给客户端，这可以减少网络IO进而提高速度
    //这种设置是一次性的，不影响下一条命令
    set_noreply_maybe(c, tokens, ntokens); //处理用户命令里面的noreply

    //键值的长度太长了。KEY_MAX_LENGTH为250
    if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) {
        out_string(c, "CLIENT_ERROR bad command line format");
        return;
    }

    int i;

    for(i = 0; i < ntokens; i++) 
        printf("yang test : <value:%s>
", tokens[i].value);

    key = tokens[KEY_TOKEN].value;
    nkey = tokens[KEY_TOKEN].length;

    //将字符串转成unsigned long，获取false、exptime_int、vlen。
    //它们的字符串形式必须是纯数字，否则转换失败，返回false
    if (! (safe_strtoul(tokens[2].value, (uint32_t *)&flags)
           && safe_strtol(tokens[3].value, &exptime_int)
           && safe_strtol(tokens[4].value, (int32_t *)&vlen))) {
        out_string(c, "CLIENT_ERROR bad command line format");
        return;
    }

    /* Ubuntu 8.04 breaks when I pass exptime to safe_strtol */
    exptime = exptime_int;

    /* Negative exptimes can underflow and end up immortal. realtime() will
       immediately expire values that are greater than REALTIME_MAXDELTA, but less
       than process_started, so lets aim for that. */
    if (exptime < 0) //此时会立即过期失效
        exptime = REALTIME_MAXDELTA + 1; //REALTIME_MAXDELTA等于30天

    // does cas value exist?
    if (handle_cas) { //只有cas命令这里才会满足条件

        /*
        set yang 1 1 3 2
        abc
        STORED
        第一行末尾的2是实际需要的
          */
        if (!safe_strtoull(tokens[5].value, &req_cas_id)) {
            out_string(c, "CLIENT_ERROR bad command line format");
            return;
        }
    }
    //在存储item数据的时候，都会自动在数据的最后加上"
"
    vlen += 2; //+2是因为data后面还要加上"
"这两个字符
    if (vlen < 0 || vlen - 2 < 0) {
        out_string(c, "CLIENT_ERROR bad command line format");
        return;
    }

    //根据所需的大小分配对应的item，并给这个item赋值
    // 除了time和refcount成员外，其他的都赋值了。并把键值、flag这些值都拷贝
    //到item后面的buff里面了，至于data，因为现在都还没拿到所以还没赋值
    //realtime(exptime)是直接赋值给 item的exptime成员
    it = item_alloc(key, nkey, flags, realtime(exptime), vlen);

    if (settings.detail_enabled) {
        stats_prefix_record_set(key, nkey);
    }
    
    if (it == 0) {//没内存了，获取item失败
        if (! item_size_ok(nkey, flags, vlen))
            out_string(c, "SERVER_ERROR object too large for cache");
        else
            out_of_memory(c, "SERVER_ERROR out of memory storing object");
        /* swallow the data line */
        c->write_and_go = conn_swallow;
        c->sbytes = vlen;

        /* Avoid stale data persisting in cache because we failed alloc.
         * Unacceptable for SET. Anywhere else too? */
        if (comm == NREAD_SET) { //这次从小对key进行set，但是却没有成功，则需要删除primary_hashtable中的该key
            it = item_get(key, nkey);
            if (it) {
                item_unlink(it);
                item_remove(it);
            }
        }

        return;
    }

    //set cas等命令行中的expire保存到it->expire  cas保存在it->data->case中的
    ITEM_set_cas(it, req_cas_id); //填充cas部分

    //本函数并不会把item插入到哈希表和LRU队列，这个插入工作由
    //complete_nread_ascii函数完成  当从客户端读取玩数据部分后再complete_nread中把item添加到hash和LRU队列中
    c->item = it;
    c->ritem = ITEM_data(it);//数据直通车 
    c->rlbytes = it->nbytes; //等于vlen(要比用户输入的长度大2，因为要加上
)
    c->cmd = comm;  
    conn_set_state(c, conn_nread); //继续去read数据部分+

}

/*
 * Allocates a new item.
 */
item *item_alloc(char *key, size_t nkey, int flags, rel_time_t exptime, int nbytes) {
    item *it;
    /* do_item_alloc handles its own locks */
    it = do_item_alloc(key, nkey, flags, exptime, nbytes, 0);
    return it;
}

客户端申请存储key value会调用到do_item_alloc

//key、flags、exptime三个参数是用户在使用set、add命令存储一条数据时输入的参数
//nkey是key字符串的长度。nbytes则是用户要存储的data长度+2，因为在data结尾处还要加上"
",data的
在该函数前+2的
//cur_hv则是根据键值key计算得到的哈希值
item *do_item_alloc(char *key, const size_t nkey, const int flags,
                    const rel_time_t exptime, const int nbytes,
                    const uint32_t cur_hv) { //do_item_alloc (分配item) 和 item_free （释放item）
    uint8_t nsuffix;
    item *it = NULL;
    char suffix[40];

    //要存储这个item需要的总空间  nkey+1是因为set key 0 0 3
  abc
  对应的key行后面有

    size_t ntotal = item_make_header(nkey + 1, flags, nbytes, suffix, &nsuffix);
    if (settings.use_cas) {
        ntotal += sizeof(uint64_t);
    }

    //根据大小判断从属于哪个slab
    unsigned int id = slabs_clsid(ntotal);
    if (id == 0) //0表示不属于任何一个slab  slabclass是从1开始的
        return 0;

    mutex_lock(&cache_lock);
    /* do a quick check if we have any expired items in the tail.. */
    int tries = 5;
    /* Avoid hangs if a slab has nothing but refcounted stuff in it. */
    int tries_lrutail_reflocked = 1000;
    int tried_alloc = 0;
    item *search;
    item *next_it;
    void *hold_lock = NULL;
    rel_time_t oldest_live = settings.oldest_live;

    search = tails[id];
    /* We walk up *only* for locked items. Never searching for expired.
     * Waste of CPU for almost all deployments */

    //第一次看这个for循环，直接认为search等于NULL，直接看for循环后面的代码
    //这个循环里面会在对应LRU队列中查找过期失效的item，最多尝试tries个item。
    //从LRU的队尾开始尝试。如果item被其他worker线程引用了，那么就尝试下一个
    //如果没有的被其他worker线程所引用，那么就测试该item是否过期失效
    //如果过期失效了，那么就可以使用这个item(最终会返回这个item)。如果没有
    //过期失效，那么不再尝试其他item了(因为是从LRU队列的队尾开始尝试的,对尾的都没有失效，其他前面的肯定不会失效)，
    //直接调用slabs_alloc申请一个新的内存存储item。如果申请新内存都失败，
    //那么在允许LRU淘汰的情况下就会启动踢人机制
    
    for (; tries > 0 && search != NULL; tries--, search=next_it) {
        /* we might relink search mid-loop, so search->prev isn't reliable */
        next_it = search->prev;
        if (search->nbytes == 0 && search->nkey == 0 && search->it_flags == 1) {
            /* We are a crawler, ignore it. */
            //这是一个爬虫item，直接跳过
            tries++; //爬虫item不计入尝试的item数中
            continue;
        }
        uint32_t hv = hash(ITEM_key(search), search->nkey);
        /* Attempt to hash item lock the "search" item. If locked, no
         * other callers can incr the refcount
         */
        /* Don't accidentally grab ourselves, or bail if we can't quicklock */
        //尝试抢占锁，抢不了就走人，不等待锁。
        if (hv == cur_hv || (hold_lock = item_trylock(hv)) == NULL)
            continue;
        /* Now see if the item is refcount locked */
        if (refcount_incr(&search->refcount) != 2) { //引用计数>=3  //引用数，还有其他线程在引用，不能霸占这个item  
            /* Avoid pathological case with ref'ed items in tail */
            //刷新这个item的访问时间以及在LRU队列中的位置
            do_item_update_nolock(search);
            tries_lrutail_reflocked--;
            tries++;
            refcount_decr(&search->refcount);

            //此时引用数>=2
            itemstats[id].lrutail_reflocked++;
            /* Old rare bug could cause a refcount leak. We haven't seen
             * it in years, but we leave this code in to prevent failures
             * just in case */
            if (settings.tail_repair_time &&
                    search->time + settings.tail_repair_time < current_time) {
                itemstats[id].tailrepairs++;
                search->refcount = 1;
                do_item_unlink_nolock(search, hv);
            }
            if (hold_lock)
                item_trylock_unlock(hold_lock);

            if (tries_lrutail_reflocked < 1)
                break;

            continue;
        }


//search指向的item的refcount等于2，这说明此时这个item除了本worker线程外，没有其他任何worker线程索引其。可以放心释放并重用这个item  
          
//因为这个循环是从lru链表的后面开始遍历的。所以一开始search就指向了最不常用的item，如果这个item都没有过期。那么其他的比其更常用  
//的item就不要删除了(即使它们过期了)。此时只能向slabs申请内存   
        /* Expired or flushed */
        if ((search->exptime != 0 && search->exptime < current_time)
            || (search->time <= oldest_live && oldest_live <= current_time)) { 
            ////search指向的item是一个过期失效的item，可以使用之  
            itemstats[id].reclaimed++;
            if ((search->it_flags & ITEM_FETCHED) == 0) {
                itemstats[id].expired_unfetched++;
            }
            it = search;
            //重新计算一下这个slabclass_t分配出去的内存大小直接霸占旧的item就需要重新计算  
            slabs_adjust_mem_requested(it->slabs_clsid, ITEM_ntotal(it), ntotal); 
            do_item_unlink_nolock(it, hv);  
            /* Initialize the item block: */
            it->slabs_clsid = 0;
        } else if ((it = slabs_alloc(ntotal, id)) == NULL) { //该id对于的slab中没有过期的item，则从slab重新获取一个item
            tried_alloc = 1;
            if (settings.evict_to_free == 0) { //设置了不进行LRU淘汰item  
                itemstats[id].outofmemory++; //此时只能向客户端回复错误了  
            } else {
                //此刻，过期失效的item没有找到，申请内存又失败了。看来只能使用  
                //LRU淘汰一个item(即使这个item并没有过期失效)  
                itemstats[id].evicted++;//增加被踢的item数  
                itemstats[id].evicted_time = current_time - search->time;
                if (search->exptime != 0) //即使一个item的exptime成员设置为永不超时(0)，还是会被踢的  
                    itemstats[id].evicted_nonzero++;
                if ((search->it_flags & ITEM_FETCHED) == 0) { 
                    itemstats[id].evicted_unfetched++;
                }
                it = search;
                slabs_adjust_mem_requested(it->slabs_clsid, ITEM_ntotal(it), ntotal);
                do_item_unlink_nolock(it, hv);
                /* Initialize the item block: */
                it->slabs_clsid = 0;

                /* If we've just evicted an item, and the automover is set to
                 * angry bird mode, attempt to rip memory into this slab class.
                 * TODO: Move valid object detection into a function, and on a
                 * "successful" memory pull, look behind and see if the next alloc
                 * would be an eviction. Then kick off the slab mover before the
                 * eviction happens.
                 */
                //一旦发现有item被踢，那么就启动内存页重分配操作  
                //这个太频繁了，不推荐   
                if (settings.slab_automove == 2)
                    slabs_reassign(-1, id);
            }
        }

        //引用计数减一。此时该item已经没有任何worker线程索引其，并且哈希表也  
        //不再索引其  
        refcount_decr(&search->refcount);
        /* If hash values were equal, we don't grab a second lock */
        if (hold_lock)
            item_trylock_unlock(hold_lock);
        break;
    }

    if (!tried_alloc && (tries == 0 || search == NULL))
        it = slabs_alloc(ntotal, id);

    if (it == NULL) {
        itemstats[id].outofmemory++;
        mutex_unlock(&cache_lock);
        return NULL;
    }

    assert(it->slabs_clsid == 0);
    assert(it != heads[id]);

    /* Item initialization can happen outside of the lock; the item's already
     * been removed from the slab LRU.
     */
    it->refcount = 1;     /* the caller will have a reference */  //新开盘的默认初值为1
    mutex_unlock(&cache_lock);
    it->next = it->prev = it->h_next = 0;
    it->slabs_clsid = id;

    DEBUG_REFCNT(it, '*');
    it->it_flags = settings.use_cas ? ITEM_CAS : 0;
    it->nkey = nkey;
    it->nbytes = nbytes;  //set cas等命令行中的expire保存到it->expire  cas保存在it->data->case中的
    memcpy(ITEM_key(it), key, nkey);
    it->exptime = exptime;
    memcpy(ITEM_suffix(it), suffix, (size_t)nsuffix);
    it->nsuffix = nsuffix; //实际分配的item空间大小为it->nkey +1 + it->nsuffix +it->nbytes
    return it;
}

主要逻辑有：

调用slabs_clsid() -->寻找适合给定大小的item存储的slab

/*
 * Figures out which slab class (chunk size) is required to store an item of
 * a given size.
 *
 * Given object size, return id to use when allocating/freeing memory for object
 * 0 means error: can't store such a large object
 */

unsigned int slabs_clsid(const size_t size) {
    int res = POWER_SMALLEST; //res的初始值为1

    //返回0表示查找失败，因为slabclass数组中，第一个元素是没有使用的
    if (size == 0)
        return 0;
    //因为slabclass数组中各个元素能分配的item大小是升序的
    //所以从小到大直接判断可在数组找到最小但又能满足的元素
    while (size > slabclass[res].size)
        if (res++ == power_largest)     /* won't fit in the biggest slab */
            return 0;
    return res;
}

调用do_slabs_alloc返回slots指向的item，并使slots指向下一个item

//从id对应的slabclass[id]中的一个chunk中获取所需的size空间
void *slabs_alloc(size_t size, unsigned int id) {
    void *ret;

    pthread_mutex_lock(&slabs_lock);
    ret = do_slabs_alloc(size, id);
    pthread_mutex_unlock(&slabs_lock);
    return ret;
}
//向slabclass申请一个item。在调用函数之前，已经调用slabs_clsid函数确定
//本次申请是向哪个slabclass_t申请item了，参数id就是指明是向哪个slabclass_t
//申请item。如果该slabclass_t是有空闲item，那么就从空闲的item队列分配一个
//如果没有空闲item，那么就申请一个内存页。再从新申请的页中分配一个item
// 返回值为得到的item，如果没有内存了，返回NULL
static void *do_slabs_alloc(const size_t size, unsigned int id) {
    slabclass_t *p;
    void *ret = NULL;
    item *it = NULL;

    if (id < POWER_SMALLEST || id > power_largest) {//下标越界
        MEMCACHED_SLABS_ALLOCATE_FAILED(size, 0);
        return NULL;
    }

    p = &slabclass[id];
    assert(p->sl_curr == 0 || ((item *)p->slots)->slabs_clsid == 0);

    //如果p->sl_curr等于0，就说明该slabclass_t没有空闲的item了。
    //此时需要调用do_slabs_newslab申请一个内存页
    /* fail unless we have space at the end of a recently allocated page,
       we have something on our freelist, or we could allocate a new page */
    if (! (p->sl_curr != 0 || do_slabs_newslab(id) != 0)) {
        /* We don't have more memory available */
        //当p->sl_curr等于0并且do_slabs_newslab的返回值等于0时，进入这里
        ret = NULL; //说明指定给该memcached内存达到了上限
    } else if (p->sl_curr != 0) {
        /* return off our freelist */
        //除非do_slabs_newslab调用失败，否则都会来到这里。无论一开始sl_curr是否为0.
        //p->slots指向第一个空闲的item，此时要把第一个空闲的item分配出去
        it = (item *)p->slots;
        p->slots = it->next;//slots指向下一个空闲的item
        if (it->next) it->next->prev = 0;
        p->sl_curr--; //空闲数目减一
        ret = (void *)it;
    }

    if (ret) {
        p->requested += size;//增加slabclass分配出去的字节数
        MEMCACHED_SLABS_ALLOCATE(size, id, p->size, ret);
    } else {
        MEMCACHED_SLABS_ALLOCATE_FAILED(size, id);
    }

    return ret;
}

初始化slabs,分配chunk

//slabclass_t中slab的数目是慢慢增多的。该函数的作用是为slabclass_t申请多一个slab
//参数id指明是slabclass数组中的那个slabclass_t
static int do_slabs_newslab(const unsigned int id) {
    slabclass_t *p = &slabclass[id];
    //setting.slab_rassingn的默认值为false，这里就采用false
    int len = settings.slab_reassign ? settings.item_size_max
        : p->size * p->perslab;
    char *ptr;

    //mem_malloced的值通过环境变量设置，默认为0
    if ((mem_limit && mem_malloced + len > mem_limit && p->slabs > 0) ||
        (grow_slab_list(id) == 0) || //增长slab_list(失败返回0)。一般会成功，除非无法分配内存
        ((ptr = memory_allocate((size_t)len)) == 0)) { //分配len字节内存(也就是一个页)

        MEMCACHED_SLABS_SLABCLASS_ALLOCATE_FAILED(id);
        return 0;
    }

    memset(ptr, 0, (size_t)len);//清空内存块是必须的

    //将这块内存切成一个个的item，当然item的大小由id所控制
    split_slab_page_into_freelist(ptr, id);
    //这里很巧妙，如果是第一次do_slabs_newslab，那么p->slabs=0，++之后就=1，
    //第二次来自然就是2，这次会把第2次申请的page也就是ptr串在p->slab_list[1]上
    //将分配得到的内存页交由slab_list掌管
    p->slab_list[p->slabs++] = ptr; //salb_list[0] = ptr,然后slabs++后为1，也就是[0]指向第一个item
    mem_malloced += len;
    MEMCACHED_SLABS_SLABCLASS_ALLOCATE(id);

    return 1;
}

slab_list第一次是存16*8个字节

//增加slab_list成员指向的内存，也就是增大slab_list数组。使得可以有更多的slab分配器
//除非内存分配失败，否则都是返回-1，无论是否真正增大了
//扩充trunk数目,重新分配slabs个数,默认是分配16个页,后续按照2倍增加
static int grow_slab_list (const unsigned int id) {
    slabclass_t *p = &slabclass[id];
    if (p->slabs == p->list_size) {//用完了之前申请到的slab_list数组的所有元素
        size_t new_size =  (p->list_size != 0) ? p->list_size * 2 : 16;
        void *new_list = realloc(p->slab_list, new_size * sizeof(void *));
        if (new_list == 0) return 0;
        p->list_size = new_size;
        p->slab_list = new_list;
    }
    return 1;
}

//把一个slab，例如1M空间切割成perslab个chunk，通过next和prev指针链接在一起
static void split_slab_page_into_freelist(char *ptr, const unsigned int id) {
    slabclass_t *p = &slabclass[id];
    int x;
    for (x = 0; x < p->perslab; x++) {
        do_slabs_free(ptr, 0, id);
        ptr += p->size;
    }
}

转载：https://blog.csdn.net/unix21/article/details/8572529