Message Queue Pattern

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101

# Example 0065: Message Queue Pattern
# Distributed message queue with multiple producers and consumers
# Demonstrates: roles, message passing, FIFO ordering, producer-consumer coordination

---
deadlock_detection: false
---

role Producer:
    action Init:
        self.sent_count = 0

    atomic fair action Produce:
        require self.sent_count < MAX_MESSAGES_PER_PRODUCER

        msg_id = self.sent_count
        msg = record(producer_id=self.__id__, msg_id=msg_id, payload="msg_" + str(msg_id))

        # Send message to queue (atomic to prevent duplicates from crashes)
        queue.Enqueue(msg)
        self.sent_count = self.sent_count + 1

role Consumer:
    action Init:
        self.received_count = 0
        self.received_messages = []

    atomic fair action Consume:
        require self.received_count < MAX_MESSAGES_TOTAL

        # Receive message from queue
        msg = queue.Dequeue(self.__id__)

        if msg != None:
            self.received_messages.append(msg)
            self.received_count = self.received_count + 1

role Queue:
    action Init:
        self.queue = []
        self.total_enqueued = 0

    atomic func Enqueue(msg):
        # Accept message into queue (atomic to prevent inconsistency)
        self.queue.append(msg)
        self.total_enqueued = self.total_enqueued + 1

    atomic func Dequeue(consumer_id):
        # FIFO: dequeue from front if available
        if len(self.queue) == 0:
            return None

        msg = self.queue[0]
        self.queue = self.queue[1:]  # Remove first element
        return msg

NUM_PRODUCERS = 2
NUM_CONSUMERS = 2
MAX_MESSAGES_PER_PRODUCER = 2
MAX_MESSAGES_TOTAL = NUM_PRODUCERS * MAX_MESSAGES_PER_PRODUCER

action Init:
    queue = Queue()

    producers = []
    for i in range(NUM_PRODUCERS):
        producers.append(Producer())

    consumers = []
    for i in range(NUM_CONSUMERS):
        consumers.append(Consumer())

# Safety: total messages consumed <= total enqueued
always assertion ValidDelivery:
    total_consumed = 0
    for c in consumers:
        total_consumed = total_consumed + c.received_count
    return total_consumed <= queue.total_enqueued

# Safety: queue maintains FIFO order
always assertion QueueFIFO:
    if len(queue.queue) >= 2:
        # Messages should be ordered by when they were enqueued
        return queue.queue[0].msg_id <= queue.queue[1].msg_id or \
               queue.queue[0].producer_id != queue.queue[1].producer_id
    return True

# Safety: no message duplicates in any consumer
always assertion NoDuplicates:
    for consumer in consumers:
        msg_ids = [(msg.producer_id, msg.msg_id) for msg in consumer.received_messages]
        if len(msg_ids) != len(set(msg_ids)):
            return False
    return True

# Liveness: all produced messages eventually consumed
eventually always assertion AllMessagesConsumed:
    total_consumed = 0
    for c in consumers:
        total_consumed = total_consumed + c.received_count
    return total_consumed == MAX_MESSAGES_TOTAL and len(queue.queue) == 0

StateSpaceOptions: options:{max_actions:100 max_concurrent_actions:2} deadlock_detection:false
Nodes: 225, queued: 0
Valid Nodes: 225 Unique states: 225
Checking eventually always AllMessagesConsumed
IsLive: true
PASSED: Model checker completed successfully

gdoc_arrow_left_alt Producer-Consumer Pattern Cache Coherence Pattern gdoc_arrow_right_alt