// 两个感叹号的作用是将所有的非零值转化为1
#define LIKELY(x) __builtin_expect(!!(x), 1)
#define UNLIKELY(x) __builtin_expect(!!(x), 0)

int likely(int x)
{
    if(LIKELY(x))
    {
        x = 5;
    }
    else
    {
        x = 6;
    }

    return x;
}

int unlikely(int x)
{
    if(UNLIKELY(x))
    {
        x = 5;
    }
    else
    {
        x = 6;
    }

    return x;
}

int normal(int x)
{
    if(x)
    {
        x = 5;
    }
    else
    {
        x = 6;
    }

    return x;
}

struct LRUHandle {
  void* value;
  void (*deleter)(const Slice&, void* value);
  // next_hash 代表在hash桶中下一个元素的位置
  LRUHandle* next_hash;
  // 双链表中下个元素的位置
  LRUHandle* next;
  // 双链表中上个元素的位置
  LRUHandle* prev;
  size_t charge;      // TODO(opt): Only allow uint32_t?
  size_t key_length;
  // 使用次数
  uint32_t refs;
  // key 的hash值用于在LRUCache数组以及hash表中定位
  uint32_t hash;      // Hash of key(); used for fast sharding and comparisons
  char key_data[1];   // Beginning of key

  Slice key() const {
    // For cheaper lookups, we allow a temporary Handle object
    // to store a pointer to a key in "value".
    if (next == this) {
      return *(reinterpret_cast<Slice*>(value));
    } else {
      return Slice(key_data, key_length);
    }
  }
};

class HandleTable {
 public:
  HandleTable() : length_(0), elems_(0), list_(NULL) { Resize(); }
  ~HandleTable() { delete[] list_; }
  // 查找，直接调用FindPointer方法
  LRUHandle* Lookup(const Slice& key, uint32_t hash) {
    return *FindPointer(key, hash);
  }
 // 插入，使用FindPointer方法找到插入位置，在hashtable 对位位置插入，当hashtable 中的元素个数大于桶的个数时候触发hashtable 的扩容
  LRUHandle* Insert(LRUHandle* h) {
    LRUHandle** ptr = FindPointer(h->key(), h->hash);
    LRUHandle* old = *ptr;
    h->next_hash = (old == NULL ? NULL : old->next_hash);
    *ptr = h;
    if (old == NULL) {
      ++elems_;
      if (elems_ > length_) {
        // Since each cache entry is fairly large, we aim for a small
        // average linked list length (<= 1).
        Resize();
      }
    }
    return old;
  }
// 删除hashtable中的元素，同样使用FindPointer方法定位到在hashtable 中的位置
  LRUHandle* Remove(const Slice& key, uint32_t hash) {
    LRUHandle** ptr = FindPointer(key, hash);
    LRUHandle* result = *ptr;
    if (result != NULL) {
      *ptr = result->next_hash;
      --elems_;
    }
    return result;
  }

 private:
  // The table consists of an array of buckets where each bucket is
  // a linked list of cache entries that hash into the bucket.
  uint32_t length_;
  uint32_t elems_;
  LRUHandle** list_;

  // Return a pointer to slot that points to a cache entry that
  // matches key/hash.  If there is no such cache entry, return a
  // pointer to the trailing slot in the corresponding linked list.
  // 找到hash 元素的地址
  LRUHandle** FindPointer(const Slice& key, uint32_t hash) {
	//找到hash 桶的位置，由于length_是2的指数次幂，所以使用 按位与替换mod操作进行加速
    LRUHandle** ptr = &list_[hash & (length_ - 1)];
    // 遍历hash桶中的单链表直到找到相应节点或单链表结束
    while (*ptr != NULL &&
           ((*ptr)->hash != hash || key != (*ptr)->key())) {
      ptr = &(*ptr)->next_hash;
    }
    return ptr;
  }
  // hash 表扩容
  void Resize() {
	// 每次扩容都是翻倍，保证hash桶的个数是2的指数次幂
    uint32_t new_length = 4;
    while (new_length < elems_) {
      new_length *= 2;
    }
    // 创建一个新的hashtable
    LRUHandle** new_list = new LRUHandle*[new_length];
    memset(new_list, 0, sizeof(new_list[0]) * new_length);
    uint32_t count = 0;
    // 逐个桶迁移到新的hashtable
    for (uint32_t i = 0; i < length_; i++) {
      LRUHandle* h = list_[i];
      while (h != NULL) {
        LRUHandle* next = h->next_hash;
        uint32_t hash = h->hash;
        LRUHandle** ptr = &new_list[hash & (new_length - 1)];
        h->next_hash = *ptr;
        *ptr = h;
        h = next;
        count++;
      }
    }
    assert(elems_ == count);
    // 释放掉老的hashtable
    delete[] list_;
    // 替换为新的hashtable
    list_ = new_list;
    length_ = new_length;
  }
};

Cache::Handle* LRUCache::Insert(
    const Slice& key, uint32_t hash, void* value, size_t charge,
    void (*deleter)(const Slice& key, void* value)) {
  MutexLock l(&mutex_);
  // 构造一个新的LRUHandle 对象
  LRUHandle* e = reinterpret_cast<LRUHandle*>(
      malloc(sizeof(LRUHandle)-1 + key.size()));
  e->value = value;
  e->deleter = deleter;
  e->charge = charge;
  e->key_length = key.size();
  e->hash = hash;
  e->refs = 2;  // One from LRUCache, one for the returned handle
  memcpy(e->key_data, key.data(), key.size());
  // 插入到双向链表表头的位置
  LRU_Append(e);
  usage_ += charge;

  LRUHandle* old = table_.Insert(e);
  // 替换掉LRUCache中重复的元素
  if (old != NULL) {
    LRU_Remove(old);
    Unref(old);
  }
  // 触发LRU 淘汰
  while (usage_ > capacity_ && lru_.next != &lru_) {
    // 淘汰双向链表最后的元素
    LRUHandle* old = lru_.next;
    // 删除双向链表中的元素
    LRU_Remove(old);
    // 删除hashtable中的元素
    table_.Remove(old->key(), old->hash);
    Unref(old);
  }
  // 返回刚刚插入的Handle对象，此对象引用计数为2
  return reinterpret_cast<Cache::Handle*>(e);
}

void LRUCache::Erase(const Slice& key, uint32_t hash) {
  MutexLock l(&mutex_);
  // 删除hashtable中对应的元素，
  // 如果该元素存在在hashtable中则在双向链表中进行删除
  LRUHandle* e = table_.Remove(key, hash);
  if (e != NULL) {
    LRU_Remove(e);
    Unref(e);
  }
}

class Arena {
 public:
  Arena();
  ~Arena();

  // Return a pointer to a newly allocated memory block of "bytes" bytes.
  // 分配指定大小的内存，并返回分配内存的首地址
  char* Allocate(size_t bytes);
  // 分配指定大小并且对齐的内存
  // Allocate memory with the normal alignment guarantees provided by malloc
  char* AllocateAligned(size_t bytes);

  // Returns an estimate of the total memory usage of data allocated
  // by the arena.
  // 统计使用了多少内存
  size_t MemoryUsage() const {
    return reinterpret_cast<uintptr_t>(memory_usage_.NoBarrier_Load());
  }

 private:
  // 分配内存，根据情况决定是否使用新的block
  char* AllocateFallback(size_t bytes);
  // 分配一个新的block
  char* AllocateNewBlock(size_t block_bytes);

  // Allocation state
  // 指向一个block中未被使用的内存首地址
  char* alloc_ptr_;
  // 表示一个block还剩余多少空间
  size_t alloc_bytes_remaining_;

  // Array of new[] allocated memory blocks
  // 内存池数组
  std::vector<char*> blocks_;

  // Total memory usage of the arena.
  // 统计内存使用，原子变量
  port::AtomicPointer memory_usage_;
  
  // No copying allowed
  // 将拷贝构造和赋值构造设置为私有
  Arena(const Arena&);
  void operator=(const Arena&);
};

char* Arena::AllocateAligned(size_t bytes) {
  // 获取当前系统指针大小
  const int align = (sizeof(void*) > 8) ? sizeof(void*) : 8;
  // 指针大小必须是2的整数次幂，2的整数次幂的二进制表示中
  // 有且只有1位是1
  assert((align & (align-1)) == 0);   // Pointer size should be a power of 2
  // 判断bytes是不是align 的整数倍，由于align是2的
  // 整数次幂，所以对align的取模运算可以转化为
  // 对（align - 1）进行按位与
  size_t current_mod = reinterpret_cast<uintptr_t>(alloc_ptr_) & (align-1);
  // 为了对齐内存需要新增的大小
  size_t slop = (current_mod == 0 ? 0 : align - current_mod);
  // needed 表示分配对齐的内存所需的大小
  // 后面的逻辑同前
  size_t needed = bytes + slop;
  char* result;
  if (needed <= alloc_bytes_remaining_) {
    result = alloc_ptr_ + slop;
    alloc_ptr_ += needed;
    alloc_bytes_remaining_ -= needed;
  } else {
    // AllocateFallback always returned aligned memory
    result = AllocateFallback(bytes);
  }
  assert((reinterpret_cast<uintptr_t>(result) & (align-1)) == 0);
  return result;
}

char* Arena::AllocateNewBlock(size_t block_bytes) {
  // 分配空间
  char* result = new char[block_bytes];
  // 新的block 加入内存池
  blocks_.push_back(result);
  // 更新memory_usage_
  memory_usage_.NoBarrier_Store(
      reinterpret_cast<void*>(MemoryUsage() + block_bytes + sizeof(char*)));
  return result;
}

Arena::~Arena() {
  for (size_t i = 0; i < blocks_.size(); i++) {
    delete[] blocks_[i];
  }
}

virtual void CreateFilter(const Slice* keys, int n, std::string* dst) const {
    // Compute bloom filter size (in both bits and bytes)
    // bloomfilter 需要多少bit
    size_t bits = n * bits_per_key_;

    // For small n, we can see a very high false positive rate.  Fix it
    // by enforcing a minimum bloom filter length.
    if (bits < 64) bits = 64;
    // 对齐,方便内存读写以及后续位置索引
    size_t bytes = (bits + 7) / 8;
    bits = bytes * 8;
	// 在string 中分配空间
    const size_t init_size = dst->size();
    dst->resize(init_size + bytes, 0);
    // string 的最后一个byte存储使用的hash 函数的个数
    dst->push_back(static_cast<char>(k_));  // Remember # of probes in filter
    // 获得string内部的char 型数组
    char* array = &(*dst)[init_size];
    // 逐个将每个key 写入bloom fliter
    for (int i = 0; i < n; i++) {
      // Use double-hashing to generate a sequence of hash values.
      // See analysis in [Kirsch,Mitzenmacher 2006].
      // leveldb 使用一个hash 函数，每次对hash值向右循环移位17个bit来模拟实现多个hash 函数的效果
      uint32_t h = BloomHash(keys[i]);
      // 每次向右循环移位17个bit
      const uint32_t delta = (h >> 17) | (h << 15);  // Rotate right 17 bits
      for (size_t j = 0; j < k_; j++) {
        // 在整个bit 数组的位置
        const uint32_t bitpos = h % bits;
        // 在char型数组的位置
        array[bitpos/8] |= (1 << (bitpos % 8));
        // 更新获得一个新的hash 数值
        h += delta;
      }
    }
  }

virtual bool KeyMayMatch(const Slice& key, const Slice& bloom_filter) const {
   const size_t len = bloom_filter.size();
   if (len < 2) return false;

   const char* array = bloom_filter.data();
   // 最后一个byte数值代表使用了多少hash函数
   // 除最后一个byte之外代表bit数组
   const size_t bits = (len - 1) * 8;

   // Use the encoded k so that we can read filters generated by
   // bloom filters created using different parameters.
   const size_t k = array[len-1];
   if (k > 30) {
     // Reserved for potentially new encodings for short bloom filters.
     // Consider it a match.
     return true;
   }
// 使用相同的方法模拟多个hash函数计算的hash值
   uint32_t h = BloomHash(key);
   const uint32_t delta = (h >> 17) | (h << 15);  // Rotate right 17 bits
   for (size_t j = 0; j < k; j++) {
     const uint32_t bitpos = h % bits;
     // 找到一个bit位置不匹配，提前返回false
     // 在bit数组中位置的索引和设置时的方法一致
     if ((array[bitpos/8] & (1 << (bitpos % 8))) == 0) return false;
     // 更新获得下一个hash value
     h += delta;
   }
   // 全部匹配return true
   return true;
 }

private:
  // 使用枚举类型定义skiplist 最高高度
  enum { kMaxHeight = 12 };

  // Immutable after construction
  // 用户定制的比较器
  Comparator const compare_;
  // leveldb 实现的简单的内存分配器
  Arena* const arena_;    // Arena used for allocations of nodes
  // skiplist 的前置哨兵节点
  Node* const head_;

  // Modified only by Insert().  Read racily by readers, but stale
  // values are ok.
  // 记录当前skiplist使用的最高高度
  port::AtomicPointer max_height_;   // Height of the entire list

template<typename Key, class Comparator>
bool SkipList<Key,Comparator>::Contains(const Key& key) const {
  // 找到大于等于当前key的第一个node,然后判断node 的key
  // 和传入的key 是否相等
  Node* x = FindGreaterOrEqual(key, NULL);
  if (x != NULL && Equal(key, x->key)) {
    return true;
  } else {
    return false;
  }
}

template<typename Key, class Comparator>
typename SkipList<Key,Comparator>::Node* SkipList<Key,Comparator>::FindGreaterOrEqual(const Key& key, Node** prev)
    const {
  Node* x = head_;
  // level 从0 开始编码
  int level = GetMaxHeight() - 1;
  while (true) {
	// 定位到当前level的下一个节点
    Node* next = x->Next(level);
    // key 没有在当前区间
    if (KeyIsAfterNode(key, next)) {
      // Keep searching in this list
      x = next;
    } else {
	  // key 在当前区间，在低level 继续查找，
	  // 在查找的同时设置prev 节点
      if (prev != NULL) prev[level] = x;
      // 在最低level找到相应位置
      if (level == 0) {
        return next;
      } else {
        // Switch to next list
        level--;
      }
    }
  }
}

template<typename Key, class Comparator>
typename SkipList<Key,Comparator>::Node*
SkipList<Key,Comparator>::FindLessThan(const Key& key) const {
  Node* x = head_;
  int level = GetMaxHeight() - 1;
  while (true) {
    assert(x == head_ || compare_(x->key, key) < 0);
    \// 在当前level 查找
    Node* next = x->Next(level);
    //  if 分支为true 的时候表示需要查找的位置在当前区间
    if (next == NULL || compare_(next->key, key) >= 0) {
      // 在最后一层停止查找
      if (level == 0) {
        return x;
      } else {
        // Switch to next list
        level--;
      }
    } else {
      // 在当前level 就找到了比key 小的节点
      x = next;
    }
  }
}

// length 参数代表拼接的字符串长度
void composeLongstringWithOperator(const unsigned int length,std::string& long_string)
{
    for (size_t i = 0; i < length / 9; i++)
    {
        char str[10];
        // randStr 方法构造长度为9的随机字符串
        long_string += (randStr(str,9));
    }

}

void composeLongstringWithAppend(const unsigned int length,std::string& long_string)
{
    for (size_t i = 0; i < length / 9; i++)
    {
        char str[10];
        long_string.append(randStr(str,9));
    }
}

void composeLongstringWithReserveAndOperator(const unsigned int length,std::string& long_string)
{
    long_string.reserve(length);
    for (size_t i = 0; i < length / 9; i++)
    {
        char str[10];
        long_string += (randStr(str,9));
    }
}

void composeLongstringWithAppend(const unsigned int length,std::string& long_string)
{
    for (size_t i = 0; i < length / 9; i++)
    {
        char str[10];
        long_string.append(randStr(str,9));
    }
}

#include <iostream>
#include <string>
#include <ctime>
#include <chrono>
char* randStr(char* str,const int len)
{
    int i;
    for(i = 0; i < len; ++i)
    {
        str[i] = 'A' + rand() % 26;
    }
    str[++i] = '\0';
    return str;
}

int main(int argc, char* argv[])
{
    (void) argc;
    // 第一个参数代表生成的字符串的长度
    const unsigned int length = atoi(argv[1]);
    // 第二个参数代表使用哪种方法进行拼接
    const unsigned int type = atoi(argv[2]);

    srand(time(NULL));
    auto start = std::chrono::high_resolution_clock::now();
    switch(type)
    {
        case 1:
            std::cout << "composeLongstringWithReserveAndAppend";
            for(int i = 0; i < 10000; i++)
            {
                std::string long_string;
                composeLongstringWithReserveAndAppend(length,long_string);
            }
            break;
        case 2:
            std::cout << "composeLongstringWithReserveAndOperator";
            for(int i = 0; i < 10000; i++)
            {
                std::string long_string;
                composeLongstringWithReserveAndOperator(length,long_string);
            }
            break;
        case 3:
            std::cout << "composeLongstringWithAppend";
            for(int i = 0; i < 10000; i++)
            {
                std::string long_string;
                composeLongstringWithAppend(length,long_string);
            }
            break;
        case 4:
            std::cout << "composeLongstringWithOperator";
            for(int i = 0; i < 10000; i++)
            {
                std::string long_string;
                composeLongstringWithOperator(length,long_string);
            }
            break;
        default:
            return 0;
    }
    auto end = std::chrono::high_resolution_clock::now();
    std::chrono::duration<double> diff = end - start;
    std::cout << " cost " << 1000 * diff.count() << " ms\n";

    return 0;
}

char* EncodeVarint32(char* dst, uint32_t v) {
  // Operate on characters as unsigneds
  unsigned char* ptr = reinterpret_cast<unsigned char*>(dst);
  static const int B = 128;
  if (v < (1<<7)) {
    *(ptr++) = v;
  } else if (v < (1<<14)) {
    *(ptr++) = v | B;
    *(ptr++) = v>>7;
  } else if (v < (1<<21)) {
    *(ptr++) = v | B;
    *(ptr++) = (v>>7) | B;
    *(ptr++) = v>>14;
  } else if (v < (1<<28)) {
    *(ptr++) = v | B;
    *(ptr++) = (v>>7) | B;
    *(ptr++) = (v>>14) | B;
    *(ptr++) = v>>21;
  } else {
    *(ptr++) = v | B;
    *(ptr++) = (v>>7) | B;
    *(ptr++) = (v>>14) | B;
    *(ptr++) = (v>>21) | B;
    *(ptr++) = v>>28;
  }
  return reinterpret_cast<char*>(ptr);
}

// 针对一个byte情况直接处理，大于一个byte 时，
// 使用 GetVarint32PtrFallback 处理
inline const char* GetVarint32Ptr(const char* p,
                                  const char* limit,
                                  uint32_t* value) {
  if (p < limit) {
    uint32_t result = *(reinterpret_cast<const unsigned char*>(p));
    if ((result & 128) == 0) {
      *value = result;
      return p + 1;
    }
  }
  return GetVarint32PtrFallback(p, limit, value);
}

const char* GetVarint32PtrFallback(const char* p,
                                   const char* limit,
                                   uint32_t* value) {
  uint32_t result = 0;
  for (uint32_t shift = 0; shift <= 28 && p < limit; shift += 7) {
    uint32_t byte = *(reinterpret_cast<const unsigned char*>(p));
    p++;
    if (byte & 128) {
      // More bytes are present
      result |= ((byte & 127) << shift);
    } else {
      result |= (byte << shift);
      *value = result;
      return reinterpret_cast<const char*>(p);
    }
  }
  return NULL;
}

char* EncodeVarint64(char* dst, uint64_t v) {
  static const int B = 128;
  unsigned char* ptr = reinterpret_cast<unsigned char*>(dst);
  while (v >= B) {
    *(ptr++) = (v & (B-1)) | B;
    v >>= 7;
  }
  *(ptr++) = static_cast<unsigned char>(v);
  return reinterpret_cast<char*>(ptr);
}

const char* GetVarint64Ptr(const char* p, const char* limit, uint64_t* value) {
  uint64_t result = 0;
  for (uint32_t shift = 0; shift <= 63 && p < limit; shift += 7) {
    uint64_t byte = *(reinterpret_cast<const unsigned char*>(p));
    p++;
    if (byte & 128) {
      // More bytes are present
      result |= ((byte & 127) << shift);
    } else {
      result |= (byte << shift);
      *value = result;
      return reinterpret_cast<const char*>(p);
    }
  }
  return NULL;
}

/*数组的大小都是2指数次幂，这样的好处是可以将 index % hashsize 
变为 index & hashmask*/
#define hashsize(n) ((ub4)1<<(n))
#define hashmask(n) (hashsize(n)-1)
// 初始化函数的作用是为主表分配空间
void assoc_init(const int hashtable_init) {
    if (hashtable_init) {
        hashpower = hashtable_init;
    }
    primary_hashtable = calloc(hashsize(hashpower), sizeof(void *));
    if (! primary_hashtable) {
        fprintf(stderr, "Failed to init hashtable.\n");
        exit(EXIT_FAILURE);
    }
    STATS_LOCK();
    stats_state.hash_power_level = hashpower;
    stats_state.hash_bytes = hashsize(hashpower) * sizeof(void *);
    STATS_UNLOCK();
}

item *assoc_find(const char *key, const size_t nkey, const uint32_t hv) {
    item *it;
    unsigned int oldbucket;
    /* expanding 标识扩展是否完成。
    expand_bucket表示当前扩展的进度，使用位与操作查找bucket位置 */
    if (expanding &&
        (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket)
    {
        it = old_hashtable[oldbucket];
    } else {
        it = primary_hashtable[hv & hashmask(hashpower)];
    }

    item *ret = NULL;
    int depth = 0;
    // 循环单链表查找指定key
    while (it) {
        if ((nkey == it->nkey) && (memcmp(key, ITEM_key(it), nkey) == 0)) {
            ret = it;
            break;
        }
        it = it->h_next;
        ++depth;
    }
    MEMCACHED_ASSOC_FIND(key, nkey, depth);
    return ret;
}

static item** _hashitem_before (const char *key, const size_t nkey, const uint32_t hv) {
    item **pos;
    unsigned int oldbucket;
    // 同理是确定是在old_hashtable 还是在primary_hashtable
    if (expanding &&
        (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket)
    {
        pos = &old_hashtable[oldbucket];
    } else {
        pos = &primary_hashtable[hv & hashmask(hashpower)];
    }
    // 从头结点的位置开始顺序遍历单链表中的节点
    while (*pos && ((nkey != (*pos)->nkey) || memcmp(key, ITEM_key(*pos), nkey))) {
        pos = &(*pos)->h_next;
    }
    return pos;
}

/* grows the hashtable to the next power of 2. */
static void assoc_expand(void) {
    old_hashtable = primary_hashtable;

    primary_hashtable = calloc(hashsize(hashpower + 1), sizeof(void *));
    if (primary_hashtable) {
        if (settings.verbose > 1)
            fprintf(stderr, "Hash table expansion starting\n");
        hashpower++;
        expanding = true;
        expand_bucket = 0;
        STATS_LOCK();
        stats_state.hash_power_level = hashpower;
        stats_state.hash_bytes += hashsize(hashpower) * sizeof(void *);
        stats_state.hash_is_expanding = true;
        STATS_UNLOCK();
    } else {
        primary_hashtable = old_hashtable;
        /* Bad news, but we can keep running. */
    }
}

void assoc_start_expand(uint64_t curr_items) {
    if (started_expanding)
        return;

    if (curr_items > (hashsize(hashpower) * 3) / 2 &&
          hashpower < HASHPOWER_MAX) {
        started_expanding = true;
        pthread_cond_signal(&maintenance_cond);
    }
}

int start_assoc_maintenance_thread() {
    int ret;
    char *env = getenv("MEMCACHED_HASH_BULK_MOVE");
    if (env != NULL) {
        hash_bulk_move = atoi(env);
        if (hash_bulk_move == 0) {
            hash_bulk_move = DEFAULT_HASH_BULK_MOVE;
        }
    }
    pthread_mutex_init(&maintenance_lock, NULL);
    if ((ret = pthread_create(&maintenance_tid, NULL,
                              assoc_maintenance_thread, NULL)) != 0) {
        fprintf(stderr, "Can't create thread: %s\n", strerror(ret));
        return -1;
    }
    return 0;
}

static void *assoc_maintenance_thread(void *arg) {

    mutex_lock(&maintenance_lock);
    while (do_run_maintenance_thread) {
        int ii = 0;

        /* There is only one expansion thread, so no need to global lock. */
        // 循环每次扩展的全部bucket
        for (ii = 0; ii < hash_bulk_move && expanding; ++ii) {
            item *it, *next;
            unsigned int bucket;
            void *item_lock = NULL;

            /* bucket = hv & hashmask(hashpower) =>the bucket of hash table
             * is the lowest N bits of the hv, and the bucket of item_locks is
             *  also the lowest M bits of hv, and N is greater than M.
             *  So we can process expanding with only one item_lock. cool! */
            /* expand_bucket需要锁保护,由于处于同一个bucket
             中的特性是这些item 的hv 的低N位是完全相同，对应的
             item_lock 的位置靠hv 的低M位确定，由于item_lock
             数组大小小于桶数组的大小，所以有 M < N ,也就是说处
             于同一个桶中的item拥有相同item_lock,所以在遍历桶中
             所有的item 的时候不需要在额外获取item_lock。这里的
             设计非常精妙~ */
            if ((item_lock = item_trylock(expand_bucket))) {
                    /* 遍历bucket 中全部item,插入到
                    primary_hashtable 中相应bucket */
                    for (it = old_hashtable[expand_bucket]; NULL != it; it = next) {
                        next = it->h_next;
                        bucket = hash(ITEM_key(it), it->nkey) & hashmask(hashpower);
                        it->h_next = primary_hashtable[bucket];
                        primary_hashtable[bucket] = it;
                    }
                    // old_hashtable 中bucket 内容设置为空
                    old_hashtable[expand_bucket] = NULL;
                    // 维护当前扩展的进度
                    expand_bucket++;
                    /* 如果扩展已经全部完成则设置expanding为
                    false ,释放old_hashtable 的内存*/
                    if (expand_bucket == hashsize(hashpower - 1)) {
                        expanding = false;
                        free(old_hashtable);
                        STATS_LOCK();
                        stats_state.hash_bytes -= hashsize(hashpower - 1) * sizeof(void *);
                        stats_state.hash_is_expanding = false;
                        STATS_UNLOCK();
                        if (settings.verbose > 1)
                            fprintf(stderr, "Hash table expansion done\n");
                    }

            } else {
                usleep(10*1000);
            }
            // 释放资源
            if (item_lock) {
                item_trylock_unlock(item_lock);
                item_lock = NULL;
            }
        }
        // 如果不在进行扩展，则设置条件变量，等待被触发扩展
        if (!expanding) {
            /* We are done expanding.. just wait for next invocation */
            started_expanding = false;
            pthread_cond_wait(&maintenance_cond, &maintenance_lock);
            /* assoc_expand() swaps out the hash table entirely, so we need
             * all threads to not hold any references related to the hash
             * table while this happens.
             * This is instead of a more complex, possibly slower algorithm to
             * allow dynamic hash table expansion without causing significant
             * wait times.
             */
            pause_threads(PAUSE_ALL_THREADS);
            assoc_expand();
            pause_threads(RESUME_ALL_THREADS);
        }
    }
    return NULL;
}

if (nthreads < 3) {
    power = 10;
} else if (nthreads < 4) {
    power = 11;
} else if (nthreads < 5) {
    power = 12;
} else if (nthreads <= 10) {
    power = 13;
} else if (nthreads <= 20) {
    power = 14;
} else {
    /* 32k buckets. just under the hashpower default. */
    power = 15;
}
/* 保证分段锁的数目小于hashtable 桶的个数，这样设计的好处之一
是在扩展的时候针对一个桶中的所有item 对应的是同一个
item_lock*/
if (power >= hashpower) {
    fprintf(stderr, "Hash table power size (%d) cannot be equal to or less than item lock table (%d)\n", hashpower, power);
    fprintf(stderr, "Item lock table grows with `-t N` (worker threadcount)\n");
    fprintf(stderr, "Hash table grows with `-o hashpower=N` \n");
    exit(1);
}

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);
}