Memory allocator

实验目标：优化内存分配器的并行性能，通过将单全局锁+单空闲链表的设计改为每CPU锁+每CPU空闲链表，减少多核环境下的锁竞争。

当前的瓶颈：

1
2
3
4
struct {
  struct spinlock lock;
  struct run *freelist;
} kmem;

所有CPU共享同一个空闲链表和一个锁， 任何kalloc/kfree操作都需要竞争这个全局锁

解决方案：每个CPU操作自己的链表和锁

打开kalloc.c,先修改数据结构

1
2
3
4
struct {
  struct spinlock lock;
  struct run *freelist;
} kmem[NCPU];

初始化每个CPU锁

1
2
3
4
5
6
7
8
9
void
kinit()
{
  // 为每个CPU初始化锁
  for(int i = 0; i < NCPU; i++) {
    initlock(&kmem[i].lock, "kmem");
  }
  freerange(end, (void*)PHYSTOP);
}

修改kalloc

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
void *
kalloc(void)
{
  struct run *r;
  int id;

  // 安全获取当前CPU ID
  push_off();
  id = cpuid();
  pop_off();

  // 先尝试从当前CPU分配
  acquire(&kmem[id].lock);
  r = kmem[id].freelist;
  if(r) {
    kmem[id].freelist = r->next;
    release(&kmem[id].lock);
    
    if(r)
      memset((char*)r, 5, PGSIZE); // fill with junk
    return (void*)r;
  }
  release(&kmem[id].lock);

  // 当前CPU链表空，从其他CPU偷取
  for(int i = 0; i < NCPU; i++) {
    if(i == id) continue;  // 跳过自己
    
    acquire(&kmem[i].lock);
    r = kmem[i].freelist;
    if(r) {
      kmem[i].freelist = r->next;
      release(&kmem[i].lock);
      
      if(r)
        memset((char*)r, 5, PGSIZE);
      return (void*)r;
    }
    release(&kmem[i].lock);
  }

  return 0;
}

修改kfree

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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;

   // 安全获取当前CPU ID
  push_off();
  int id = cpuid();
  pop_off();

  // 使用当前CPU的锁，而不是全局锁
  acquire(&kmem[id].lock);
  r->next = kmem[id].freelist;
  kmem[id].freelist = r;
  release(&kmem[id].lock);
}

测试通过

Read-write lock

本实验要实现读写自旋锁，以提高多核下的读取并发性能。主要要实现读模式：允许多个读者同时持有锁，同时禁止写者进入；写模式：只允许一个写者持有锁，禁止其他读者和写者进入。

实验思路：在读写锁结构体中引入一把标准的自旋锁（lk）来严格保护 readers（读者数）、writer（写锁持有位）和 pending_writers（写者排队数）这三个状态变量；所有读写操作在检查或修改这些状态前必须先持有 lk。同时根据题目采用写者优先，写者在尝试获取锁前先增加 pending_writers，读者一旦发现有等待的写者便主动避让。

下面开始写代码：

首先在spinlock.h里修改

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
#ifdef LAB_LOCK
// Reader-writer lock.
struct rwspinlock {
  // 自旋锁，用来保护下面的 3 个计数器
  struct spinlock lk; 
  
  int readers;           // 读者数量
  int writer;            // 写者是否持有 (0 或 1)
  int pending_writers;   // 写者排队计数 (实现写者优先)

  // 调试信息
  char *name;
  struct cpu *cpu;
};
#endif

然后修改spinlock.c

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
static void
read_acquire_inner(struct rwspinlock *rwlk)
{
  while(1) {
    acquire(&rwlk->lk); // 1. 关门 (持有锁)

    // 2. 检查：是否有写者持有？是否有写者排队？
    if (rwlk->writer == 0 && rwlk->pending_writers == 0) {
      rwlk->readers++;    // 进场
      release(&rwlk->lk); // 开门 (放锁)
      return;
    }

    // 3. 条件不满足，开门，并在外面自旋重试
    release(&rwlk->lk);
  }
}

static void
read_release_inner(struct rwspinlock *rwlk)
{
  acquire(&rwlk->lk); // 关门
  rwlk->readers--;    // 离场
  release(&rwlk->lk); // 开门
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
static void
write_acquire_inner(struct rwspinlock *rwlk)
{
  // 1. 举手排队 (必须加锁修改)
  acquire(&rwlk->lk);
  rwlk->pending_writers++;
  release(&rwlk->lk);

  while(1) {
    acquire(&rwlk->lk); // 关门

    // 2. 检查：必须完全空闲 (无读 且 无写)
    if (rwlk->readers == 0 && rwlk->writer == 0) {
      rwlk->writer = 1;        // 抢占
      rwlk->pending_writers--; // 取消排队
      
      rwlk->cpu = mycpu();
      release(&rwlk->lk);      // 开门
      return;
    }

    release(&rwlk->lk); // 条件不满足，开门重试
  }
}

static void
write_release_inner(struct rwspinlock *rwlk)
{
  acquire(&rwlk->lk); // 关门
  
  rwlk->cpu = 0;
  rwlk->writer = 0;   // 释放写状态

  release(&rwlk->lk); // 开门
}

1
2
3
4
5
6
7
8
9
10
11
void
initrwlock(struct rwspinlock *rwlk)
{
  rwlk->name = "rwlock";
  // 必须初始化内部的锁
  initlock(&rwlk->lk, "rwlk_inner"); 
  rwlk->readers = 0;
  rwlk->writer = 0;
  rwlk->pending_writers = 0;
  rwlk->cpu = 0;
}

本以为修改完成，但是结果却是这样

1
2
3
4
5
rwspinlock_test(0): 0
rwspinlock_test(2): -1
rwspinlock_test(1): 0
rwspinlock_test(3): -1
rwlktest: 2/4 CPUs succeeded

找了很久的原因，直到我把delay函数里的10000改为100000，才能通过

1
2
3
4
5
6
7
8
9
static uint
delay()
{
  static uint v;
  for (int i = 0; i < 100000; i++) {
    __atomic_fetch_add(&v, 1, __ATOMIC_RELAXED);
  }
  return __atomic_load_n(&v, __ATOMIC_RELAXED);
}

ai的解释：原有的 delay 时间过短（仅微秒级），无法覆盖宿主机几十毫秒的调度间隙，导致“读者”在“写者”被冻结期间误判为无人排队而抢先进入。增加 delay 的本质是用软件层面的等待时间换取物理层面的调度容错，强迫读者空转更久，从而保证即使写者被宿主机暂时挂起，也能在读者结束等待前“复活”并完成排队登记，消除了因双重调度延迟导致的逻辑误判。

测试通过

1
2
3
4
5
rwspinlock_test(0): 0
rwspinlock_test(1): 0
rwspinlock_test(3): 0
rwspinlock_test(2): 0
rwlktest: 4/4 CPUs succeeded