我应该怎么做才能使两个接收过程在 Promela 模型中不连续两次？

Question

我是旋转的初学者。我正在尝试让模型交替运行两个接收进程（模型中称为消费者的函数），即。 （消费者 1，消费者 2，消费者 1，消费者 2，...）。但是当我运行这段代码时，我的 2 个消费者进程的输出随机显示。有人能帮我吗？ 这是我正在努力解决的代码。

mtype = {P, C};

mtype turn = P;

chan ch1 = [1] of {bit};
byte current_consumer = 1;
byte previous_consumer;

active [2] proctype Producer()
{`
    bit a = 0;
    do
    :: atomic {
        turn == P ->
            ch1 ! a;
            printf("The producer %d --> sent %d!\n", _pid, a);
            a = 1 - a;
            turn = C;
    }
    od
    
}

active [2] proctype Consumer()
{
    bit b;

    do
    :: atomic{
        turn == C ->
            current_consumer = _pid;
            ch1 ? b;
            printf("The consumer %d --> received %d!\n\n", _pid, b);
            assert(current_consumer == _pid);
            turn = P;
    }
    od
    
}

Sample out is as photo

Answer 1

首先，让我提请您注意atomic's documentation的这段摘录：

如果原子序列中的任何语句阻塞，则原子性丢失，然后允许其他进程开始执行语句。当被阻塞的语句再次变为可执行时，原子序列的执行可以随时恢复，但不一定立即恢复。在进程可以恢复序列剩余部分的原子执行之前，该进程必须首先与系统中的所有其他活动进程竞争以重新获得控制权，也就是说，它必须首先被调度执行。

在您的模型中，这目前不会导致任何问题，因为ch1 是一个缓冲通道（即它的大小为>= 1）。但是，模型中的任何微小变化都可能破坏这个不变量。

---

从 cmets 中，我了解到您的目标是替代消费者，但您并不真正关心哪个生产者正在发送数据。

说实话，您的模型已经包含两个流程如何相互交替的示例：

• 生产者/消费者通过 turn 相互交替，每次分配一个不同的值
• 生产者/消费者通过ch1 交替也，因为它的大小为1

但是，这两种方法都是交替的生产者/消费者而不是消费者本身。

---

我喜欢的一种方法是使用eval (see docs) 进行消息过滤：每个消费者都知道自己的id，在单独的频道中等待带有自己的id 的令牌，并且仅在此情况下可用它开始做一些工作。

byte current_consumer;
chan prod2cons = [1] of { bit };
chan cons = [1] of { byte };

proctype Producer(byte id; byte total)
{
    bit a = 0;
    do
        :: true ->
            // atomic is only for printing purposes
            atomic {
                prod2cons ! a;
                printf("The producer %d --> sent %d\n", id, a);
            }
            a = 1 - a;
    od
}

proctype Consumer(byte id; byte total)
{
    bit b;
    do
        :: cons?eval(id) ->
            current_consumer = id;
            atomic {
                prod2cons ? b;
                printf("The consumer %d --> received %d\n\n", id, b);
            }
            assert(current_consumer == id);

            // yield turn to the next Consumer
            cons ! ((id + 1) % total)
    od
}

init {
    run Producer(0, 2);
    run Producer(1, 2);

    run Consumer(0, 2);
    run Consumer(1, 2);

    // First consumer is 0
    cons!0;
}

这个模型，简单地说：

• 生产者/消费者交替通过prod2cons，一个大小为1 的通道。这会强制执行以下行为：在 some 生产者创建消息后 some 消费者必须使用它。
• 消费者通过cons 交替，一个大小为1 的通道包含一个token 值，指示当前允许哪个消费者执行某些工作。所有消费者都可以查看cons 的内容，但只有匹配id 的消费者才能消费令牌并继续前进。在轮到它的最后，消费者用链中的下一个id 创建一个新的令牌。消费者以循环法的方式交替进行。

输出是：

      The producer 0 --> sent 0
                  The consumer 1 --> received 0

          The producer 1 --> sent 1
              The consumer 0 --> received 1

          The producer 1 --> sent 0
                  The consumer 1 --> received 0

           ...

      The producer 0 --> sent 0
                  The consumer 1 --> received 0

      The producer 0 --> sent 1
              The consumer 0 --> received 1

      The producer 0 --> sent 0
                  The consumer 1 --> received 0

      The producer 0 --> sent 1
              The consumer 0 --> received 1

请注意，生产者不一定会相互交替，而消费者则根据要求这样做。