PFS_PAGE_SIZE  // 每个page的大小, global_thread_container中默认为256
PFS_PAGE_COUNT // page的最大个数，global_thread_container中默认为256

class PFS_buffer_scalable_container {
  PFS_cacheline_atomic_size_t m_monotonic;            // 单调递增的原子变量，用于无锁选择page
  PFS_cacheline_atomic_size_t m_max_page_index;       // 当前已分配的最大page index
  size_t m_max_page_count;                            // 最大page个数，超过后将不再分配新page
  std::atomic<array_type *> m_pages[PFS_PAGE_COUNT];  // page数组
  native_mutex_t m_critical_section;                  // 创建新page时需要的一把锁
}
复制

value_type *allocate(pfs_dirty_state *dirty_state) {
  current_page_count = m_max_page_index.m_size_t.load();
  
  monotonic = m_monotonic.m_size_t.load();
  monotonic_max = monotonic + current_page_count;
  while (monotonic < monotonic_max) {
    index = monotonic % current_page_count;
    array = m_pages[index].load();
    pfs = array->allocate(dirty_state);
    if  (pfs) {
      // 分配成功返回
      return pfs;
    } else {
      // 分配失败，尝试下一个page， 
      // 因为m_monotonic是并发累加的，这里有可能本地monotonic变量并不是线性递增的，有可能是从1 直接变为 3或更大，
      // 所以当前while循环并不是严格轮训所有page，很大可能是跳着尝试，换者说这里并发访问下大家一起轮训所有的page。
      // 这个算法其实是有些问题的，会导致某些page被跳过忽略，从而加剧扩容新page的几率，后面会详细分析。
      monotonic = m_monotonic.m_size_t++;
    }
  }
  
  // 轮训所有Page后没有分配成功，如果没有达到上限的话，开始扩容page
  while (current_page_count < m_max_page_count) {
    // 因为是并发访问，为了避免同时去创建新page，这里有一个把同步锁，也是整个PFS内存分配唯一的锁
    native_mutex_lock(&m_critical_section);
    // 拿锁成功，如果array已经不为null，说明已经被其它线程创建成功
    array = m_pages[current_page_count].load();
    if (array == nullptr) {
      // 抢到了创建page的责任
      m_allocator->alloc_array(array);
      m_pages[current_page_count].store(array);
      ++m_max_page_index.m_size_t;
    }
    native_mutex_unlock(&m_critical_section);
    
    // 在新的page中再次尝试分配
    pfs = array->allocate(dirty_state);
    if (pfs) {
      // 分配成功并返回
      return pfs;
    }
    // 分配失败，继续尝试创建新的page直到上限
  }
}
复制

class PFS_buffer_default_array {
PFS_cacheline_atomic_size_t m_monotonic;      // 单调递增原子变量，用来选择free的record
size_t m_max;                                 // record的最大个数
T *m_ptr;                                     // record对应的PFS对象，比如PFS_thread
}
复制

value_type *allocate(pfs_dirty_state *dirty_state) {
  // 从m_monotonic记录的位置开始尝试轮序查找
  monotonic = m_monotonic.m_size_t++;
  monotonic_max = monotonic + m_max;

  while (monotonic < monotonic_max) {
    index = monotonic % m_max;
    pfs = m_ptr + index;
  
    // m_lock是pfs_lock结构，free/dirty/allocated三状态是由这个数据结构来维护的
    // 后面会详细介绍它如何实现原子状态迁移的
    if (pfs->m_lock.free_to_dirty(dirty_state)) {
      return pfs;
    }
    // 当前record不为free,原子变量++尝试下一个
    monotonic = m_monotonic.m_size_t++;
  }
}
复制

// 下面3个宏主要就是用来位操作的，方便操作state或version
#define VERSION_MASK 0xFFFFFFFC
#define STATE_MASK 0x00000003
#define VERSION_INC 4

bool free_to_dirty(pfs_dirty_state *copy_ptr) {
  uint32 old_val = m_version_state.load();

  // 判断当前state是否为FREE，如果不是，直接返回失败
  if ((old_val & STATE_MASK) != PFS_LOCK_FREE) {
    return false;
  }

  uint32 new_val = (old_val & VERSION_MASK) + PFS_LOCK_DIRTY;

  // 当前state为free，尝试将state修改为dirty，atomic_compare_exchange_strong属于乐观锁，多个线程可能同时
  // 修改该原子变量，但只有1个修改成功。
  bool pass =
      atomic_compare_exchange_strong(&m_version_state, &old_val, new_val);

  if (pass) {
    // free to dirty 成功
    copy_ptr->m_version_state = new_val;
  }

  return pass;
}

void dirty_to_allocated(const pfs_dirty_state *copy) {
  /* Make sure the record was DIRTY. */
  assert((copy->m_version_state & STATE_MASK) == PFS_LOCK_DIRTY);
  /* Increment the version, set the ALLOCATED state */
  uint32 new_val = (copy->m_version_state & VERSION_MASK) + VERSION_INC +
                   PFS_LOCK_ALLOCATED;

  m_version_state.store(new_val);
}
复制

struct pfs_lock {
  void allocated_to_free(void) {
    /*
      If this record is not in the ALLOCATED state and the caller is trying
      to free it, this is a bug: the caller is confused,
      and potentially damaging data owned by another thread or object.
    */
    uint32 copy = copy_version_state();
    /* Make sure the record was ALLOCATED. */
    assert(((copy & STATE_MASK) == PFS_LOCK_ALLOCATED));
    /* Keep the same version, set the FREE state */
    uint32 new_val = (copy & VERSION_MASK) + PFS_LOCK_FREE;

    m_version_state.store(new_val);
  }
}
复制

  while (monotonic < monotonic_max) {
    index = monotonic % current_page_count;
    array = m_pages[index].load();
    pfs = array->allocate(dirty_state);
    if  (pfs) {
       // 记录分配成功的index
       m_monotonic.m_size_t.store(index);
      return pfs;
    } else {
      // 局部变量递增，避免掉并发累加而跳过某些pages
      monotonic++;
    }
  }
复制