博客

XV6 Lab7：Locks

  • Post author:
  • Post category:其他




Lab7：Locks


本 lab 的任务是优化 xv6 锁，以减少锁竞争。


我们将会重新设计代码以降低锁竞争，提高多核机器上系统的并行性，需要修改数据结构和锁的策略。


详细要求及提示见链接

：


https://pdos.csail.mit.edu/6.1810/2021/labs/lock.html


参考文章：



MIT6.S081 2021 locks



[mit6.s081] 笔记 Lab8: Locks | 锁优化


阅读指路：


• xv6book Chapter 6: “Locking” and the corresponding code.

• xv6book Section 3.5: “Code: Physical memory allocator”

• xv6book Section 8.1 through 8.3: “Overview”, “Buffer cache layer”, and “Code: Buffer cache”

•

kernel/kalloc.c


•

kernel/bio.c



Memory allocator (moderate)


在未修改前，所有内存块由一个锁管理，若有多个进程并发地获取内存，则会造成非常多的锁等待。

用户程序 /kalloctest 给xv6的内存分配器加压：三个进程增加和缩小它们的地址空间，导致对kalloc和kfree的多次调用，形成锁的竞争。 

$ kalloctest
start test1
test1 results:
--- lock kmem/bcache stats
lock: kmem: #fetch-and-add 83375 #acquire() 433015
lock: bcache: #fetch-and-add 0 #acquire() 1260
--- top 5 contended locks:
lock: kmem: #fetch-and-add 83375 #acquire() 433015
lock: proc: #fetch-and-add 23737 #acquire() 130718
lock: virtio_disk: #fetch-and-add 11159 #acquire() 114
lock: proc: #fetch-and-add 5937 #acquire() 130786
lock: proc: #fetch-and-add 4080 #acquire() 130786
tot= 83375
test1 FAIL



目标：

kalloctest中锁争用的

根本原因是kalloc()只有一个空闲列表，由一个锁保护

。

要消除锁争用，必须重新设计内存分配器，以避免单一的锁和列表。

基本思想是为每个CPU维护一个空闲列表，每个列表都有自己的锁。不同CPU上的分配和释放可以并行运行，因为每个CPU将在不同的列表上运行。

主要的挑战将是处理这样的情况：一个CPU的空闲列表是空的，但另一个CPU的列表有空闲内存；在这种情况下，一个CPU必须“窃取”另一个CPU的空闲列表的一部分。这种行为可能会引发锁竞争，但希望这种情况不会经常发生。



Hints

  • 可以使用常量

    NCPU

    （

    kernel/param.h

    ）

  • freerange()

    函数将所有空闲资源分配给正在运行freerange的CPU

  • cpuid

    函数返回当前的CPU编号，但是在调用前必须关闭中断，用函数

    push_off()

    和

    pop_off()

    实现中断的开关
  • PS：

    snprintf()

    函数 （

    kernel/sprintf.c

    ）可以给出锁的统一格式的命名



主要工作：


在 kernel/kalloc.c 中实现：
 

struct {
  struct spinlock lock;
  struct run *freelist;
} kmem[NCPU]; // 【每个CPU分配一个kmem锁】

#define MAX_NUM_PAGES 100	// 【最大“偷窃”页数设为100】

在knit中对每个锁初始化： 

void
kinit()
{
  char name[10];
  for (int i = 0; i < NCPU; i++) {
    snprintf(name, 10, "kmem-%d", i);
    initlock(&kmem[i].lock, name);
  }
  freerange(end, (void*)PHYSTOP);
}

// Let freerange give all free memory to the CPU running freerange.【无需修改】
void
freerange(void *pa_start, void *pa_end)
{
  char *p;
  p = (char*)PGROUNDUP((uint64)pa_start);
  for(; p + PGSIZE <= (char*)pa_end; p += PGSIZE)
    kfree(p);
}

// Free the page of physical memory pointed at by v,
// which normally should have been returned by a
// call to kalloc().  (The exception is when
// initializing the allocator; see kinit above.)
void
kfree(void *pa)
{
  struct run *r;

  if(((uint64)pa % PGSIZE) != 0 || (char*)pa < end || (uint64)pa >= PHYSTOP)
    panic("kfree");

  // Fill with junk to catch dangling refs.
  memset(pa, 1, PGSIZE);

  r = (struct run*)pa;

/*My Code:*/
  push_off();
  int cpu = cpuid();
  pop_off();
  // 空闲页加入对应CPU的空闲页链表
  acquire(&kmem[cpu].lock);
  r->next = kmem[cpu].freelist;
  kmem[cpu].freelist = r;	// 维护链表（从头部插入）
  release(&kmem[cpu].lock);
}

kalloc函数（分配一个物理页），对每个CPU的空闲内存单独管理，没有空闲内存时到其他CPU“偷窃” 

// Allocate one 4096-byte page of physical memory.
// Returns a pointer that the kernel can use.
// Returns 0 if the memory cannot be allocated.
void *
kalloc(void)
{
  struct run *r;

  // 原代码：
  // acquire(&kmem.lock);
  // r = kmem.freelist;
  // if(r)
  //   kmem.freelist = r->next;
  // release(&kmem.lock);
  
  push_off();
  int cpu = cpuid();
  pop_off();

  acquire(&kmem[cpu].lock);
  r = kmem[cpu].freelist;

  if(r)
    kmem[cpu].freelist = r->next;
  release(&kmem[cpu].lock);
  if(r == 0)
    r = get_others_pages(cpu);  // 【如果当前CPU的空闲页链表为空，到其他CPU的空闲页中取】
  
  if(r)
    memset((char*)r, 5, PGSIZE); // fill with junk
  return (void*)r;
}

偷窃其他CPU的空闲内存： 

void * get_others_pages(int cpu){
  int count = 0;
  struct run *start = 0;
  struct run *end = 0;
  for(int i = 0; i < NCPU; i++){
    if(i == cpu)
      continue;
    
    acquire(&kmem[i].lock);
    
    start = kmem[i].freelist;
    end = kmem[i].freelist;
    if(!start){	// 这个CPU也没有空闲内存
      release(&kmem[i].lock);
      continue;
    }
   	// 链表向后，直到结束或者达到100页
    while(end && count < MAX_NUM_PAGES){
      end = end->next;
      count++;
    }
    if(end){ 
      kmem[i].freelist = end->next;	// 后面还有空闲内存，freelist接在后面
      end->next = 0;
    }
    else kmem[i].freelist = 0;
    release(&kmem[i].lock);

    acquire(&kmem[cpu].lock);
    kmem[cpu].freelist = start->next;	// cpu的空闲内存freelist从start开始
    release(&kmem[cpu].lock);
    break;
    
  }
  return (void*) start;
}



Buffer cache (hard)

多个进程同时使用文件系统的时候，bcache.lock 上会发生严重的锁竞争。

bcache.lock

锁用于保护磁盘区块缓存，在原本的设计中，由于该锁的存在，多个进程不能同时操作（申请、释放）磁盘缓存。

你的目标是，修改cache管理策略，降低锁冲突。


We suggest you look up block numbers in the cache with a hash table that has a lock per hash bucket.

  • 可以用

    固定数量的buckets

    ，并且哈希表不需要动态变化，用一个质数（例如13）可以减少hash值的冲突

  • 在哈希表中查找一个buffer，以及为新的buffer分配一个entry的操作

    必须是原子的


  • Remove the list of all buffers

    (bcache.head etc.) and instead


    time-stamp buffers using the time of their last use

    (i.e., using

    ticks in kernel/trap.c). With this change

    brelse doesn’t need to

    acquire the bcache lock

    , and

    bget can select the least-recently used

    block based on the time-stamps

    .

  • 在

    bget()

    函数中

    serialize eviction 连续驱逐

    是可行的 (i.e., the part of bget that


    selects a buffer to re-use when a lookup misses in the cache

    ).

  • 有时候需要同时持有两个锁，例如在eviction过程中需要持有bcache lock和一个bucket lock，这种情况下注意

    避免死锁

    。

  • 当转移一个block（将一个buffer从一个bucket转移到另一个bucket时），需要注意处理block仍然hash到相同的bucket的情况，需要

    避免死锁

    。

  • xv6 book Chaper8 部分内容：

在这里插入图片描述

Buffer cache缓存磁盘块，并同步访问它们，确保一个块只能同时被内核中的一个进程访问。

8.2 Buffer cache layer

buffer缓存有两项工作：

(1) 同步访问磁盘块，以确保磁盘块在内存中只有一个buffer缓存，并且一次只有一个内核线程能使用该buffer缓存；

(2) 缓存使用较多的块，这样它们就不需要从慢速磁盘中重新读取（代码见bio.c）

buffer缓存的主要接口包括bread和bwrite，bread返回一个在内存中可以读取和修改的块副本buf，bwrite将修改后的buffer写到磁盘上相应的块。

内核线程在使用完一个buffer后，必须通过调用brelse释放它。


buffer缓存为每个buffer的都设有sleeplock（在struct buf内定义），以确保每次只有一个线程使用buffer（从而使用相应的磁盘块）

；bread 返回的buffer会被锁定，而brelse释放锁。

buffer缓存

有固定数量的buffer来存放磁盘块

，这意味着如果文件系统需要一个尚未被缓存的块，buffer缓存必须回收一个当前存放其他块的buffer。buffer缓存为新块寻找最近使用最少的buffer【

LRU替换策略

】，因为最近使用最少的buffer是最不可能被再次使用的buffer。



主要工作：

  • 在

    kernel/buf.h

    中完成： 
struct buf {
  int valid;   // has data been read from disk?
  int disk;    // does disk "own" buf?
  uint dev;
  uint blockno;
  struct sleeplock lock;
  uint refcnt;
  struct buf *prev; // LRU cache list
  struct buf *next;
  uchar data[BSIZE];

  uint lastuse_time;  // 【上次使用时间的时间戳】
  int owner;	// 【当前归属的bucket】
};
  • 在

    kernel/bio.c

    中完成：

首先定义

NBUCKET

表示管理的bucket个数，并且定义hash映射（磁盘块号blockno到bucket号） 

#define NBUCKET 13
#define BUCKET_HASH(dev, blockno) (blockno % NBUCKET)

bcache结构体中加入

bucket

（buffer cache的管理单位，将一堆buf数组组织在一起成为bucket）

以及

bucket_lock

作为每个bucket的互斥锁 

struct {
  struct buf buf[NBUF];
  struct spinlock lock;

  struct buf bucket[NBUCKET];
  struct spinlock bucket_locks[NBUCKET];
} bcache;

binit 函数：初始化所有 bucket 以及 buffer 

void
binit(void)
{
  initlock(&bcache.lock, "bcache_lock");

// 【初始化每个bucket】
  char name[20];
  for(int i = 0; i < NBUCKET; i++) {
    snprintf(name, 20, "bucket_lock_%d", i);
    initlock(&bcache.bucket_locks[i], name);
    bcache.bucket[i].next = 0;
  }
//【初始化每个buffer】
  for(int i = 0; i < NBUF; i++){
    struct buf *b = &bcache.buf[i];
    initsleeplock(&b->lock, "buffer");
    b->lastuse_time = 0;
    b->refcnt = 0;
    // 【初始：将所有buffer装到bucket[0]】
    b->next = bcache.bucket[0].next;
    bcache.bucket[0].next = b;
    b->owner = 0;
  }
}

bget 函数

（给定磁盘块，在buffer cache上分配一个buffer进行缓存）
 

static struct buf*
bget(uint dev, uint blockno)
{

  uint index = BUCKET_HASH(dev, blockno);	// 【找到归属于哪个bucket管理】

  acquire(&bcache.bucket_locks[index]); // 获取对应bucket的锁

  // Is the block already cached?
  // 【在对应的bucket里面找，是否已经缓存了这个块】
  struct buf *b = bcache.bucket[index].next;
  while(b){
    if(b->dev == dev && b->blockno == blockno){
      b->refcnt++;
      release(&bcache.bucket_locks[index]);
      acquiresleep(&b->lock);
      return b;
    }
    b = b->next;
  }

  // Not cached.没有被缓存
  // 我们需要从所有buckets中, 用最近最久未使用策略(LRU), 寻找合适的buffer来缓存这个块,
  // 这表明我们需要获取它们对应的bucket_lock.
  
  // 【占有一个bucket_lock时, 尝试再获取其他bucket锁是很不安全的, 很容易循环等待造成死锁, 
  // 所以我们首先释放当前获取的bucket锁】
  release(&bcache.bucket_locks[index]);

  // 【获取大锁lock】
  acquire(&bcache.lock);

  // 【获取大锁后, 再检查一遍是否已缓存(可能有其他进程所做的操作)
  // 防止出现同一个块缓存在多个buffer的问题】
  b = bcache.bucket[index].next;
  while(b){
    if(b->dev == dev && b->blockno == blockno){
      acquire(&bcache.bucket_locks[index]);
      b->refcnt++;
      release(&bcache.bucket_locks[index]);
      release(&bcache.lock);
      acquiresleep(&b->lock);
      return b;
    }
    b = b->next;
  }

  // Not Cached.仍然没有被缓存
  // 【此时需要采用LRU机制选择对应的buffer】:
  struct buf *before_ans = 0;
  uint min_lastuse_time = 0xffffffff;
  uint owner = -1;
  for(int i = 0; i < NBUCKET; i++){
    acquire(&bcache.bucket_locks[i]); // 获取bucket_lock[i]
    int flag = 0; // 标记
    b = &bcache.bucket[i];
    while(b->next) {
      if(b->next->refcnt == 0 && before_ans == 0){	
      //【还没找到一个buffer(before_ans==0), 此时碰到一个空闲buffer, 暂时作为before_ans】
        before_ans = b;
        min_lastuse_time = b->next->lastuse_time;
        flag = 1;
      }
      else if(b->next->refcnt == 0 && b->next->lastuse_time < min_lastuse_time) {
      //【已经找到buffer, 但是碰到空闲buffer, 满足lastuse_time更小, 应该用这个】
        before_ans = b;
        min_lastuse_time = b->next->lastuse_time;
        flag = 1;
      }
      b = b->next;
    }
    if(flag) {  // 【bucket[i]找到新的buffer】
      if(owner != -1)	// 有暂时的owner 
        release(&bcache.bucket_locks[owner]);	// 【释放之前的owner bucket的锁】
        //【要注意一直持有的是目前最新的owner的锁！！！】
      owner = i;	// 设置新的owner
    } else {  // 【bucket[i]没有找到新的buffer: 释放buck_lock[i] (记得在循环一开始先获得了锁) 】
      release(&bcache.bucket_locks[i]);
    }
  }
  if(before_ans == 0) {	// 最终都没有找到一个buffer
    panic("bget: No buffer.");
  }

  struct buf *ans = before_ans->next;
  
  if(owner != index) { // 【找到的buffer不在当前的bucket[index]】
    // buffer从原来的bucket[owner]移除:
    before_ans->next = ans->next;
    release(&bcache.bucket_locks[owner]);

    // buffer添加到bucket[index]:
    acquire(&bcache.bucket_locks[index]);
    ans->next = bcache.bucket[index].next;
    bcache.bucket[index].next = ans;
  }

  // buffer重置:
  ans->dev = dev;
  ans->blockno = blockno;
  ans->refcnt = 1;
  ans->valid = 0;
  ans->owner = index;

  release(&bcache.bucket_locks[index]);	// 释放bucket[index]锁
  //【无论owner等不等于index，执行到此处都只持有bucket[index]的锁】
  release(&bcache.lock);	// 释放大锁
  acquiresleep(&ans->lock);
  return ans;
}

brelse 函数： 

// Release a locked buffer.
void
brelse(struct buf *b)
{
// 【必须持有大锁lock】
  if(!holdingsleep(&b->lock))
    panic("brelse");
// 唤醒 buffer lock
  releasesleep(&b->lock);

  uint index = BUCKET_HASH(b->dev, b->blockno);	// 找到block归属于哪个bucket
  acquire(&bcache.bucket_locks[index]);
  b->refcnt--;
  if (b->refcnt == 0) {	//没有进程引用这块buffer, 则为空闲状态, lastuse_time设为当前时钟
    b->lastuse_time = ticks;
  }
  release(&bcache.bucket_locks[index]);
}

最后加入两个维护block的refcnt的函数： 

void
bpin(struct buf *b) {
  
  uint index = BUCKET_HASH(b->dev, b->blockno);
  acquire(&bcache.bucket_locks[index]);
  b->refcnt++;
  release(&bcache.bucket_locks[index]);
}

void
bunpin(struct buf *b) {

  uint index = BUCKET_HASH(b->dev, b->blockno);
  acquire(&bcache.bucket_locks[index]);
  b->refcnt--;
  release(&bcache.bucket_locks[index]);
}


版权声明：本文为weixin_53215555原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。