转载请注明出处：http://blog.csdn.net/luotuo44/article/details/42913549

锁级别：

        从前面的《扩展哈希表》知道：哈希表进行扩展时，有一个专门的线程负责将item数据从旧哈希表迁移到新哈希表(由此，也称这个线程为迁移线程)。此外，还有一些worker线程会时不时访问item(包括插入、删除和获取)。这些线程的对item所做的操作基本上都是互斥的，必须加锁控制。

        如果只使用一个锁，抢到该锁才能使用哈希表，没有抢到则不能使用。这样的话memcached的效率将变得相当低。为此，memcached像数据库那样使用不同级别的锁。memcached定义了两个级别的锁，段级别和全局级别。在平时(不进行哈希表扩展时)，使用段级别的锁。在扩展哈希表时，使用全局级别的锁。

        段级别是什么级别？将哈希表按照几个桶一段几个桶一段地平均分，一个段就对应有多个桶。所以整个哈希表有多个段级别锁。由于段级别锁的数量在程序的一开始就已经确定了，不会再变的了。而随着哈希表的扩展，桶的数量是会增加的。所以随着哈希表的扩展，越来越多的桶对应一个段，也就是说越来越多的桶对应一个锁。

        在哈希表扩展时，迁移线程和workers线程都使用全局锁。这些线程竞争全局锁，抢到锁才允许对哈希表的item进行操作。在非扩展时，迁移线程处于休眠状态，workers线程使用段级别锁，抢到了某个段锁，就允许访问对应的多个桶。这样如果不同的worker线程访问不同的段，那么就可以同时访问了，增加了并发量。

        下面看一下段级别锁和全局级别锁的定义。thread_init函数分配并初始化段级别锁。

static pthread_mutex_t *item_locks;//指向段锁数组的指针
/* size of the item lock hash table */
static uint32_t item_lock_count;//段锁的数量
static unsigned int item_lock_hashpower;
static pthread_mutex_t item_global_lock;//全局锁

#define hashsize(n) ((unsigned long int)1<<(n))

void thread_init(int nthreads, struct event_base *main_base) {
    int         i;
    int         power;

    pthread_mutex_init(&cache_lock, NULL);

    pthread_mutex_init(&init_lock, NULL);
    pthread_cond_init(&init_cond, NULL);

	//nthreads是workers线程的数量，由main函数调用时传入来
    if (nthreads < 3) {
        power = 10;
    } else if (nthreads < 4) {
        power = 11;
    } else if (nthreads < 5) {
        power = 12;
    } else {//最大为13
        /* 8192 buckets, and central locks don't scale much past 5 threads */
        power = 13;
    }

	//power是2的幂
    item_lock_count = hashsize(power);
    item_lock_hashpower = power;

	//哈希表中段级别的锁。并不是一个桶就对应有一个锁。而是多个桶共用一个锁
    item_locks = calloc(item_lock_count, sizeof(pthread_mutex_t));
    if (! item_locks) {
        perror("Can't allocate item locks");
        exit(1);
    }
    for (i = 0; i < item_lock_count; i++) {
        pthread_mutex_init(&item_locks[i], NULL);
    }


    pthread_mutex_init(&item_global_lock, NULL);

    ...
}

切换锁级别：

        现在看一下怎么使用段级别锁和全局级别锁。迁移线程并不会使用段级别锁，在assoc.c的assoc_maintenance_thread函数中，迁移线程只会调用item_lock_global()函数锁上全局锁item_global_lock。这里主要是看workers线程是怎么使用段级别锁和全局级别锁的。

worker线程的锁级别：

        workers线程如果要访问哈希表的item，会先调用item_lock函数进行加锁。item_lock函数会根据需要自动选择使用段级别锁还是全局级别锁。下面是具体的代码。

//memcached.h文件
//item锁级别
enum item_lock_types {
    ITEM_LOCK_GRANULAR = 0, //段级别
    ITEM_LOCK_GLOBAL //全局级别
};

//thread.c文件
static pthread_key_t item_lock_type_key;//线程私有数据的键值

void item_lock(uint32_t hv) {
	//获取线程私有变量
    uint8_t *lock_type = pthread_getspecific(item_lock_type_key);
	//likely这个宏定义用于代码指令优化
	//likely(*lock_type == ITEM_LOCK_GRANULAR)用来告诉编译器
	//*lock_type等于ITEM_LOCK_GRANULAR的可能性很大
	if (likely(*lock_type == ITEM_LOCK_GRANULAR)) {//使用段级别锁的概率很大
		//对某些桶的item加锁
        mutex_lock(&item_locks[hv & hashmask(item_lock_hashpower)]);
    } else {
    	//对所有item加锁
        mutex_lock(&item_global_lock);
    }
}

void item_unlock(uint32_t hv) {
    uint8_t *lock_type = pthread_getspecific(item_lock_type_key);
    if (likely(*lock_type == ITEM_LOCK_GRANULAR)) {
        mutex_unlock(&item_locks[hv & hashmask(item_lock_hashpower)]);
    } else {
        mutex_unlock(&item_global_lock);
    }
}

        可以看到memcached根据线程私有变量(对应的键值为item_lock_type_key)确定当前要使用哪个锁。只要为每一个worker线程都设置键值为item_lock_type_key的线程私有数据。要切换锁，直接修改线程的私有数据即可。接着看一下workers线程私有数据的初始化。

static LIBEVENT_THREAD *threads;

void thread_init(int nthreads, struct event_base *main_base) {
	...
	pthread_key_create(&item_lock_type_key, NULL);

    for (i = 0; i < nthreads; i++) {
		//创建worker线程，线程函数为worker_libevent, 线程参数为&threads[i]
        create_worker(worker_libevent, &threads[i]);
    }
	...
}

static void *worker_libevent(void *arg) {//这个函数也是在初始化时调用的
    LIBEVENT_THREAD *me = arg;

    me->item_lock_type = ITEM_LOCK_GRANULAR;//初试状态使用段级别锁
	//为workers线程设置线程私有数据
	//因为所有的workers线程都会调用这个函数，所以所有的workers线程都设置了相同键值的
	//线程私有数据
    pthread_setspecific(item_lock_type_key, &me->item_lock_type);
	...
}

实现切换：

        可以看到每个线程的线程私有数据是每个线程都独有的LIBEVENT_THREAD结构体的成员变量item_lock_type。只要根据需要把workers线程的item_lock_type变量修改就可以完成锁的切换。哈希表迁移线程会在assoc.c文件中的assoc_maintenance_thread函数调用switch_item_lock_type函数，让所有的workers线程都切换到段级别锁或者全局级别锁。现在来看一下具体是怎么实现的。

void switch_item_lock_type(enum item_lock_types type) {
    char buf[1];
    int i;

    switch (type) {
        case ITEM_LOCK_GRANULAR:
            buf[0] = 'l';//用l表示ITEM_LOCK_GRANULAR 段级别锁
            break;
        case ITEM_LOCK_GLOBAL:
            buf[0] = 'g';//用g表示ITEM_LOCK_GLOBAL 全局级别锁
            break;
        default:
            fprintf(stderr, "Unknown lock type: %d\n", type);
            assert(1 == 0);
            break;
    }

    pthread_mutex_lock(&init_lock);
    init_count = 0;
    for (i = 0; i < settings.num_threads; i++) {
		//通过向worker监听的管道写入一个字符通知worker线程
        if (write(threads[i].notify_send_fd, buf, 1) != 1) {
            perror("Failed writing to notify pipe");
            /* TODO: This is a fatal problem. Can it ever happen temporarily? */
        }
    }

	//等待所有的workers线程都把锁切换到type指明的锁类型
    wait_for_thread_registration(settings.num_threads);
    pthread_mutex_unlock(&init_lock);
}

static void wait_for_thread_registration(int nthreads) {
    while (init_count < nthreads) {
        pthread_cond_wait(&init_cond, &init_lock);
    }
}

        因为所有的workers线程都在处于event_base循环中，可以直接往workers线程监听的管道中写入一个字节就能通知workers线程了。

        迁移线程为什么要这么迂回曲折地切换workers线程的锁类型呢？直接修改所有线程的LIBEVENT_THREAD结构的item_lock_type成员变量不就行了吗？

        这主要是因为迁移线程不知道worker线程此刻在干些什么。如果worker线程正在访问item，并抢占了段级别锁。此时你把worker线程的锁切换到全局锁，等worker线程解锁的时候就会解全局锁(参考前面的item_lock和item_unlock代码)，这样程序就崩溃了。所以不能迁移线程去切换，只能迁移线程通知worker线程，然后worker线程自己去切换。当然是要worker线程忙完了手头上的事情后，才会去修改切换的。所以迁移线程在通知完所有的worker线程后，会调用wait_for_thread_registration函数休眠等待所有的worker线程都切换到指定的锁类型后才醒来。

        现在来看一下workers线程是怎么切换的。因为前面迁移线程往workers线程监听的管道写入了一个字符，所以我们直接看workers线程设置的管道event监听函数thread_libevent_process。

static void thread_libevent_process(int fd, short which, void *arg) {
    LIBEVENT_THREAD *me = arg;
    char buf[1];

    if (read(fd, buf, 1) != 1)
        if (settings.verbose > 0)
            fprintf(stderr, "Can't read from libevent pipe\n");

    switch (buf[0]) {
  	... 
    case 'l':
    me->item_lock_type = ITEM_LOCK_GRANULAR;//切换item到段级别
	//唤醒睡眠在init_cond条件变量上的迁移线程
    register_thread_initialized();
        break;
    case 'g':
    me->item_lock_type = ITEM_LOCK_GLOBAL;//切换item锁到全局级别
    register_thread_initialized();
        break;
    }
}


static void register_thread_initialized(void) {
    pthread_mutex_lock(&init_lock);
    init_count++;
    pthread_cond_signal(&init_cond);
    pthread_mutex_unlock(&init_lock);
}

按需切换：

        现在已经看完了基础设施的构建，来看回迁移线程是怎么调控一切的吧。

void item_lock_global(void) {
    mutex_lock(&item_global_lock);
}

void item_unlock_global(void) {
    mutex_unlock(&item_global_lock);
}


static void *assoc_maintenance_thread(void *arg) {

	//do_run_maintenance_thread是全局变量，初始值为1，在stop_assoc_maintenance_thread
	//函数中会被赋值0，终止迁移线程
    while (do_run_maintenance_thread) {
        int ii = 0;

        /* Lock the cache, and bulk move multiple buckets to the new
         * hash table. */
        item_lock_global();//锁上全局级别的锁，全部的item都在全局锁的控制之下
		//锁住哈希表里面的item。不然别的线程对哈希表进行增删操作时，会出现
		//数据不一致的情况.在item.c的do_item_link和do_item_unlink可以看到
		//其内部也会锁住cache_lock锁.
        mutex_lock(&cache_lock);

		...//这里是迁移一个桶的数据到新哈希表

		//遍历完一个桶的所有item后，就释放锁
        mutex_unlock(&cache_lock);
        item_unlock_global();//释放全局锁


        if (!expanding) {//不再需要迁移数据了。
            /* finished expanding. tell all threads to use fine-grained(细粒度的) locks */
			//进入到这里，说明已经不需要迁移数据(停止扩展了)。
			//告诉所有的workers线程，访问item时，切换到段级别的锁。
			//会阻塞到所有workers线程都切换到段级别的锁
            switch_item_lock_type(ITEM_LOCK_GRANULAR);
			
            slabs_rebalancer_resume();
            /* We are done expanding.. just wait for next invocation */
            mutex_lock(&cache_lock);
            started_expanding = false; //重置

			//挂起扩展线程，直到别的线程插入数据后发现item数量已经到了1.5倍哈希表大小，
			//此时调用别的线程调用assoc_start_expand函数，该函数会调用pthread_cond_signal
			//唤醒扩展线程
            pthread_cond_wait(&maintenance_cond, &cache_lock);
            /* Before doing anything, tell threads to use a global lock */
            mutex_unlock(&cache_lock);
            slabs_rebalancer_pause();

			//从maintenance_cond条件变量中醒来，说明又要开始扩展哈希表和迁移数据了。
			//迁移线程在迁移一个桶的数据时是锁上全局级别的锁.
			//此时workers线程不能使用段级别的锁，而是要使用全局级别的锁，
			//所有的workers线程和迁移线程一起，争抢全局级别的锁.
			//哪个线程抢到了，才有权利访问item.
			//下面一行代码就是通知所有的workers线程，把你们访问item的锁切换
			//到全局级别的锁。switch_item_lock_type会通过条件变量休眠等待，
			//直到，所有的workers线程都切换到全局级别的锁，才会醒来过
            switch_item_lock_type(ITEM_LOCK_GLOBAL);
            mutex_lock(&cache_lock);
            assoc_expand();//申请更大的哈希表,并将expanding设置为true
            mutex_unlock(&cache_lock);
        }
    }
    return NULL;
}

        眼尖的读者可能还看到了mutex_lock(&cache_lock)和slabs_rebalancer_resume()。不错，这又是对两个锁进行加锁处理。为什么要加这两个锁呢？是因为除了worker线程外还有其他一些线程，这些线程会操作LRU队列和哈希表。但这些线程没有像worker线程那样，可以被通知。所以只能再使用另外的大锁。当然这些线程大部分时间都是在休眠，对性能不会影响太大。由于涉及其他线程，本篇博文也是不会进一步进行讲解这两个锁。

引用计数：

为何需要引用计数：

        读者如果对C++的shared_ptr有所了解，那会更容易看懂接下来的内容。因为shared_ptr也用到引用计数的概念。

        为了保证线程安全，在访问和操作一个item时就必须加锁。而加锁就必然会导致性能的下降。如果在处理读操作的一开始就加锁，直到处理完读操作才解锁(即全程加锁)，那么对于一些热门数据将难于进行更新操作(也就是写操作)。这是因为对于热门数据有读操作会相当频繁，写操作将迟迟得不到执行。如果不全程加锁，那么又会出现一个worker线程在读一个item，而另外一个worker线程在删除这个item。如果这个item被删除了，那么正在读的item就操作一个非法的item。为了性能和处理这种情况，memcached使用引用计数技术。这里的引用计数C++的智能指针shared_ptr原理是一样的。当没有线程在引用这个item后，就会删除这个item(实际是将内存归还给slab分配器)。

        memcached为了使用引用计数技术，在item结构体定义了一个refcount成员，用于记录这个item被引用(被worker线程占用)的总数。当然增加和减少item的引用计数都必须是原子操作。为此，memcached定义了两个函数。

unsigned short refcount_incr(unsigned short *refcount) {
#ifdef HAVE_GCC_ATOMICS
    return __sync_add_and_fetch(refcount, 1);
#elif defined(__sun)
    return atomic_inc_ushort_nv(refcount);
#else
    unsigned short res;
    mutex_lock(&atomics_mutex);
    (*refcount)++;
    res = *refcount;
    mutex_unlock(&atomics_mutex);
    return res;
#endif
}

unsigned short refcount_decr(unsigned short *refcount) {
#ifdef HAVE_GCC_ATOMICS
    return __sync_sub_and_fetch(refcount, 1);
#elif defined(__sun)
    return atomic_dec_ushort_nv(refcount);
#else
    unsigned short res;
    mutex_lock(&atomics_mutex);
    (*refcount)--;
    res = *refcount;
    mutex_unlock(&atomics_mutex);
    return res;
#endif
}

//refcount_incr(&it->refcount);一般是这样调用的
// refcount_decr(&it->refcount)

        如果不懂__sync_add_and_fetch和__sync_sub_and_fetch，那么赶紧谷歌之。因为它们是比较重要的函数，可以用来制作无锁队列。这两个函数都会返回操作后的值。

怎么使用引用计数：

        当然即使有了引用计数还是需要加锁的。因为在获取item和增加引用计数这一间隔，可能有其他线程把这个item给删除了。所以一般流程是这样：worker线程先加锁，然后获取item，之后增加这item的引用计数，最后释放锁。此时worker线程就占有了这个item，其他worker线程在执行删除操作时必须检测这个item的引用计数是否为0，也就是检查是否还有其他worker线程在使用(引用)这个item。下面举一个例子。        

item *item_get(const char *key, const size_t nkey) {
    item *it;
    uint32_t hv;
    hv = hash(key, nkey);
    item_lock(hv);
    it = do_item_get(key, nkey, hv);
    item_unlock(hv);
    return it;
}


/** wrapper around assoc_find which does the lazy expiration logic */
//调用do_item_get的函数都已经加上了item_lock(hv)段级别锁或者全局锁
item *do_item_get(const char *key, const size_t nkey, const uint32_t hv) {
    item *it = assoc_find(key, nkey, hv);//assoc_find函数内部没有加锁
    
    if (it != NULL) {//找到了，此时item的引用计数至少为1
        refcount_incr(&it->refcount);//线程安全地自增一
		...
    }

 	...

    return it;
}

void item_remove(item *item) {
    uint32_t hv;
    hv = hash(ITEM_key(item), item->nkey);

    item_lock(hv);
    do_item_remove(item);
    item_unlock(hv);
}

void do_item_remove(item *it) {
    
    assert((it->it_flags & ITEM_SLABBED) == 0);
    assert(it->refcount > 0);

    if (refcount_decr(&it->refcount) == 0) {//引用计数等于0的时候归还
        item_free(it);//归还该item给slab分配器
    }
}

        可以看到，这是因为减少一个item的引用数可能要删除这个item。为什么呢？考虑这样的情景，线程A因为要读一个item而增加了这个item的引用计数，此时线程B进来了，它要删除这个item。这个删除命令是肯定会执行的，而不是说这个item被别的线程引用了就不执行删除命令。但又肯定不能马上删除，因为线程A还在使用这个item，此时memcached就采用延迟删除的做法。线程B执行删除命令时减多一次item的引用数，使得当线程A释放自己对item的引用后，item的引用数变成0。此时item就被释放了(归还给slab分配器)。

        有一点要注意：当一个item插入到哈希表和LRU队列后，那么这个item就被哈希表和LRU队列所引用了。此时，如果没有其他线程在引用这个item的话，那么这个item的引用数为1(哈希表和LRU队列看作一个引用)。所以一个worker线程要删除一个item(当然在删除前这个worker线程要占有这个item)，那么需要减少两次item的引用数，一次是减少哈希表和LRU队列的引用，另外一次是减少自己的引用。所以经常能在代码中看到删除一个item需要调用函数do_item_unlink (it, hv)和do_item_remove(it)这两个函数。

tail_repair_time：

        考虑这样的情况:某个worker线程通过refcount_incr增加了一个item的引用数。但由于某种原因(可能是内核出了问题)，这个worker线程还没来得及调用refcount_decr就挂了。此时这个item的引用数就肯定不会等于0，也就是总有worker线程占用着它.但实际上这个worker线程早就挂了。所以对于这种情况需要修复。修复也很多简单：直接把这个item的引用计数赋值为1。

        根据什么判断某一个worker线程挂了呢?首先在memcached里面，一般来说，任何函数都的调用都不会耗时太大的，即使这个函数需要加锁。所以如果这个item的最后一次访问时间距离现在都比较遥远了，但它却还被一个worker线程所引用，那么就几乎可以判断这个worker线程挂了。在1.4.16版本之前，这个时间距离都是固定的为3个小时。从1.4.16开就使用settings.tail_repair_time存储时间距离，可以在启动memcached的时候设置，默认时间距离为1个小时。现在这个版本1.4.21默认都不进行这个修复了，settings.tail_repair_time的默认值为0。因为memcached的作者很少看到这个bug了，估计是因为操作系统的进一步稳定。上面的版本说明来自链接1和链接2。

        上面进行了理论说明，下面看一下memcached实现吧。

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;
    item *it = NULL;
    char suffix[40];
	//要存储这个item需要的总空间
    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);

    item *search;
    item *next_it;

    search = tails[id];
    for (;search != NULL; search=next_it) {
        next_it = search->prev;

        uint32_t hv = hash(ITEM_key(search), search->nkey);
		
        /* Now see if the item is refcount locked */
        if (refcount_incr(&search->refcount) != 2) {

            refcount_decr(&search->refcount);
            /* 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) {//在这个时间距离内都没有访问过
                search->refcount = 1;//释放线程对item的引用
                do_item_unlink_nolock(search, hv);//这里会把item从哈希表和LRU队列中删除并将引用计数减一
            }
            continue;
        }

		...
    }

	...

}