分布式缓存系统 Memcached slab和item的主要操作

上节在分析slab内存管理机制时分析Memcached整个Item存储系统的初始化过程slabs_init()函数：分配slabclass数组空间，到最后将各slab划分为各种级别大小的空闲item并挂载到对应大小slab的空闲链表slots上。本节将继续分析对slab和item的主要操作过程。

slab机制中所采用的LRU算法：

在memcached运行过程中，要把一个item调入内存，但内存已无空闲空间时，为了保证程序能正常运行，系统必须从内存中调出一部分数据，送磁盘的对换区中。但应将哪些数据调出，须根据一定的算法来确定。通常，把选择换出数据(页面)的算法称为页面置换算法(Page Replacement Algorithms)。 Memcached采用最近最久未使用（LRU）置换算法，是根据数据（页面）调入内存后的使用情况进行决策的。由于无法预测各页面将来的使用情况，只能利用“最近的过去”作为“最近的将来”的近似，因此，LRU置换算法是选择最近最久未使用的页面予以淘汰。

每个slabclass维护一个双向链表，所有分配出去的item按照最近访问时间依次放到该链表中，该链表也就相当于LRU队列。所有slabclass的链表头 尾分表保存在*heads、 *tails两个数组中。当内存不足时，memcached会对应大小的slabclass维护的双向链表的尾部开始检测，即最近最久未使用的item，往前一直寻找合适的item予以淘汰。所以该LRU算法为局部的slabclass淘汰机制。但是在一些特定情形也会可能引起一些不必要的麻烦，可以在运行时加入”-M”参数禁止该算法。

Slab相关操作：

/*
 *根据给定的大小在slabclass数组中找到合适的slabclass，返回该slabclass的下标
 *如果给定大小为0，或者大于最大的slab的item大小，则返回0
 */
unsigned int slabs_clsid(const size_t size) {
    int res = POWER_SMALLEST;

    if (size == 0)
        return 0;
    while (size > slabclass[res].size)
        if (res++ == power_largest)    /* won't fit in the biggest slab */
            return 0;//物件过大，不能直接存放
    return res;
}

//当该slabcalss的slots空闲链表为空时为指定大小的item分配空间：
//size为给定的item大小，id为对应的slabclass编号,返回item指针
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) {//id编号不合法
        MEMCACHED_SLABS_ALLOCATE_FAILED(size, 0);
        return NULL;
    }

    p = &slabclass[id];//获取对应的slabclass
    assert(p->sl_curr == 0 || ((item *)p->slots)->slabs_clsid == 0);

    /* 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 */
 //如果空闲链表为空且 为该大小的slab分配一个slabcalss(1M)失败
    if (! (p->sl_curr != 0 || do_slabs_newslab(id) != 0)) {
        /* We don't have more memory available */
        ret = NULL;//如果所有空间都用尽的时候，则返回NULL表示目前资源已经枯竭了。
    } else if (p->sl_curr != 0) {//不管是原来的空闲item或者新分配的item,都将挂在到slots上
        /* return off our freelist */
        it = (item *)p->slots;//返回空闲链表的头item
        p->slots = it->next;
        if (it->next) it->next->prev = 0;
        p->sl_curr--;
        ret = (void *)it;
    }

    if (ret) {
        p->requested += size;
        MEMCACHED_SLABS_ALLOCATE(size, id, p->size, ret);
    } else {
        MEMCACHED_SLABS_ALLOCATE_FAILED(size, id);
    }

    return ret;
}
//创建空闲item ，挂载到对应slabclass的空闲链表中  
static void do_slabs_free(void *ptr, const size_t size, unsigned int id) {  
    slabclass_t *p;  
    item *it;  
  
    assert(((item *)ptr)->slabs_clsid == 0);  
    assert(id >= POWER_SMALLEST && id <= power_largest);//判断id有效性  
    if (id < POWER_SMALLEST || id > power_largest)  
        return;  
  
    MEMCACHED_SLABS_FREE(size, id, ptr);  
    p = &slabclass[id];  
  
    it = (item *)ptr;  
    it->it_flags |= ITEM_SLABBED;  
    it->prev = 0;  
    it->next = p->slots;//挂载到slabclass的空闲链表中  
    if (it->next) it->next->prev = it;  
    p->slots = it;  
  
    p->sl_curr++;//空闲item个数+1  
    p->requested -= size;//已经申请到的空间数量更新  
    return;  
} 

Item相关操作：

主要接口函数：

do_item_alloc() ：

首先调用slabs_clsid给item归类，然后调用slabs_alloc()函数分配内存，当可使用空间枯竭了的时候，就开始使用LRU算法了，从一个对应大小的尾部tails[id]开始，向前不断尝试能否释放，当发现一个item当前没有被使用(引用计数refcount为0)，且其生存时间已经为0或生存时间大于现在时间，就果断的把它释放掉。并再次调用slabs_alloc(),作者在此提到有一种非常罕见的bug能够使引用计数(refcount)发生泄漏，处理方法是，当item连续保持锁定状态3小时以上，说明它已经足够陈旧了，应该果断将其释放。最后再将数据复制到分配的空间内。

 item_free()：

 调用slabs_free将其释放。

 do_item_link()：

 1.调用assoc_insert()将item指针插入hash表中

 2.调用get_cas_id()给it的cas_id赋值。

3.调用item_link_q(),把该item插入LRU队列的最前面

do_item_unlink ()：
1.调用assoc_delete()在hash表中删除此item

2.调用item_unlink_q()从该slab class去除此item的连接，此处考虑了item在链表头部，尾部等各种情况。

3.最后当引用计数为0的时候，调用item_free()将其释放。

 do_item_remove () ：

减少引用计数refcount，当发现引用计数为0的时候，就将其释放。

 do_item_update ()：

先调用item_unlink_q(),更新了时间以后，再调用item_link_q()。将其重新连接到LRU队列之中，即让该item移到LRU队列的最前。

do_item_replace ()：

 调用do_item_unlink()解除原有item的连接，再调用do_item_link()连接到新的item。

 item_get () ：

值得说明的是，memcached的懒惰删除逻辑在这里有体现。就是当你需要get 一个item的时候才考虑该item是否应该删除。 首先调用do_item_get_notedeleted()函数，根据关键字使用assoc_find()查找item,如果没找到，返回NULL，再判断是否达到最大的生存时间，如果是的话，就do_item_unlink该item，返回NULL。 如果该item没有满足删除条件，将其引用计数加1，并返回该item的指针。

具体分析见代码：

//key:键 nkey：键的长度 falgs：键falg    exptime:过期时间（相对） cur_hv哈希值
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) {
    uint8_t nsuffix;//key的后缀长度
    item *it = NULL;
    char suffix[40];//存放key后缀
 //该item所需的总长度
    size_t ntotal = item_make_header(nkey + 1, flags, nbytes, suffix, &nsuffix);
    if (settings.use_cas) {
        ntotal += sizeof(uint64_t);
    }

 //以item总长来计算所属的slabclass
    unsigned int id = slabs_clsid(ntotal);
    if (id == 0)
        return 0;

    mutex_lock(&cache_lock);
    /* do a quick check if we have any expired items in the tail.. */
    int tries = 5;//执行5次尝试检查，（快速，而不是全部都检查一遍）
    /* Avoid hangs if a slab has nothing but refcounted stuff in it. */
 //将所有过期时间旧于设定的oldest_live时间的所有item视为过期item，以避免那些实际没用，但却被一直引用的item
    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;//对所有item设定的item最长生存时间

    search = tails[id];//serch指向该slabclass对应的item链表尾部
    /* We walk up *only* for locked items. Never searching for expired.
    * Waste of CPU for almost all deployments */
    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. */
            tries++;
            continue;
        }
  //计算哈希值，用于锁定serch的该item，避免其他调用者增加该item的引用计数
        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 */
  //如果该item正好是当前的item,则忽略
        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) {
            /* Avoid pathological case with ref'ed items in tail */
            do_item_update_nolock(search);
            tries_lrutail_reflocked--;
            tries++;
            refcount_decr(&search->refcount);
            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;
        }

        /* Expired or flushed */
  ////还没过期，或者已经过期但是最近访问时间小于设定的最大过期时间，并且设定的最大过期时间小于当前时间
        if ((search->exptime != 0 && search->exptime < current_time)
            || (search->time <= oldest_live && oldest_live <= current_time))
      {
            itemstats[id].reclaimed++;
            if ((search->it_flags & ITEM_FETCHED) == 0) {
                itemstats[id].expired_unfetched++;
            }
            it = search;
            slabs_adjust_mem_requested(it->slabs_clsid, ITEM_ntotal(it), ntotal);
            do_item_unlink_nolock(it, hv);//从哈希表  及item链表中删除，递减引用计数
            /* Initialize the item block: */
            it->slabs_clsid = 0;//注意：slabs_clsid = 0表明不属于任何slabclass
        } else if ((it = slabs_alloc(ntotal, id)) == NULL) {
            tried_alloc = 1;
            if (settings.evict_to_free == 0) {
                itemstats[id].outofmemory++;
            } else {
                itemstats[id].evicted++;
                itemstats[id].evicted_time = current_time - search->time;
                if (search->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.
                */
                if (settings.slab_automove == 2)
                    slabs_reassign(-1, id);
            }
        }

        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.
    */
 //初始化item。 由于该item已不在lru队列中，因此可以无锁操作
    it->refcount = 1;    //引用计数初始化为1 /* the caller will have a reference */
    mutex_unlock(&cache_lock);
    it->next = it->prev = it->h_next = 0;
    it->slabs_clsid = id;//其所属id已经计算好

    DEBUG_REFCNT(it, '*');
    it->it_flags = settings.use_cas ? ITEM_CAS : 0;
    it->nkey = nkey;//
    it->nbytes = nbytes;//
    memcpy(ITEM_key(it), key, nkey);
    it->exptime = exptime;
    memcpy(ITEM_suffix(it), suffix, (size_t)nsuffix);
    it->nsuffix = nsuffix;
    return it;
}

//释放一个item，挂载到slots的头部
void item_free(item *it) {
    size_t ntotal = ITEM_ntotal(it);
    unsigned int clsid;
    assert((it->it_flags & ITEM_LINKED) == 0);
    assert(it != heads[it->slabs_clsid]);
    assert(it != tails[it->slabs_clsid]);
    assert(it->refcount == 0);

    /* so slab size changer can tell later if item is already free or not */
    clsid = it->slabs_clsid;
    it->slabs_clsid = 0;
    DEBUG_REFCNT(it, 'F');
    slabs_free(it, ntotal, clsid);
}

//将item插入到哈希表  lru队列
int do_item_link(item *it, const uint32_t hv) {
    MEMCACHED_ITEM_LINK(ITEM_key(it), it->nkey, it->nbytes);
    assert((it->it_flags & (ITEM_LINKED|ITEM_SLABBED)) == 0);
    mutex_lock(&cache_lock);
    it->it_flags |= ITEM_LINKED;
    it->time = current_time;

    STATS_LOCK();
    stats.curr_bytes += ITEM_ntotal(it);
    stats.curr_items += 1;
    stats.total_items += 1;
    STATS_UNLOCK();

    /* Allocate a new CAS ID on link. */
    ITEM_set_cas(it, (settings.use_cas) ? get_cas_id() : 0);
    assoc_insert(it, hv);//将该item指针插入到hash表的对应bucket的头部（正常情况下插入到主表，扩容时插入到原表）
    item_link_q(it);//将该item指针插入到对应slab的头部
    refcount_incr(&it->refcount);//递增引用计数
    mutex_unlock(&cache_lock);

    return 1;
}

//将该item从哈希表  lru队列中删除，递减引用计数
void do_item_unlink(item *it, const uint32_t hv) {
    MEMCACHED_ITEM_UNLINK(ITEM_key(it), it->nkey, it->nbytes);
    mutex_lock(&cache_lock);
    if ((it->it_flags & ITEM_LINKED) != 0) {
        it->it_flags &= ~ITEM_LINKED;
        STATS_LOCK();
        stats.curr_bytes -= ITEM_ntotal(it);
        stats.curr_items -= 1;
        STATS_UNLOCK();
        assoc_delete(ITEM_key(it), it->nkey, hv);//在哈希表中删除该item
        item_unlink_q(it);//在slab中删除该item
        do_item_remove(it);//如果item引用计数减到0，则释放，挂载到slots头部
    }
    mutex_unlock(&cache_lock);
}

//item引用计数为0，则释放
void do_item_remove(item *it) {
    MEMCACHED_ITEM_REMOVE(ITEM_key(it), it->nkey, it->nbytes);
    assert((it->it_flags & ITEM_SLABBED) == 0);
    assert(it->refcount > 0);

    if (refcount_decr(&it->refcount) == 0) {
        item_free(it);
    }
}

//注意：避免过于频繁的获取该item带来的性能影响，设定最小更新时间ITEM_UPDATE_INTERVAL：
//即至少要这么久才更新一次该item在lru中的位置
//先从lru链表删除，然后更新访问时间，最后在插入到lru的头部
void do_item_update(item *it) {
    MEMCACHED_ITEM_UPDATE(ITEM_key(it), it->nkey, it->nbytes);
 //是否达到更新的要求
    if (it->time < current_time - ITEM_UPDATE_INTERVAL) {
        assert((it->it_flags & ITEM_SLABBED) == 0);

        mutex_lock(&cache_lock);
        if ((it->it_flags & ITEM_LINKED) != 0) {
            item_unlink_q(it);//从lru链表删除
            it->time = current_time;//更新最近访问时间
            item_link_q(it);//更新位置：放到lru头部
        }
        mutex_unlock(&cache_lock);
    }
}

//以新的item替换给定的item（哈希表  lru）
int do_item_replace(item *it, item *new_it, const uint32_t hv) {
    MEMCACHED_ITEM_REPLACE(ITEM_key(it), it->nkey, it->nbytes,
                          ITEM_key(new_it), new_it->nkey, new_it->nbytes);
    assert((it->it_flags & ITEM_SLABBED) == 0);

    do_item_unlink(it, hv);//将旧的item从哈希表  lru队列去掉
    return do_item_link(new_it, hv);//将该新的item加入哈希表  lru中
}

//如果该item过期则返回NULL，且同时在哈希表、lru中删除。（get中体现的正是惰性删除机制）
item *do_item_get(const char *key, const size_t nkey, const uint32_t hv) {
    //mutex_lock(&cache_lock);
    item *it = assoc_find(key, nkey, hv);//在哈希表中查找该item，然后利用该item的clsid信息去lru中查找该item
    if (it != NULL) {
        refcount_incr(&it->refcount);
        /* Optimization for slab reassignment. prevents popular items from
        * jamming in busy wait. Can only do this here to satisfy lock order
        * of item_lock, cache_lock, slabs_lock. */
        if (slab_rebalance_signal &&
            ((void *)it >= slab_rebal.slab_start && (void *)it < slab_rebal.slab_end)) {
            do_item_unlink_nolock(it, hv);
            do_item_remove(it);//递减引用计数
            it = NULL;
        }
    }
    //mutex_unlock(&cache_lock);
    int was_found = 0;

    if (settings.verbose > 2) {
        int ii;
        if (it == NULL) {
            fprintf(stderr, "> NOT FOUND ");
        } else {
            fprintf(stderr, "> FOUND KEY ");
            was_found++;
        }
        for (ii = 0; ii < nkey; ++ii) {
            fprintf(stderr, "%c", key[ii]);
        }
    }

    if (it != NULL) {
        if (settings.oldest_live != 0 && settings.oldest_live <= current_time &&
            it->time <= settings.oldest_live) {//超过最大生存时间，则从lru中删除
            do_item_unlink(it, hv);
            do_item_remove(it);
            it = NULL;
            if (was_found) {
                fprintf(stderr, " -nuked by flush");
            }
        } else if (it->exptime != 0 && it->exptime <= current_time) {//过期
            do_item_unlink(it, hv);
            do_item_remove(it);
            it = NULL;
            if (was_found) {
                fprintf(stderr, " -nuked by expire");
            }
        } else {
            it->it_flags |= ITEM_FETCHED;
            DEBUG_REFCNT(it, '+');
        }
    }

    if (settings.verbose > 2)
        fprintf(stderr, " ");

    return it;
}

经过前面的几节分析，到此对于Memcached中的内存操作，数据item的存储体系即基本存取操作等基础部分的剖析已完成。 
接下来将主要分析Memcached中的核心部分：I/O框架库Libevent的具体应用，以及多线程的处理机制等。这部分也正是前段时间刚接触的内容，因此接下来将给出更为详细的讲解。