在 xv6 中实现自旋锁功能，以便能够从用户级别使用 API

Question

xv6 有 spinlock.c 文件，用于为内核创建自旋锁。但我需要实现在用户级别使用的自旋锁 API。例如，我将实现 sp_create() 以在用户级别创建自旋锁。或者调用 sp_acquire(int id) 来获取锁等。为此，我应该创建系统调用并将实际实现放在内核中。 xv6 具有自旋锁功能，但只能在内核级别使用。

我想过创建系统调用，它实际上是调用 spinlock.c 中的相应函数来创建锁，获取它，释放它等。但是由于中断禁用的一些问题，它不起作用。

我在下面复制我到目前为止写的代码：

//system call for lock_take():
int  l_take(int lockid) {

  struct proc *curproc = myproc();

  //process will take lock
  ..
    acquire(&LL.arraylockList[lockid].spnLock);  
  ..
  return 0; 
}

我在这里遇到的问题是它给了我关于恐慌的错误：sched locks 我认为这是因为 acquire() 代码中有 pushcli() 。

void
acquire(struct spinlock *lk)
{
    pushcli(); // disable interrupts to avoid deadlock.
    if (holding(lk)) panic("acquire");

    // The xchg is atomic.
    while (xchg(&lk->locked, 1) != 0)
        ;

    // Tell the C compiler and the processor to not move loads or stores
    // past this point, to ensure that the critical section's memory
    // references happen after the lock is acquired.
    __sync_synchronize();

    // Record info about lock acquisition for debugging.
    lk->cpu = mycpu();
    getcallerpcs(&lk, lk->pcs);
}

然后我将代码复制到一个新函数 acquire2() 并在我的系统调用中使用它，其中 pushcli() 被注释掉：

acquire2(struct spinlock *lk)
{
        //    pushcli(); // disable interrupts to avoid deadlock.
    if (holding(lk)) panic("acquire");

    // The xchg is atomic.
    while (xchg(&lk->locked, 1) != 0) {
      ;
    }

    // Tell the C compiler and the processor to not move loads or stores
    // past this point, to ensure that the critical section's memory
    // references happen after the lock is acquired.
    __sync_synchronize();

    // Record info about lock acquisition for debugging.
    lk->cpu = mycpu();
    getcallerpcs(&lk, lk->pcs);
}

但是，错误消息更改为： 恐慌：在启用中断的情况下调用 mycpu()

原来是禁用中断是不允许的。因此，不应使用 pushcli() 和 popcli()。然后我需要弄清楚如何以原子方式运行 mycpu() 。它的实现是这样的：

// Must be called with interrupts disabled to avoid the caller being rescheduled
// between reading lapicid and running through the loop.
struct cpu *
mycpu(void)
{
    int apicid, i;

    if (readeflags() & FL_IF) panic("mycpu called with interrupts enabled\n");

    apicid = lapicid();
    // APIC IDs are not guaranteed to be contiguous. Maybe we should have
    // a reverse map, or reserve a register to store &cpus[i].
    for (i = 0; i < ncpu; ++i) {
        if (cpus[i].apicid == apicid) return &cpus[i];
    }
    panic("unknown apicid\n");
}

for 循环和它上面的行需要原子执行。 我该怎么做？

Answer 1

向后工作（从问题的末尾开始）...

我该怎么做？

如果你必须这样做；您首先要找到一种不涉及循环的替代方法。

可能有许多可能的替代方案（假设 80x86；使用“通过 GS 或 FS 以某种方式找到的 CPU 本地结构，每个 CPU 使用不同的 TSS 并使用它来查找 CPU 编号，在确保之后使用 RDTSCP 指令它被支持并配置为适合，窃取/重新利用像 DR3 这样的寄存器，使用分页来创建“CPU 特定区域”，...）。

for 循环和它上面的行需要原子执行。

为什么？假设该表是在启动期间生成的，并且此后从未更改（并且没有热插拔 CPU 支持）；或假设表仅在 CPU 需要查找自身时以确保其正确性的方式更改（认识到已脱机的 CPU 无法尝试查找自身）；没有理由（我可以看到）循环需要是原子的。

原来是禁用中断是不允许的。

注意：这是不对的，应该是“事实证明必须禁用中断”。

这是因为操作系统正在使用“禁用 IRQ”来禁用/推迟任务切换。这不是一个好的或必要的要求（内核可以在其他地方有自己的标志/变量来禁用/推迟任务切换而不禁用 IRQ；以及 2 种不同类型的自旋锁，其中一种由 IRQ 处理程序执行的代码使用禁用两者和一个从未被可能由 IRQ 处理程序执行的代码使用的代码，这些处理程序仅禁用/推迟任务并且从不禁用 IRQ）。当然，这需要对内核进行一些重大更改。

请注意，对于您发布的代码，获取自旋锁极有可能会导致所有 IRQ 被禁用，然后释放自旋锁会导致 IRQ 再次重新启用。这意味着（如果您让用户空间通过系统调用使用内核代码）用户空间可以获取任何自旋锁，然后永远占用 100% 的 CPU 时间（通过从不释放自旋锁）。换句话说，您计划做的是制造一个巨大的“拒绝服务”安全漏洞。

我想过创建系统调用，实际上是调用 spinlock.c 中的相应函数来创建锁、获取锁、释放锁等。

这是你所有问题的原因。

无需创建系统调用，只需为用户空间创建一些自旋锁代码。您可以通过将内核的函数直接复制（使用“复制和粘贴”）到您的用户空间代码（或者可能是库）中并删除所有不相关的内容以最终得到类似的结果：

my_acquire(int *lk)
{
    // The xchg is atomic.
    while (xchg(lk, 1) != 0) {
      ;
    }

    // Tell the C compiler and the processor to not move loads or stores
    // past this point, to ensure that the critical section's memory
    // references happen after the lock is acquired.
    __sync_synchronize();
}

注意：为此，您还需要将xchg() 和__sync_synchronize() 的代码复制到用户空间；但这应该不是问题。

但是；我应该指出内核中的代码不好。通常你想使用“测试；然后原子地测试和设置”的方法，这样（在争论中）你没有做任何原子的事情（只是在循环中做测试）；并且（对于 80x86）您想在循环中使用 pause 指令来提高“循环退出速度”并提高核心中其他逻辑 CPU 的速度（用于超线程）；并且（对于内核）您不想在测试时禁用 IRQ，而只想为“原子测试和设置”禁用 IRQ，以避免无缘无故破坏“IRQ 延迟”。

在用户级实现自旋锁

总的来说，这也可能是个坏主意。

问题是（如果不是大规模的安全灾难）在您持有锁时可能会发生任务切换；导致所有其他任务（例如可能有许多 CPU）浪费它们的整个时间片旋转而没有任何获得锁的希望（因为锁已经被未运行的任务获取并且不会运行，因为任务正忙于浪费 CPU 时间而是白白旋转）。

解决这个问题（不是大规模安全灾难）的方法是互斥体/信号量。具体来说，解决方案是能够告诉内核的调度程序“在它可以获取所需的互斥体/信号量之前，不要给这个任务 CPU 时间”。当然，这可以实现为“快乐情况”（当根本没有争用时）纯粹在用户空间中处理，并且内核（以及系统调用的开销等）仅在实际需要时才涉及.