我可以在等待/信号信号量中切换测试和修改部分吗？

Question

wait() 和 signal() 信号量的经典 none-busy-waiting 版本实现如下。在此版本中，value 可以为负数。

//primitive
wait(semaphore* S)
{
    S->value--;
    if (S->value < 0)
    {
        add this process to S->list;
        block();
    }
}

//primitive
signal(semaphore* S)
{
    S->value++;
    if (S->value <= 0)
    {
        remove a process P from S->list;
        wakeup(P);
    }
}

问题：下面的版本也正确吗？这里我先测试一下，修改一下值。如果您能给我展示一个它不起作用的场景，那就太好了。

//primitive wait().
//If (S->value > 0), the whole function is atomic
//otherise, only if(){} section is atomic
wait(semaphore* S)
{
    if (S->value <= 0)
    {
        add this process to S->list;
        block();
    }
    // here I decrement the value after the previous test and possible blocking
    S->value--;
}

//similar to wait()
signal(semaphore* S)
{
    if (S->list is not empty)
    {
        remove a process P from S->list;
        wakeup(P);
    }
    // here I increment the value after the previous test and possible waking up
    S->value++;
}

编辑：

我的动机是想弄清楚我是否可以使用后一个版本来实现互斥，并且没有死锁，没有饥饿。

Answer 1

您的修改版本引入了竞争条件：

• 线程 A: if(S->Value < 0) // Value = 1
• 线程 B: if(S->Value < 0) // Value = 1
• 线程 A: S -> 值--; // 值 = 0
• 线程 B: S->Value--; // Value = -1

两个线程都获取了 count=1 信号量。哎呀。请注意，即使它们是不可抢占的（见下文），也存在另一个问题，但为了完整性，这里讨论原子性以及真正锁定协议的工作原理。

当使用这样的协议时，确定您正在使用的原子原语非常重要。原子原语看起来是即时执行的，而不与任何其他操作交织在一起。你不能只把一个大函数称为原子函数，而是将其称为原子函数。你必须使用其他原子基元以某种方式使其原子化。

大多数 CPU 都提供称为“原子比较和交换”的原语。从这里开始我将其缩写为 cmpxchg。语义如下：

bool cmpxchg(long *ptr, long old, long new) {
    if (*ptr == old) {
        *ptr = new;
        return true;
    } else {
        return false;
    }
}

cmpxchg 未使用此代码实现。它位于 CPU 硬件中，但行为有点像这样，只是原子地。

现在，让我们添加一些额外的有用函数（由其他原语构建）：

• add_waitqueue(waitqueue) - 将进程状态设置为睡眠并将我们添加到等待队列，但继续执行（原子）
• Schedule() - 切换线程。如果我们处于睡眠状态，我们不会再次运行，直到被唤醒（阻塞）
• remove_waitqueue（waitqueue） - 从等待队列中删除我们的进程，然后将我们的状态设置为唤醒（如果它还没有）（原子）
• 内存_屏障（ ） - 确保在此点之前的任何逻辑读/写实际上都在该点之前执行，避免令人讨厌的内存排序问题（我们假设所有其他原子原语都带有一个空闲的内存屏障，尽管这并不总是正确的）（CPU/编译器原语）

这是典型的信号量获取例程的外观。它比您的示例复杂一点，因为我已经明确确定了我正在使用的原子操作：

void sem_down(sem *pSem)
{
    while (1) {
        long spec_count = pSem->count;
        read_memory_barrier(); // make sure spec_count doesn't start changing on us! pSem->count may keep changing though
        if (spec_count > 0)
        {
            if (cmpxchg(&pSem->count, spec_count, spec_count - 1)) // ATOMIC
                return; // got the semaphore without blocking
            else
                continue; // count is stale, try again
        } else { // semaphore count is zero
            add_waitqueue(pSem->wqueue); // ATOMIC
            // recheck the semaphore count, now that we're in the waitqueue - it may have changed
            if (pSem->count == 0) schedule(); // NOT ATOMIC
            remove_waitqueue(pSem->wqueue); // ATOMIC
            // loop around again to try to acquire the semaphore
        }
    }
}

您会注意到非零 pSem->count 的实际测试，在现实世界的 semaphore_down 函数中，是由 cmpxchg 完成的。你不能相信任何其他读物；读取该值后，该值可能会立即发生变化。我们根本无法将值检查和值修改分开。

这里的spec_count是推测。这很重要。我基本上是在猜测计数会是多少。这是一个很好的猜测，但它只是一个猜测。如果我的猜测错误，cmpxchg 将失败，此时例程必须循环并重试。如果我猜为 0，那么我要么会被叫醒（因为当我在等待队列上时它不再为零），要么我会注意到它在计划测试中不再为零。

您还应该注意，没有可能的方法使包含阻塞操作的函数成为原子的。这是荒谬的。根据定义，原子函数似乎是立即执行的，不与其他任何东西交错。但根据定义，阻塞函数会等待其他事情发生。这是不一致的。同样，任何原子操作都不能在阻塞操作中“拆分”，如您的示例所示。

现在，您可以通过声明函数不可抢占来消除很多这种复杂性。通过使用锁或其他方法，您只需确保信号量代码中一次只有一个线程运行（当然不包括阻塞）。但问题仍然存在。从值 0 开始，其中 C 已将信号量取下两次，然后：

• 线程 A: if (S->Value < 0) // Value = 0
• 线程 A: 阻止....
• 线程 B: if ( S->Value < 0) // Value = 0
• 线程 B: Block....
• 线程 C: S->Value++ // value = 1
• 线程 C: Wakeup(A)
• （线程 C 调用 signal() （线程 C 调用wait
• 线程 C: S->Value++ // value = 2
• 线程 C: Wakeup(B)
• ()）
• 线程 C: if (S->Value <= 0) // Value = 2
• 线程 C: S->Value-- // Value = 1
• // A 和 B 已被唤醒
• 线程 A: S->Value-- // Value = 0
• 线程 B: S->Value-- //值= -1

您可能可以通过循环重新检查 S->value 来修复此问题 - 假设您在单处理器计算机上并且您的信号量代码是可抢占的。不幸的是，这些假设在所有桌面操作系统上都是错误的:)

有关真实锁定协议如何工作的更多讨论，您可能会对论文“Fuss、Futexes 和 Furwocks：Linux 中的快速用户级锁定"