libpmem

libpmem是一个非常底层的 C 语言库，负责处理与持久内存相关的 CPU 指令、以最佳方式将数据复制到持久内存以及文件映射。有关 libpmem 的更多信息，请访问 http://pmem.io/pmdk/libpmem/。
我们的示例使用的 C 代码在 Linux上构建与运行，使用libpmem把持久内存作为SSD存储的持久缓存页，需要考虑这样的特性：
1.一个4T的SSD设备可以分成1024*1024个block，每个block里面有1024个4KB 大小的页, 这些页面要保存在持久内存中并持久化。
2.要考虑每一个页的事务性，即在系统崩溃者突然断电的情况下不能出现脏页。
3.所有的持久内存内的数据可以恢复。
4.在这个例子中我们只考虑单线程的情况。
所有的页面需要保存在持久内存中间，需要考虑到以下几点：
1.数据的对齐，数据写到持久内存最好要64字节对齐，所以要考虑数据的布局。
2.block中的4K页不能使用原地更新，因为一旦发生了断电，更新可能被中断从而产生脏的数据。
3.block中的4K页可以通过分配器去进行管理，但是由于我们只有一个固定大小的数据需要管理，可以简单的通过meta data来管理。
4.数据的布局关系到数据的恢复，不能保存绝对的指针而必须是一个offset，以便在重启或崩溃后恢复数据。
示例 1（360 行）中使用的libpmem核心接口主要包括：
1.pmem_base = (char *)pmem_map_file(filename, PMEM_SIZE, PMEM_FILE_CREATE, 0666, &mapped_len, &is_pmem))
2.pmem_memset_persist(pmem_base,0x00,PMEM_SIZE);
3.pmem_memset_persist(pmem_meta,0x0,8); 原子操作
4.pmem_memcpy_persist(PAGE_FROM_META(page_new), content, 4096); 非原子操作
5.pmem_memcpy_persist(page_new,&atomic_value,8); 原子操作
6.pmem_persist(page_old,8); 原子操作

在持久内存的编程里面，个人认为以下这两点是持久内存编程的基础，所有的上层的库和应用都是基于这两个设定来保证数据的完整性和一致性。
1.pmem_memcpy_persist和pmem_memcpy_nodrain有什么区别怎么去理解？pmem_memcpy_persist可以保证数据已经写到了持久内存，pmem_memcpy_nodrain不能保证。像non-temporal write是一个乱序的写入，我们不能清楚哪些数据已经落到了持久内存，哪些还没有，而fence（drain）可以保证所有写的数据都落到持久内存中。ADR和eADR对这个部分的实现有影响，如果是ADR，在数据持久化的过程中有以下2个步骤
step1：数据写入CPU Cache （这个地方没有谈及NTW,NTW 不经过cache）
step2： 数据通过clflushopt/clflush/clwb等指令刷新到持久内存中，fense在clflushopt和clwb是必须的。
而eADR，由于整个cpu cache中的数据在突然断电的情况下会保存到持久内存，所以没有必要经过上述step2，只需要将数据写到CPU cache就可以了。
2.每个支持持久内存的平台都有一套具备原子性的原生内存操作。在英特尔硬件上，原子持久存储为 8 字节。因此，如果正在传输与持久内存对齐的 8 字节存储过程中程序或系统发生崩溃，在恢复时，这8字节包含的要么全是老内容，要么全是新内容。英特尔处理器拥有存储超过 8 字节的指令，但这些指令不具有故障原子性，所以发生电源故障等事件时可能会遭到破坏，变成一个脏的新旧数据的混合。
示例 1：libpmem实现持久页缓存

#include <iostream>
#include <chrono>
#include <stdio.h>
#include <stdint.h>
#include <stdlib.h>
#include <libpmem.h>
#include <time.h>
#include <string.h>

#define BLOCK_PAGE_NUM 1024
#define REQ_BLOCK 10
#define REQ_PAGE 0x3ff
#define DISK_BLOCK_NUM 1048576 //1024*1024 blocks 
#define MAGIC_NUMBER 0xabcdabcd
/****************************************
 * \
 * pmem layout as:
 ---8byte magic number---|--- 8byte page_meta[0]---|---8byte page_meta[1]---|...|---4k page[0]---|---4k page[1]---|
Page meta 和真正的4k页有对应关系。 
rebuild the block dram and freelist structure after the system restart. 
 */
//8 bytes atomic,必须一起更新。不能出现一个域更新了，而另外的域还没有更新的情况
typedef struct page_meta {
  uint64_t valid:1; //valid 表示该page是否已经分配和使用，如果是0，表示该page没有被使用
  uint64_t sn:31;  //sn来表示页的新旧，sn越大表示页越新，old page一般sn会被重置为0
  uint64_t req_id:32;  //req_id, which is in the range from 1~1024*1024*1024
}page_meta_t;
//magic number，可以不定义这个数据结构，因为位置是固定的
typedef struct pmem_layout {
  uint64_t magic_number;
  uint64_t page_offset;    //real page offset，保证这个offset 4K对齐。
}pmem_layout_t;
char * pmem_base; //持久内存的基地址
uint64_t page_address;  //真正的也得首地址。

#define PAGE_SIZE 4096
#define PMEM_META_SIZE sizeof(page_meta_t)
#define MAGIC_NUM_SIZE sizeof(pmem_layout_t)
#define PMEM_SIZE 100*1024*1024*1024UL  //定义我们使用持久内存的大小100GB
#define PMEM_PAGE_NUMBERS ((PMEM_SIZE-MAGIC_NUM_SIZE)/(PAGE_SIZE+PMEM_META_SIZE)-1) //根据持久内存的大小，我们可以知道我们有多少个内存页
#define PAGE_ID_FROM_META(meta) (((uint64_t)meta-MAGIC_NUM_SIZE-(uint64_t)pmem_base)/PMEM_META_SIZE+1) //根据持久内存的page meta而知道page_id
#define PAGE_FROM_META(meta) (char *)(page_address + PAGE_SIZE * (((uint64_t)meta - MAGIC_NUM_SIZE- (uint64_t)pmem_base)/PMEM_META_SIZE)) //meta->real page
#define PAGE_META_FROM_ID(id) (page_meta_t *) ((uint64_t)pmem_base+MAGIC_NUM_SIZE+(id-1)*PMEM_META_SIZE) //从pageid知道page meta

/*****************************************
 * *
 * disk structure,这个数据结构放在内存中，在初始化的时候扫描page meta数据，恢复disk内存数据结构
 * [[page_id1,page_id2]* BLOCK_PAGE_NUM],cache_pages_cnt]* DISK_BLOCK_NUM
 */
typedef struct block_page {
  uint64_t page_id1:32;
  uint64_t page_id2:32;
}block_page_t;
typedef struct block_data {
  block_page_t pages[BLOCK_PAGE_NUM];
  uint32_t cached_pages_cnt;  //if the count over one threshold, might need to write back to the SSD.   
}block_data_t;
typedef struct disk {
  block_data_t blocks[DISK_BLOCK_NUM];
}disk_t;
disk_t * disk;

//a free list that always pick the free page from the free_list[free_num-1];
//初始化的时候扫描page meta数据，恢复free pages的数据结构
typedef struct free_pages{
  uint64_t free_num;
  page_meta_t ** free_list;
}free_pages_t;
free_pages_t * free_pages;

//初始化，传入的是文件的名称，文件名称也可以定义在头文件中间。
int cbs_init(const char * filename)
{
  size_t mapped_len, pagecache_num;
  int is_pmem;
  int i;
  //将持久内存映射到虚拟地址空间并得到基地址pmem_base
  if ((pmem_base = (char *)pmem_map_file(filename, PMEM_SIZE,
                                            PMEM_FILE_CREATE, 0666,
                                            &mapped_len, &is_pmem)) == NULL) {
    perror("Pmem map file failed");
    return -1;
  }
  printf("pmem_map_file mapped_len=%ld, pmem_base=%p, is_pmem=%ld\n", mapped_len, pmem_base, is_pmem);
  pmem_layout_t * pmem_data =(pmem_layout_t *) pmem_base;
  page_meta_t * pmem_meta = (page_meta_t *)((uint64_t)pmem_base+sizeof(pmem_layout_t));
  
  //分配内存数据结构
  disk=(disk_t *)calloc(sizeof(char),sizeof(disk_t));
  if(disk == NULL) return -1;

  free_pages = (free_pages_t*)(malloc(sizeof(free_pages_t)));
  free_pages->free_list =(page_meta_t **) malloc(PMEM_PAGE_NUMBERS*sizeof(page_meta_t *));
  if(free_pages == NULL || free_pages->free_list == NULL) {
    printf("DRAM space is not enough for store the free_list\n");
    return -1;
  }
  
  //check magic number，如果magic number已经写过了，那可以恢复disk数据结构，否则所有的页都是free的，建立free pages数据结构
  if(pmem_data->magic_number != MAGIC_NUMBER) {
    //first time init and write the whole block structure to 0
    //pmem_memset_persist(pmem_base,0x00,PMEM_SIZE);
    pmem_memset_persist(pmem_meta,0x0,sizeof(page_meta_t)*PMEM_PAGE_NUMBERS);
    page_address = (uint64_t)(((uint64_t )pmem_meta + sizeof(page_meta_t)*PMEM_PAGE_NUMBERS + PAGE_SIZE) & 0xfffffffffffff000);
    pmem_data->page_offset = page_address - (uint64_t)pmem_base;
    pmem_persist(&(pmem_data-> page_offset),sizeof(pmem_data-> page_offset));
    
    pmem_data->magic_number = MAGIC_NUMBER;
    pmem_persist(pmem_data,sizeof(pmem_data->magic_number));
    
    for(i=0;i<PMEM_PAGE_NUMBERS;i++) {
      free_pages->free_list[i]=pmem_meta;
      pmem_meta+=1;
    }
    free_pages->free_num=PMEM_PAGE_NUMBERS;
  }
  else {
    int j=0;
    uint64_t block_id,page_id;
    uint64_t req_id;
    //magic number check pass, that means we might have free pages caches. 
    for(i=0;i<PMEM_PAGE_NUMBERS;i++) {
      if(pmem_meta->valid == 0) {
        free_pages->free_list[j]=pmem_meta;
        pmem_meta+=1;
        j++;
      } else {
        //fill the disk structure.
        req_id = pmem_meta->req_id;
        block_id = req_id >> 10;
        page_id = req_id & 0x3ff;
        if(disk->blocks[block_id].pages[page_id].page_id1 == 0) {
          disk->blocks[block_id].pages[page_id].page_id1 = PAGE_ID_FROM_META(pmem_meta);
          disk->blocks[block_id].cached_pages_cnt++;
        }else {
          disk->blocks[block_id].pages[page_id].page_id1 = PAGE_ID_FROM_META(pmem_meta);
        }
        pmem_meta +=1;
      }
    }
    free_pages->free_num=j;
    page_address=pmem_data->page_offset + (uint64_t) pmem_base;
  }
  printf("init done, pagecache_num=%d,free page number=%d\n",PMEM_PAGE_NUMBERS, free_pages->free_num);
  return 0;
}

int get_cached_count() 
{
  int i;
  int cnt=0;
  for(i=0;i<DISK_BLOCK_NUM;i++) {
    cnt+=disk->blocks[i].cached_pages_cnt;
  }
  return cnt;
}

//写入或者更新一个页，输入req_id表示更新到什么地方，content表示更新的内容
int write_req(uint64_t req_id, unsigned char * content) {
  //req_id>>10是block_id(1024 pages/block); req_id&0x3ff是该block中的page_id.
  uint64_t block_id = req_id >> REQ_BLOCK;
  uint64_t req_page_id = req_id & REQ_PAGE;
  uint64_t atomic_value=0;
  uint64_t page_id1, page_id2;
  uint64_t free_num=free_pages->free_num;

  if(block_id > DISK_BLOCK_NUM) {
    printf(" write req_id is not valid and over the disk range\n");
    return -1;
  }
  page_id1=disk->blocks[block_id].pages[req_page_id].page_id1;
  page_id2=disk->blocks[block_id].pages[req_page_id].page_id2;
  page_meta_t ** free_list = free_pages->free_list;
  if(free_num ==0 ) {
    printf("thre is no free pages in the pmem\n");
  }

  if(page_id1==0 && page_id2==0) { 
    //there is no page in the location, add one page; Get one free page meta in the free_list
    page_meta_t * page_new = free_list[free_num-1];
    pmem_memcpy_persist(PAGE_FROM_META(page_new), content, 4096);  //从pagemeta中拿到4k page，将数据写入并持久化
    atomic_value = req_id<<32|1;
    pmem_memcpy_persist(page_new,&atomic_value,8); //更新page meta valid=1，sn=0， req_id，atomic写入
    
    //更新内存中的数据结构，free number减一， block中page_id1更新，cached_page_cnt加一
    free_pages->free_num-=1;
    disk->blocks[block_id].pages[req_page_id].page_id1=PAGE_ID_FROM_META(page_new);
    disk->blocks[block_id].cached_pages_cnt++;
  } 
  else if((page_id1!=0 && page_id2==0)||(page_id1==0 && page_id2!=0)) {
    //已经有一个数据，此时需要更新，不能原地更新，必须先写道新的位置上
    page_meta_t * page_old;
    if(page_id1!=0) {
      page_old=PAGE_META_FROM_ID(page_id1);
    }else {
      page_old=PAGE_META_FROM_ID(page_id2);    
    }
    if(page_old->sn!=0) {
      page_old->sn=0;
      pmem_persist(page_old,8);  //老的页先把sn reset为0
    }
    //找到一个新的page meta和4kpage，先将4k数据写入，如果写入过程断电，由于page_meta不会更新，所以恢复不会出错。
    page_meta_t * page_new = free_list[free_num-1];
    pmem_memcpy_persist(PAGE_FROM_META(page_new), content, 4096);
    atomic_value = req_id<<32|1<<1|1;  //sn=1， req_id, valid=1, 原子写入
    pmem_memcpy_persist(page_new,&atomic_value, 8);
    disk->blocks[block_id].pages[req_page_id].page_id2=PAGE_ID_FROM_META(page_new);

    //如果此时断电，在这个req_id上有两个页都是有效的，其中sn=0的是老的数据，sn=1的是新的数据。
    //将老的数据free，将老的meta page_old写入0， sn=0；valid=0，req_id=0;    
    atomic_value=0;
    pmem_memcpy_persist(page_old,&atomic_value, 8);
    free_list[free_num-1]=page_old;
    disk->blocks[block_id].pages[req_page_id].page_id1=0;
  }  
  else if(page_id1!=0 && page_id2!=0) {  
    // 如果在这个req_id上两个页都是有效的，检查sn，sn越大的数据越新。
    page_meta_t * page1_meta, *page2_meta;    
    page1_meta=PAGE_META_FROM_ID(page_id1);
    page2_meta=PAGE_META_FROM_ID(page_id2);
    if(page1_meta->sn < page2_meta->sn) { //page_id2是新的数据
      //更新page_id1,让page_id1对应的数据更新
      pmem_memcpy_persist(PAGE_FROM_META(page1_meta), content, 4096);
      atomic_value = req_id<<32|(page2_meta->sn++)<<1|1;
      pmem_memcpy_persist(page1_meta,&atomic_value, 8);
      
      atomic_value=0; //释放page_id2对应得数据
      pmem_memcpy_persist(page2_meta,&atomic_value, 8);
      disk->blocks[block_id].pages[req_page_id].page_id2=0;
      free_num++;
      free_pages->free_num=free_num;
      free_list[free_num-1]=page2_meta;
    } else {
      pmem_memcpy_persist(PAGE_FROM_META(page2_meta), content, 4096);
      atomic_value = req_id<<32|(page1_meta->sn++)<<1|1;
      pmem_memcpy_persist(page2_meta,&atomic_value, 8);
      
      atomic_value=0;
      pmem_memcpy_persist(page2_meta,&atomic_value, 8);
      disk->blocks[block_id].pages[req_page_id].page_id1=0;
      free_num++;
      free_pages->free_num=free_num;
      free_list[free_num-1]=page1_meta;
    }
  }
  return 0;
}

void * read_req(uint64_t req_id) {
  // req_id and content; only after the init success, then this API can be called.
  uint64_t block_id = req_id >> REQ_BLOCK;
  uint64_t req_page_id = req_id & REQ_PAGE;
  uint64_t atomic_value=0;
  uint64_t page_id1, page_id2;
  if(block_id > DISK_BLOCK_NUM) {
    printf("read req_id is not valid and over the disk range\n");
    return NULL;
  }

  page_id1=disk->blocks[block_id].pages[req_page_id].page_id1;
  page_id2=disk->blocks[block_id].pages[req_page_id].page_id2;
  
  if((page_id1!=0 && page_id2 == 0) ||(page_id1==0 && page_id2!=0)) { // only page1 is there 
    page_meta_t * page_meta;
    if(page_id1!=0) {
      page_meta=PAGE_META_FROM_ID(page_id1);
    }else {
      page_meta=PAGE_META_FROM_ID(page_id2);
    }
    return PAGE_FROM_META(page_meta);
    
  } else if(page_id1!=0 && page_id2!=0) { // two pages 
    page_meta_t * page1_meta=PAGE_META_FROM_ID(page_id1);
    page_meta_t * page2_meta=PAGE_META_FROM_ID(page_id2);
    if(page1_meta->sn < page2_meta->sn) {
      return PAGE_FROM_META(page2_meta);
    } else {
      return PAGE_FROM_META(page1_meta);
    }
  } else {
    printf("cache is not exist\n");
  }

  return NULL;
}

void delete_page(uint64_t req_id) {
  // req_id and content; only after the init success, then this API can be called.
  uint64_t block_id = req_id >> REQ_BLOCK;
  uint64_t req_page_id = req_id & REQ_PAGE;
  uint64_t atomic_value=0;
  uint64_t page_id1, page_id2;
  if(block_id > DISK_BLOCK_NUM) {
    printf("read req_id is not valid and over the disk range\n");
    return ;
  }
  page_id1=disk->blocks[block_id].pages[req_page_id].page_id1;
  page_id2=disk->blocks[block_id].pages[req_page_id].page_id2;

  uint64_t free_num=free_pages->free_num;
  page_meta_t ** free_list = free_pages->free_list;
  if(free_num ==0 ) {
    printf("there is no free pages in the pmem\n");
  }
  atomic_value=0;
  if((page_id1!=0 && page_id2 == 0) ||(page_id1==0 && page_id2!=0)) { // only one page is there 
    page_meta_t * page_meta;
    if(page_id1!=0) {
      page_meta=PAGE_META_FROM_ID(page_id1);
      disk->blocks[block_id].pages[req_page_id].page_id1=0;
    }else {
      page_meta=PAGE_META_FROM_ID(page_id2);
      disk->blocks[block_id].pages[req_page_id].page_id2=0;
    }
    pmem_memcpy_persist(page_meta,&atomic_value, 8);
    free_num++;
    free_pages->free_num=free_num;
    free_list[free_num-1]=page_meta;
  } else if(page_id1!=0 && page_id2!=0) { // two pages 
    page_meta_t * page1_meta=PAGE_META_FROM_ID(page_id1);
    page_meta_t * page2_meta=PAGE_META_FROM_ID(page_id2);
    if(page1_meta->sn < page2_meta->sn) {
      pmem_memcpy_persist(page1_meta,&atomic_value, 8);
      pmem_memcpy_persist(page2_meta,&atomic_value, 8);
    } else {
      pmem_memcpy_persist(page2_meta,&atomic_value, 8);    
      pmem_memcpy_persist(page1_meta,&atomic_value, 8);
    }
    disk->blocks[block_id].pages[req_page_id].page_id1=0;
    disk->blocks[block_id].pages[req_page_id].page_id2=0;
    free_num++;
    free_list[free_num-1]=page1_meta;
    free_num++;
    free_list[free_num-1]=page2_meta;
    free_pages->free_num=free_num;
  } else {
    printf("cache is not exist\n");
  }
}

#define WRITE_COUNT 100000
#define OVERWRITE_COUNT 10000
int main() 
{
   // calculate the time
  unsigned char * page_content=(unsigned char *)malloc(PAGE_SIZE);
  uint64_t i=0;
  auto start=std::chrono::steady_clock::now();
  auto stop=std::chrono::steady_clock::now();
  std::chrono::duration<double> diff=stop-start;

  unsigned char * read_content;
  memset(page_content,0xab,PAGE_SIZE);
  
  start=std::chrono::steady_clock::now();
  cbs_init("/mnt/pmem0/cbs_file");
  stop=std::chrono::steady_clock::now();
  diff=stop-start;
  std::cout<<"cbs_init time "<<diff.count()<<std::endl; 
  std::cout<<"cached page count" << get_cached_count()<<std::endl;

  start = std::chrono::steady_clock::now();
  for(i=0;i<WRITE_COUNT;i++) {
    write_req(i,page_content);
  }
  stop=std::chrono::steady_clock::now();
  diff=stop-start;
  std::cout<<"write_req time "<<diff.count()/WRITE_COUNT<<std::endl;

  memset(page_content,0xcd,PAGE_SIZE);

  start = std::chrono::steady_clock::now();
  for(i=0;i<OVERWRITE_COUNT;i++) {
    write_req(i,page_content);
  }
  stop=std::chrono::steady_clock::now();
  diff=stop-start;
  std::cout<<"overwrite write_req update take time "<< diff.count()/OVERWRITE_COUNT<<std::endl;

  start = std::chrono::steady_clock::now(); 
  for(i=0;i<OVERWRITE_COUNT;i++) {
    read_content=(unsigned char *)read_req(i);
    memcpy(page_content,read_content,PAGE_SIZE);
  }
  stop=std::chrono::steady_clock::now();
  diff=stop-start;
  std::cout<<"overwrite read_req take time "<<diff.count()/OVERWRITE_COUNT<<std::endl;
  printf("the page should fill with paten 0xcd, 0x%x\n", page_content[0]);

  start = std::chrono::steady_clock::now();
  for(i=OVERWRITE_COUNT;i<WRITE_COUNT;i++) {
    read_content=(unsigned char *)read_req(i);
    memcpy(page_content,read_content,PAGE_SIZE);
  }
  stop=std::chrono::steady_clock::now();
  diff=stop-start;
  std::cout<<"overwrite->write count read_req take time "<<diff.count()/(WRITE_COUNT-OVERWRITE_COUNT)<<std::endl;
  printf("the page should fill with patern 0xab, 0x%x\n", page_content[0]);

  //start = std::chrono::steady_clock::now();
  //for(i=0;i<WRITE_COUNT;i++) {
  //  delete_page(i);
  //}
  //stop=std::chrono::steady_clock::now();
  //diff=stop-start;
  //std::cout<<"delete write count take time "<<diff.count()/WRITE_COUNT<<std::endl;
  return 0;
}

编译”g++ cbs_req.cpp -o cbs_req -lpmem -O2”,然后使用“taskset -c 2 ./cbs_req运行这段代码，我们看到写一个4KB页大概花费2.7us，恢复的过程大概时0.1s；测试的具体结果如下：

➜  ~ taskset -c 2 ./cbs_req_new
pmem_map_file mapped_len=107374182400, is_pmem=1
init done, pagecache_num=26163298,free page number=26163298
cbs_init time 0.201417
cached page count0
write_req time 4.26528e-06
overwrite write_req update take time 2.15705e-06
overwrite read_req take time 1.06379e-06
the page should fill with paten 0xcd, 0xcd
overwrite->write count read_req take time 1.07138e-06
the page should fill with patern 0xab, 0xab
➜  ~ vim cbs_req_new.cpp
➜  ~ taskset -c 2 ./cbs_req_new
pmem_map_file mapped_len=107374182400, is_pmem=1
init done, pagecache_num=26163298,free page number=26063298
cbs_init time 0.108116
cached page count100000
write_req time 2.19311e-06
overwrite write_req update take time 2.71467e-06
overwrite read_req take time 1.04566e-06
the page should fill with paten 0xcd, 0xcd
overwrite->write count read_req take time 1.06975e-06
the page should fill with patern 0xab, 0xab

程序员如果只想完全原始地访问持久内存并且无需库提供分配器或事务功能，那么可以将 libpmem 用作开发的基础，libpmem支持应用程序采用自定义内存管理和恢复逻辑来实现高性能的持久化应用。对于大多数程序员而言，libpmem 非常底层，使用它有一定的难度。
该程序中有一个性能优化点（由于我们第一代CLX+AEP，CLWB不能正确的工作，会导致在cacheline flush时，cache中的数据被刷出到持久域的时候同时被invalid，导致如果该数据再次被读就会cache miss，从而影响到数据读的性能），大家可以想一想可以如何优化？
在该程序中，page_meta的大小是8个字节，当一个page_meta写完后会invalid整个64字节，从而导致后面的page_meta读的时候产生cache miss。优化的方法就是：

typedef struct page_meta {
  uint64_t valid:1; //valid 表示该page是否已经分配和使用，如果是0，表示该page没有被使用
  uint64_t sn:31;   //sequence number来表示页的新旧，sequence越大表示页越新，old page一般sn会被重置为0
  uint64_t req_id:32;  //req_id, which is in the range from 1~1024*1024*1024
  uint64_t padding[7]; //for performance, avoid false sharing /read/modify/flush
}page_meta_t;
//magic number，可以不定义这个数据结构，因为位置是固定的
typedef struct pmem_layout {
  uint64_t magic_number;
  uint64_t page_offset;    //real page offset，保证这个page address 4K对齐。
  uint64_t padding[6];
}pmem_layout_t;

原先的page_meta只是8个字节，现在增加到了64字节，这样就会大大优化性能，此时性能能够做到从2.7us优化到1.7us （reduce latency 60%）:

➜  ~ taskset -c 2 ./cbs_req_new
pmem_map_file mapped_len=107374182400, is_pmem=1
init done, pagecache_num= 25811100,free page number=23588351
cbs_init time 5.33871
cached page count0
write_req time 4.46733e-06
overwrite write_req update take time 1.71743e-06
overwrite read_req take time 1.04520e-06
the page should fill with paten 0xcd, 0xcd
overwrite->write count read_req take time 1.06970e-06
the page should fill with patern 0xab, 0xab
➜  ~ taskset -c 2 ./cbs_req_new
pmem_map_file mapped_len=107374182400, is_pmem=1
init done, pagecache_num= 25811100,free page number=23488351
cbs_init time 0.847513
cached page count100000
write_req time 1.75316e-06
overwrite write_req update take time 1.82072e-06
overwrite read_req take time 1.04660e-06
the page should fill with paten 0xcd, 0xcd
overwrite->write count read_req take time 1.05974e-06
the page should fill with patern 0xab, 0xab

当你切换到ICX+BPS的平台后，CLWB可以正常工作，在cacheline被刷新后， 并不会从cache中invalid，所以cache的数据仍然可以访问而不会产生cache miss。在ICX+BPS这个优化不是必须的。下面的结果是在ICX+BPS下，不做任何padding之后的结果：写的结果类似（稍微慢一点，这个与测试的CPU的频率有关），但是读的性能比CLX+AEP有10%~20%的提升，具体的原因在这里我们不详细讨论，但是我们可以知道一点，CLWB在ICX+BPS下可以正常的工作。

[root@localhost ~]# taskset -c 2 ./cbs_req_new
pmem_map_file mapped_len=107374182400, is_pmem=1
init done, pagecache_num=26163298,free page number=26163298
cbs_init time 55.6604
cached page count0
write_req time 1.60401e-06
overwrite write_req update take time 1.80627e-06
overwrite read_req take time 1.07942e-06
the page should fill with paten 0xcd, 0xcd
overwrite->write count read_req take time 8.9152e-07
the page should fill with patern 0xab, 0xab
[root@localhost ~]# taskset -c 2 ./cbs_req_new
pmem_map_file mapped_len=107374182400, is_pmem=1
init done, pagecache_num=26163298,free page number=26063298
cbs_init time 0.164735
cached page count100000
write_req time 1.86702e-06
overwrite write_req update take time 2.00082e-06
overwrite read_req take time 8.18381e-07
the page should fill with paten 0xcd, 0xcd
overwrite->write count read_req take time 8.41381e-07
the page should fill with patern 0xab, 0xab

另一个问题是在这个程序中，当第一次创建和映射文件后，第一次写的性能需要4.59us，这个的主要原因什么？主要原因是，第一次写的时候，会产生缺页中断将虚拟地址和真实的物理页相连接并将该映射写入页表，缺页中断会导致清零的操作，代价较大。但一旦完成写入，虚拟地址和物理地址的连接已经建立，后面就不会产生缺页中断。所以，我们可以在第一次初始化的时候，通过将每个虚拟页的地址写入一个0，就可以完成缺页中断。

 //check magic number，如果magic number已经写过了，那可以恢复disk数据结构，否则所有的页都是free的，建立free pages数据结构
  if(pmem_data->magic_number != MAGIC_NUMBER) {
    //first time init and write the whole block structure to 0
    //pmem_memset_persist(pmem_base,0x00,PMEM_SIZE);
    
    pmem_memset_persist(pmem_meta,0x0,sizeof(page_meta_t)*PMEM_PAGE_NUMBERS);
    pmem_data->magic_number = MAGIC_NUMBER;
    pmem_persist(pmem_data,sizeof(pmem_layout_t));
    
    for(i=0;i<PMEM_PAGE_NUMBERS;i++) {
      free_pages->free_list[i]=pmem_meta;
      pmem_memset_persist(PAGE_FROM_META(pmem_meta),0x0,8);//pre-fault
      pmem_meta+=1;
    }
    free_pages->free_num=PMEM_PAGE_NUMBERS;
  }

虽然第一次初始化的时间从5.33s 增加到了43.46s，却可以降低第一次写入的时间到1.53us。

➜  ~ taskset -c 2 ./cbs_req_new
pmem_map_file mapped_len=107374182400, is_pmem=1
init done, pagecache_num= 25811100,free page number=23588351
cbs_init time 43.4666
cached page count0
write_req time 1.53175e-06
overwrite write_req update take time 1.69467e-06
overwrite read_req take time 1.04470e-06
the page should fill with paten 0xcd, 0xcd
overwrite->write count read_req take time 1.06070e-06
the page should fill with patern 0xab, 0xab
➜  ~ taskset -c 2 ./cbs_req_new
pmem_map_file mapped_len=107374182400, is_pmem=1
init done, pagecache_num= 25811100,free page number=23488351
cbs_init time 0.846013
cached page count100000
write_req time 1.75408e-06
overwrite write_req update take time 1.82161e-06
overwrite read_req take time 1.04456e-06
the page should fill with paten 0xcd, 0xcd
overwrite->write count read_req take time 1.06950e-06
the page should fill with patern 0xab, 0xab

可以使用工具pmemcheck来检查程序的正确性比如什么数据在持久内存中应该持久化去没有调用相应的函数，安装https://github.com/pmem/valgrind, 然后测试你的程序“valgrind --tool=pmemcheck ./your_program”，可以通过log来检查你的程序。由于pmemcheck是一个运行非常重的工具，所以对于应用的持久内存大小是有限制的。但是由于pmemcheck主要是来检测功能的，所以我们可以将我们的内存池 从100GB减少到10GB来进行验证测试。
#define PMEM_SIZE 1010241024*1024UL //定义我们使用持久内存的大小
然后可以运行这样的程序来检查：

➜  ~ valgrind --tool=pmemcheck ./cbs_req_new
==248561== pmemcheck-1.0, a simple persistent store checker
==248561== Copyright (c) 2014-2020, Intel Corporation
==248561== Using Valgrind-3.15.0 and LibVEX; rerun with -h for copyright info
==248561== Command: ./cbs_req_new
==248561==
pmem_map_file mapped_len=10737418240, is_pmem=1
init done, pagecache_num=2616328,free page number=2616328
cbs_init time 1.41468
cached page count0
write_req time 0.000361661
overwrite write_req update take time 0.000359447
overwrite read_req take time 5.06452e-08
the page should fill with paten 0xcd, 0xcd
overwrite->write count read_req take time 2.46624e-08
the page should fill with patern 0xab, 0xab
==248561==
==248561== Number of stores not made persistent: 0
==248561== ERROR SUMMARY: 0 errors
➜  ~ valgrind --tool=pmemcheck ./cbs_req_new
==248566== pmemcheck-1.0, a simple persistent store checker
==248566== Copyright (c) 2014-2020, Intel Corporation
==248566== Using Valgrind-3.15.0 and LibVEX; rerun with -h for copyright info
==248566== Command: ./cbs_req_new
==248566==
pmem_map_file mapped_len=10737418240, is_pmem=1
init done, pagecache_num=2616328,free page number=2516328
cbs_init time 0.0506498
cached page count100000
write_req time 0.000358157
overwrite write_req update take time 0.00035847
overwrite read_req take time 5.01243e-08
the page should fill with paten 0xcd, 0xcd
overwrite->write count read_req take time 2.46348e-08
the page should fill with patern 0xab, 0xab
==248566==
==248566== Number of stores not made persistent: 0
==248566== ERROR SUMMARY: 0 errors

参考资料:
1.http://pmem.io/pmdk/libpmem/;
2.https://github.com/pmem/pmdk/tree/master/src/examples/libpmem
3.https://github.com/twitter/pelikan/blob/master/src/datapool/datapool_pmem.c
4.如果要将一个block下的多个更新作为一个事务，使用libpmem需要考虑的比较复杂。比如需要有一个事务开始及完成的标识，如果事务没有完成，重启之后需要清理该block中的所有新的数据回到原有的数据。在事务开始之前，清理该block中所有有两份page 版本的数据，只保留一个版本。主要完成这样的函数int write_req_tx(uint64_t block_id, uint64_t *req_page_id, unsigned char ** content, int number) {}; 事务也是需要用到8字节的原子性来却保那些数据真正的完成了写入。
Tips:
1.保存offset而不是真正的地址，因为每次mmap()可能会映射到不一样的基地址。
2.使用8字节的原子性作为flag来保证所写入的完整性和一致性。
3.不要原地更新超过8字节的数据，而是使用sequency number（SN）来保证老的数据不会被覆盖，直到新的数据写成功才可以释放老的数据。
4.使用持久内存中的数据记录来恢复内存中的数据结构，这样可以减少编程的复杂性，同时可以提升性能。