当两个信号量同时变化并且两个信号量之一不能立即递减时会发生什么？答案

【问题标题】：What happens when two semaphores change simultaneously and one of the two cannot decrement immediately?当两个信号量同时变化并且两个信号量之一不能立即递减时会发生什么？
【发布时间】：2021-04-05 23:12:01
【问题描述】：

考虑这段代码：

struct sembuf s_op[2];

s_op[0].sem_num = old;
s_op[0].sem_op = 1;
s_op[0].sem_flg = 0;
s_op[1].sem_num = new;
s_op[1].sem_op = -1;
s_op[1].sem_flg = 0;
semop(semid, s_op, 2);

现在，如果“新”不能立即递减（因为它已经为 0），是否会有一段时间“旧”信号量增加而“新”信号量没有增加？我的意思是，这是否总是原子地发生？

【问题讨论】：

pubs.opengroup.org/onlinepubs/009695399/functions/semop.html man7.org/linux/man-pages/man2/semop.2.html: "semop() 函数应在与参数 semid 指定的信号量标识符相关联的信号量集合上自动执行用户定义的信号量操作数组。"
处理器如何做到这一点？如果它在同一个原子操作中测试并设置两个变量并且两个变量之一失败，它应该“撤消”成功的操作，因此该操作将不再是原子的，对吧？
（对不起 - 我错过了标题并评论了问题中的问题）。它由内核或libc实现。最简单的方法就是拥有一个全局锁，但这对性能不利。实际实施可能更复杂。通常，您不会发现在实现此类原语时单独使用 test-and-set，因为您通常不想不必要地旋转和使用宝贵的资源，而是在此 CPU 上安排另一个线程。
例如带有所有血腥细节的内核 5.10 实现：elixir.bootlin.com/linux/v5.10.3/source/ipc/sem.c#L1981 我不会假装我理解了代码，它可能超出了 SO 的范围。您可能需要阅读有关 LWN 的文章并深入研究源代码。
How the processor do that? 处理器无关紧要，您在用户空间。 semop 可能是一个系统调用，然后内核可以暂停所有进程，执行操作并根据需要恢复所有进程。

标签： c posix ipc semaphore atomic

【解决方案1】：

是否会有一个“旧”信号量增加而“新”信号量没有增加的时间间隔？

这是一个实现细节。实现可以检查操作是否可能，然后执行操作，或者一个接一个地执行操作，如果一个操作不能完成，则恢复已经完成的操作。或者实现可能根本不做任何事情，直到下一个 sem* 操作并查询操作，等等。底部是任何一种方式只有 observable 状态才是重要的，这样的状态其中一个增加了，而另一个则不能被任何进程观察到，因为操作是“原子地”完成的。

这总是会自动发生吗？

是的，其中“原子地”意味着没有进程可以观察操作中间的状态。这并不意味着这种状态“不存在”，因为内核必须一个接一个地查询一个操作。这意味着无法从使用 POSIX api 的进程中观察到这种状态。

【讨论】：

【解决方案2】：

以下是 Linux 5.4 中的一些实现细节。管理 sysV 信号量的源代码在 ipc/sem.c 中。

semop() 系统调用入口点调用公共的内部 do_semtimedop() 函数，并将 timeout 参数设置为 NULL： p>

SYSCALL_DEFINE3(semop, int, semid, struct sembuf __user *, tsops,
        unsigned, nsops)
{
    return do_semtimedop(semid, tsops, nsops, NULL);
}

在对参数和权限进行一些检查后，松散地说，do_semtimedop() 锁定了整个信号量集并调用另一个名为 perform_atomic_semop() 的内部例程，这使得请求的“sem 操作”。 此函数扫描信号量集两次：

第一次扫描循环确定是否有任何操作导致集合中的信号量为负值。如果是，则如果在标志中设置了 IPC_NOWAIT，则返回 -EAGAIN，如果未设置，则返回 1。如果其中一个操作导致的值大于最大值（在 include/uapi/linux/sem 中定义为 SEMVMX = 32767），它也可能返回 -ERANGE .h）。
第二次扫描循环 将请求的操作作为第一个循环确定这不会导致信号量为负值。返回 0。

static int perform_atomic_semop(struct sem_array *sma, struct sem_queue *q)
[...]
    /*
     * We scan the semaphore set twice, first to ensure that the entire
     * operation can succeed, therefore avoiding any pointless writes
     * to shared memory and having to undo such changes in order to block
     * until the operations can go through.
     */
    for (sop = sops; sop < sops + nsops; sop++) {
        int idx = array_index_nospec(sop->sem_num, sma->sem_nsems);

        curr = &sma->sems[idx];
        sem_op = sop->sem_op;
        result = curr->semval;

        if (!sem_op && result)
            goto would_block; /* wait-for-zero */

        result += sem_op;
        if (result < 0)
            goto would_block;

        if (result > SEMVMX)
            return -ERANGE;

        if (sop->sem_flg & SEM_UNDO) {
            int undo = un->semadj[sop->sem_num] - sem_op;

            /* Exceeding the undo range is an error. */
            if (undo < (-SEMAEM - 1) || undo > SEMAEM)
                return -ERANGE;
        }
    }

    for (sop = sops; sop < sops + nsops; sop++) {
        curr = &sma->sems[sop->sem_num];
        sem_op = sop->sem_op;
        result = curr->semval;

        if (sop->sem_flg & SEM_UNDO) {
            int undo = un->semadj[sop->sem_num] - sem_op;

            un->semadj[sop->sem_num] = undo;
        }
        curr->semval += sem_op;
        ipc_update_pid(&curr->sempid, q->pid);
    }

    return 0;

would_block:
    q->blocking = sop;
    return sop->sem_flg & IPC_NOWAIT ? -EAGAIN : 1;
}

回到do_semtimedop()，检查perform_atomic_semop()的返回：

如果返回值为 0，则信号量集上的任何等待任务都会在被唤醒之前被唤醒以运行其挂起的操作（即，为每个等待任务调用 do_semtimedop()）。全局锁被释放，返回0
如果返回值为负（-EAGAIN 或任何其他检测到的错误，如 -ERANGE），则释放全局锁并返回 errno
如果返回值 > 0（实际上是 1），则调用任务进入睡眠状态：它被添加到信号量集的等待队列中。调用被阻塞，直到任务被唤醒。

因此，为了回答这个问题，没有中间时间段，其中一个信号量操作完成而另一个信号量操作在同一集合上挂起。一次全部完成，或者调用任务进入休眠状态，直到它可以完成为止。

【讨论】：