Memory allocator (moderate)
要求
程序 user/kalloctest 强调 xv6 的内存分配器：三个进程增长和缩小它们的地址空间，导致对 kalloc 和 kfree 的许多调用。kalloc 和 kfree 获取kmem.lock 。Kalloctest 打印（作为“#test-and-set”）在acquire中由于尝试获取另一个内核已经持有的锁（用于 kmem 锁和其他一些锁）而循环迭代的次数。获取中的循环迭代次数是锁争用的粗略度量。kalloctest 的输出如下所示：
$ kalloctest
start test1
test1 results:
--- lock kmem/bcache stats
lock: kmem: #test-and-set 83375 #acquire() 433015
lock: bcache: #test-and-set 0 #acquire() 1260
--- top 5 contended locks:
lock: kmem: #test-and-set 83375 #acquire() 433015
lock: proc: #test-and-set 23737 #acquire() 130718
lock: virtio_disk: #test-and-set 11159 #acquire() 114
lock: proc: #test-and-set 5937 #acquire() 130786
lock: proc: #test-and-set 4080 #acquire() 130786
tot= 83375
test1 FAIL
start test2
total free number of pages: 32497 (out of 32768)
.....
test2 OK
start test3
child done 1
child done 100000
test3 OK
start test2
total free number of pages: 32497 (out of 32768)
.....
test2 OK
start test3
child done 1
child done 100000
test3 OK您可能会看到与此处显示的计数不同，并且前 5 个争用锁的顺序也不同。
acquire 为每个锁维护要获取该锁的调用计数，以及获取中的循环尝试但未能设置锁的次数。Kalloctest 调用一个系统调用，使内核打印 KMEM 和 bcache 锁（这是本练习的重点）以及 5 个最有争议的锁的计数。如果存在锁争用，则acquire循环迭代的次数将很大。系统调用返回 kmem 和 bcache 锁的循环迭代次数之和。
对于此实验，必须使用具有多核计算机。如果您使用正在执行其他操作的机器，则kalloctest打印的计数将是无稽之谈。您可以使用专用的 Athena 工作站或您自己的笔记本电脑，但不要使用拨号机。
kalloctest 中锁争用的根本原因是 kalloc() 有一个受单个锁保护的空闲列表。要消除锁争用，您必须重新设计内存分配器以避免单个锁和列表。
- 基本思想是为每个CPU维护一个空闲列表，每个列表都有自己的锁。不同 CPU 上的分配和释放可以并行运行，因为每个 CPU 将在不同的列表上运行。
- 主要挑战是处理一个 CPU 的可用列表为空，但另一个 CPU 的列表具有可用内存的情况; 在这种情况下，一个 CPU 必须“窃取”另一个 CPU 的可用列表的一部分。
-
窃取可能会引入锁争用，但希望这种情况并不常见。
您的工作是实现每个 CPU 的空闲列表，并在 CPU 的空闲列表为空时窃取。您必须为所有以“kmem”开头的锁命名。也就是说，您应该为每个锁调用
initlock，并传递一个以“kmem”开头的名称。运行 kalloctest 以查看您的实现是否减少了锁争用。要检查它是否仍然可以分配所有内存，请运行usertests sbrkmuch。您的输出将类似于下面显示的输出，尽管具体数字会有所不同，但 kmem 锁上的争用总数大大减少。确保usertests -q中的所有测试都通过。$ kalloctest start test1 test1 results: --- lock kmem/bcache stats lock: kmem: #test-and-set 0 #acquire() 42843 lock: kmem: #test-and-set 0 #acquire() 198674 lock: kmem: #test-and-set 0 #acquire() 191534 lock: bcache: #test-and-set 0 #acquire() 1242 --- top 5 contended locks: lock: proc: #test-and-set 43861 #acquire() 117281 lock: virtio_disk: #test-and-set 5347 #acquire() 114 lock: proc: #test-and-set 4856 #acquire() 117312 lock: proc: #test-and-set 4168 #acquire() 117316 lock: proc: #test-and-set 2797 #acquire() 117266 tot= 0 test1 OK start test2 total free number of pages: 32499 (out of 32768) ..... test2 OK start test3 child done 1 child done 100000 test3 OK $ usertests sbrkmuch usertests starting test sbrkmuch: OK ALL TESTS PASSED $ usertests -q ... ALL TESTS PASSED $
提示
- 您可以使用来自
kernel/param.h的常量NCPU - 让
freerange将所有可用内存提供给运行freerange的 CPU。 - 函数
cpuid返回当前内核编号，但只有在中断关闭时调用它并使用其结果才是安全的。您应该使用push_off()和pop_off()来关闭和打开中断。 - 看看
kernel/sprintf.c中的snprintf函数，了解字符串格式的想法。不过，将所有锁命名为“kmem”是可以的。 -
（可选）使用 xv6 的竞争检测器运行您的解决方案
$ make clean $ make KCSAN=1 qemu $ kalloctest ..kalloctest 可能会失败，但你不应该看到任何竞争。如果 xv6 的竞争检测器观察到竞争，它将打印两条堆栈轨迹，描述以下路线的竞争：
== race detected == backtrace for racing load 0x000000008000ab8a 0x000000008000ac8a 0x000000008000ae7e 0x0000000080000216 0x00000000800002e0 0x0000000080000f54 0x0000000080001d56 0x0000000080003704 0x0000000080003522 0x0000000080002fdc backtrace for watchpoint: 0x000000008000ad28 0x000000008000af22 0x000000008000023c 0x0000000080000292 0x0000000080000316 0x000000008000098c 0x0000000080000ad2 0x000000008000113a 0x0000000080001df2 0x000000008000364c 0x0000000080003522 0x0000000080002fdc ==========在您的操作系统上，您可以通过将回溯跟踪剪切并粘贴到
addr2line中，将其转换为带有行号的函数名称：$ riscv64-linux-gnu-addr2line -e kernel/kernel 0x000000008000ab8a 0x000000008000ac8a 0x000000008000ae7e 0x0000000080000216 0x00000000800002e0 0x0000000080000f54 0x0000000080001d56 0x0000000080003704 0x0000000080003522 0x0000000080002fdc ctrl-d kernel/kcsan.c:157 kernel/kcsan.c:241 kernel/kalloc.c:174 kernel/kalloc.c:211 kernel/vm.c:255 kernel/proc.c:295 kernel/sysproc.c:54 kernel/syscall.c:251您不需要运行竞争检测器，但您可能会发现它很有帮助。请注意，竞争检测器会显著减慢 xv6 的速度，因此您可能不想在运行用户测试时使用它。
实现
-
将管理空闲list与锁的结构体按CPU分开：将原本表示空闲内存和其对应的锁的结构体变量
kmem从一个变成多个，让每个CPU都拥有一个该结构体。因此将其改为数组，大小为NCPU个：struct { struct spinlock lock; struct run* freelist; } kmem[NCPU]; -
更新内存初始化
kinit()：在内存初始化kinit()中将原本对一个kmem变量的初始化改为对kmem数组的初始化：void kinit() { // initlock(&kmem.lock, "kmem"); int i; char* name = "kmem0"; // 内存锁名字 for (i = 0; i -
更新内存释放
kfree()：根据提示可以将释放的内存要加入到当前运行的cpu对应的空闲list上。因此在原函数基础上，先获取cpuid，在通过cpuid索引得到需要对应管理空闲list和锁的结构体void kfree(void* pa) { ... // 在将要释放的内存填充完后 push_off(); // 获取cpuid要关闭中断 cid = cpuid(); // 获取cpuid acquire(&kmem[cid].lock); r->next = kmem[cid].freelist; kmem[cid].freelist = r; release(&kmem[cid].lock); pop_off(); // 恢复中断 } -
更新内存分配
kalloc()：内存分配需要考虑：如果当前cpu对应的内存管理结构体无空闲list，则需要使用其他cpu的空闲内存。-
首先不考虑借用其他cpu的情况，将
kmem单变量更新为数组（与kfree()类似）void* kalloc(void) { struct run* r; int cid, i; push_off(); cid = cpuid(); acquire(&kmem[cid].lock); r = kmem[cid].freelist; if(r) { kmem[cid].freelist = r->next; } release(&kmem[cid].lock); ... // TODO：需要添加借用其他cpu内存的逻辑代码 pop_off(); if (r) memset((char*)r, 5, PGSIZE); // fill with junk return (void*)r; } -
在完成对本cpu空闲list的读写操作后，无论是否有空闲内存都不需要再读写这个list了因此，在释放了本cpu对应的锁之后，加上是否成功获取空闲内存的判断：若有空闲内存则无需借用其他cpu，否则需要借用其他cpu的内存：
void* kalloc(void) { ... release(&kmem[cid].lock); // 在释放本cpu锁后、恢复中断前加入借用其他cpu的逻辑代码 if (!r) { // 若本cpu无空闲内存 for (i = 0; i next; release(&kmem[cid].lock); break; } else { // 否则就直接跳过，释放锁后换下一个cpu release(&kmem[cid].lock); } } } pop_off(); ... }
-
结果
- 运行结果
- 测试结果
Buffer cache (hard)
要求
作业的这一半独立于前半部分; 无论您是否完成了前半部分，您都可以处理这一半（并通过测试）。
如果多个进程集中使用文件系统，它们可能会争用 bcache.lock，它保护 kernel/bio.c 中的磁盘块缓存。bcachetest 创建多个进程，这些进程重复读取不同的文件，以便在 bcache.lock 上产生争用; 其输出如下所示（在完成本实验之前）：
$ bcachetest
start test0
test0 results:
--- lock kmem/bcache stats
lock: kmem: #test-and-set 0 #acquire() 33035
lock: bcache: #test-and-set 16142 #acquire() 65978
--- top 5 contended locks:
lock: virtio_disk: #test-and-set 162870 #acquire() 1188
lock: proc: #test-and-set 51936 #acquire() 73732
lock: bcache: #test-and-set 16142 #acquire() 65978
lock: uart: #test-and-set 7505 #acquire() 117
lock: proc: #test-and-set 6937 #acquire() 73420
tot= 16142
test0: FAIL
start test1
test1 OK您可能会看到不同的输出，但 bcache 锁的 test-and-sets 数量会很高。如果你查看kernel/bio.c中的代码，你会发现bcache.lock保护缓存块缓冲区的列表，每个块缓冲区中的引用计数（b->refcnt）以及缓存块的身份（b->dev和b->blockno）。
修改块缓存，以便在运行
bcachetest时，bcache 中所有锁的acquire循环迭代次数接近于零。理想情况下，块缓存中涉及的所有锁的计数总和应为零，但如果总和小于 500 也可以。修改bget和brelse，以便 bcache 中不同块的并发查找和释放不太可能在锁上发生冲突（例如，不必都等待bcache.lock）。您必须保持每个缓存块最多一个副本的不变性。完成后，输出应类似于下面显示的输出（尽管不相同）。确保usertests -q仍然通过。完成后，make grade应通过所有测试。
$ bcachetest
start test0
test0 results:
--- lock kmem/bcache stats
lock: kmem: #test-and-set 0 #acquire() 32954
lock: kmem: #test-and-set 0 #acquire() 75
lock: kmem: #test-and-set 0 #acquire() 73
lock: bcache: #test-and-set 0 #acquire() 85
lock: bcache.bucket: #test-and-set 0 #acquire() 4159
lock: bcache.bucket: #test-and-set 0 #acquire() 2118
lock: bcache.bucket: #test-and-set 0 #acquire() 4274
lock: bcache.bucket: #test-and-set 0 #acquire() 4326
lock: bcache.bucket: #test-and-set 0 #acquire() 6334
lock: bcache.bucket: #test-and-set 0 #acquire() 6321
lock: bcache.bucket: #test-and-set 0 #acquire() 6704
lock: bcache.bucket: #test-and-set 0 #acquire() 6696
lock: bcache.bucket: #test-and-set 0 #acquire() 7757
lock: bcache.bucket: #test-and-set 0 #acquire() 6199
lock: bcache.bucket: #test-and-set 0 #acquire() 4136
lock: bcache.bucket: #test-and-set 0 #acquire() 4136
lock: bcache.bucket: #test-and-set 0 #acquire() 2123
--- top 5 contended locks:
lock: virtio_disk: #test-and-set 158235 #acquire() 1193
lock: proc: #test-and-set 117563 #acquire() 3708493
lock: proc: #test-and-set 65921 #acquire() 3710254
lock: proc: #test-and-set 44090 #acquire() 3708607
lock: proc: #test-and-set 43252 #acquire() 3708521
tot= 128
test0: OK
start test1
test1 OK
$ usertests -q
...
ALL TESTS PASSED
$请给出所有以“bcache”开头的锁名称。也就是说，您应该为每个锁调用 initlock，并传递一个以“bcache”开头的名称。
减少块缓存中的争用比 kalloc 更棘手，因为 bcache 缓冲区实际上在进程（以及 CPU）之间是共享的。对于 kalloc，可以通过为每个 CPU 提供自己的分配器来消除大多数争用; 这不适用于块缓存。我们建议您使用每个哈希桶具有锁的哈希表在缓存中查找块号。
在某些情况下，如果解决方案存在锁定冲突，则可以：
- 当两个进程同时使用相同的块号时。
bcachetest test0从不这样做。 - 当两个进程同时未命中缓存，并且需要查找要替换的未使用块时。
bcachetest test0从不这样做。 - 当两个进程同时使用在用于分区块和锁的任何方案中冲突的块时; 例如，如果两个进程使用的块号散列到哈希表中的同一槽。
-
bcachetest test0可能会这样做，具体取决于您的设计，但您应该尝试调整方案的细节以避免冲突（例如，更改哈希表的大小）。
bcachetest 的 test1 使用比缓冲区更多的不同块，并执行大量文件系统代码路径。
提示
- 阅读 xv6 书中块缓存的说明（第 8.1-8.3 节）
- 可以使用固定数量的存储桶，而不是动态调整哈希表的大小。使用质数存储桶（例如 13）来降低哈希冲突的可能性。
- 在哈希表中搜索缓冲区并在找不到缓冲区时为该缓冲区分配条目必须是原子的。
- 删除所有缓冲区的列表（
bcache.head等），不要实现LRU。通过此更改，brelse不需要获取 bcache 锁。在bget中，您可以选择任何具有refcnt == 0的块，而不是最近最少使用的块。 - 您可能无法以原子方式检查已缓存的 buf 和（如果未缓存）找到未使用的 buf; 如果缓冲区不在缓存中，则可能必须删除所有锁并从头开始。可以在
bget中序列化查找未使用的 buf（即，当查找未命中缓存时选择要重用的缓冲区的bget部分）。 - 在某些情况下，解决方案可能需要保留两个锁; 例如，在逐出期间，您可能需要持有 bcache 锁和每个存储桶的锁。确保避免死锁。
- 替换块时，您可以将
struct buf从一个存储桶移动到另一个存储桶，因为新块哈希到另一个存储桶。您可能遇到一个棘手的情况：新块可能会散列到与旧块相同的存储桶。在这种情况下，请确保避免死锁。 - 一些调试技巧：实现存储桶锁，但将全局
bcache.lock的获取/释放保留在bget的开头/结尾以序列化代码。确定它正确无误且没有争用条件后，请删除全局锁并处理并发问题。您还可以运行make CPUS=1 qemu以使用一个内核进行测试。 - 使用xv6的竞争检测器来查找潜在的竞争（请参阅上文如何使用竞争检测器）。
实现
-
哈希表桶数与索引方式：根据提示，需要把原本双向list的bcache改为哈希表。实现哈希表，首先需要决定桶数量和哈希索引方式，根据提示可将桶数量设为13，哈希索引一般采取取余的方式。
#define NUM_BUCKET 13 #define HASH(x) ((x) % NUM_BUCKET) -
更改
bcache结构体：更改控制bcache的结构体，将原本以双向list维护缓存的形式分桶，每个桶依然以双向list维护，同时将原本一个锁分为每个桶一个锁struct { struct spinlock locks[NUM_BUCKET]; struct buf buf[NBUF]; // Linked list of all buffers, through prev/next. // Sorted by how recently the buffer was used. // head.next is most recent, head.prev is least. struct buf heads[NUM_BUCKET]; } bcache; -
bcache初始化binit()：根据源码，初始化主要分为三步：- 初始化锁：从初始化一个锁到初始化每个桶的锁
- 初始化双向list头节点：初始化一个头节点到初始化每个桶的头节点
-
将所有buf放入list（hash table）：原本将buf放入list，现在需要把buf放入哈希表中的list
void binit(void) { struct buf* b; char* lockname = "bcache0"; uint i; // 初始化锁和头节点，放一块了 for (i = 0; i blockno); // 可以不用hash，最开始都是0 b->next = bcache.heads[i].next; b->prev = bcache.heads + i; initsleeplock(&b->lock, "buffer"); bcache.heads[i].next->prev = b; bcache.heads[i].next = b; } }
-
bcache释放brelse()及其他：由于从原本一把🔒变为分桶🔒，因此再去访问or读写bcache其中一个桶的时候就可以只获取该桶对应的锁了。受此影响需要改变的三个函数：void brelse(struct buf* b) { uint idx = HASH(b->blockno); if (!holdingsleep(&b->lock)) panic("brelse"); releasesleep(&b->lock); acquire(bcache.locks + idx); b->refcnt--; if (b->refcnt == 0) { // 将b从列表中删去 b->next->prev = b->prev; b->prev->next = b->next; // 将b加入到本桶list head的左边 b->prev = bcache.heads[idx].prev; b->next = bcache.heads + idx; bcache.heads[idx].prev->next = b; bcache.heads[idx].prev = b; } release(bcache.locks + idx); } void bpin(struct buf* b) { uint idx = HASH(b->blockno); acquire(bcache.locks + idx); b->refcnt++; release(bcache.locks + idx); } void bunpin(struct buf* b) { uint idx = HASH(b->blockno); acquire(bcache.locks + idx); b->refcnt--; release(bcache.locks + idx); } -
bcache获取bget()：bget()函数不仅🔒受到影响，由于分桶，在本桶没有refcnt == 0的缓存时还需要去其他桶中寻找。基本思路（不实现LRU）：- 先寻找是否已经有缓存，若有直接返回
- 若没有缓存则需要替换
refcnt == 0的缓存。可以先在本桶寻找是否有refcnt == 0的缓存，若有则替换后返回。 -
若本桶没有
refcnt == 0的缓存，就需要去其他桶中寻找。若找到，则需要将其从原本桶的list中删去，然后加入本桶的list中。static struct buf* bget(uint dev, uint blockno) { struct buf* b; uint jdx, idx = HASH(blockno); acquire(bcache.locks + idx); // 先获取本桶🔒 // Is the block already cached? for (b = bcache.heads[idx].next; b != bcache.heads + idx; b = b->next) { if (b->dev == dev && b->blockno == blockno) { b->refcnt++; release(bcache.locks + idx); acquiresleep(&b->lock); return b; // 若已缓存直接返回 } } // 若没有缓存先在本桶找refcnt==0的缓存 for (b = bcache.heads[idx].prev; b != bcache.heads + idx; b = b->prev) { if (b->refcnt == 0) { b->dev = dev; b->blockno = blockno; b->valid = 0; b->refcnt = 1; release(bcache.locks + idx); acquiresleep(&b->lock); return b; // 找到、替换后直接返回 } } release(bcache.locks + idx); // 本桶没有refcnt == 0的桶，就要找别的桶，但是需要先释放本桶的锁 for (jdx = HASH(idx + 1); jdx != idx; jdx = (jdx + 1) % NUM_BUCKET) { acquire(bcache.locks + jdx); for (b = bcache.heads[jdx].prev; b != bcache.heads + jdx; b = b->prev) { if (b->refcnt == 0) { // 找到后替换内容 b->dev = dev; b->blockno = blockno; b->valid = 0; b->refcnt = 1; // 然后在jdx桶list中删去b b->next->prev = b->prev; b->prev->next = b->next; release(bcache.locks + jdx); // 删去后就可以释放jdx桶的锁了 // 最后把b放入本桶list acquire(bcache.locks + idx); // 之前释放了本桶的锁，现在需要重新获取 b->next = bcache.heads[idx].next; b->prev = bcache.heads + idx; bcache.heads[idx].next->prev = b; bcache.heads[idx].next = b; release(bcache.locks + idx); // 操作后释放本桶锁 acquiresleep(&b->lock); return b; } } release(bcache.locks + jdx); } panic("bget: no buffers"); }
结果
- 运行结果
- 测试结果
总结
make grade
个人收获
- 内存和缓存是内核中重要的临界区，存在许多并发访问or读写。一方面可以通过配备一把锁统一管理的方式，这样编程时不易出现死锁但效率较低；
- 另一方面可以分开管理，例如内存通过CPU分开、bcache缓存通过哈希桶分开，这时候就要对每个分区配备独立的锁。独立的锁保证了不同分区之间可以并发or并行，从而提高了效率。
- 分开管理还需要考虑分区之间的交互问题。例如某个分区的资源不够时需要去访问其他分区。这时候就需要考虑在分区交互期间如何保持互斥和避免死锁的问题。
服务器托管，北京服务器托管，服务器租用 http://www.fwqtg.net
