SimPy制造工作流建模中Put请求竞态条件问题求助

我用SimPy对制造工作流建模时遇到了竞态条件：作为缓冲区/队列的Store需要给多个同样以Store建模的机器提供产品。当前做法是给每台机器发送put请求，假设它们在不同时间可用，之后取消多余请求，但这种方式无法处理两台及以上机器同一时间单位同时可用的情况——此时多个put会同时成功，导致实体被重复存入多个机器的队列。

当前实现代码：

def EntityHandover(node: Node) -> Generator:
    while True:
        with node.get() as item:
            entity = yield item

            receivers = [receiver.put(entity) for receiver in node.next]
            # node.next 是后继节点列表

            yield node.env.any_of(receivers)

            if sum(int(rec.processed) for rec in receivers) > 1:
                # 检测到实体被重复放入多个Store
                pass
            for rec in receivers:
                if not rec.processed:
                    print(f'Cancel: {rec=}')
                    rec.cancel()

# node 调用 env.process(EntityHandover(self))
# node.get() 和 node.put() 内部调用 node.store.get() 和 node.store.put()，并做了统计跟踪

我考虑过两个方案，但都觉得不合适：

• 创建包装类，通过回调替换实际实体，或在移交完成后删除；
• 修改SimPy源码，让put请求按优先级处理，同一时刻仅一个请求能yield。

我尝试过通过检查后继节点是否立即可用、仅发送一个请求来缓解问题，但无法解决多节点同时可用的场景。恳请提供架构或实现层面的建议。

方案1：引入实体分配协调器（推荐） #

核心思路是避免同时向多个节点发起put请求，通过集中式协调器分配实体给第一个可用的后继节点，从根源上杜绝多节点同时接收实体的可能。

实现步骤：

1. 创建协调器类，维护后继节点状态并提供分配接口：

class Coordinator:
    def __init__(self, env, nodes):
        self.env = env
        self.nodes = nodes  # 所有后继节点列表

    def assign_entity(self, entity):
        # 遍历节点，找到第一个有剩余容量的节点
        for node in self.nodes:
            if len(node.store.items) < node.store.capacity:
                yield node.put(entity)
                return
        # 所有节点都满时，等待任意节点释放空间后重试
        yield self.env.any_of([node.store.get_event() for node in self.nodes])
        yield from self.assign_entity(entity)

2. 修改移交逻辑，委托协调器分配实体：

def EntityHandover(node: Node) -> Generator:
    # 协调器可在Node初始化时创建，避免重复实例化
    coordinator = Coordinator(node.env, node.next)
    while True:
        with node.get() as item:
            entity = yield item
            # 确保同一实体仅被分配给一个节点
            yield from coordinator.assign_entity(entity)

优势：

• 彻底规避竞态条件，逻辑清晰；
• 无需修改SimPy源码，基于现有API实现；
• 可灵活扩展分配策略（如负载均衡、随机选择等）。

方案2：用互斥锁限制并发put请求 #

如果不想引入协调器，可使用SimPy的Resource作为互斥锁，确保同一时间仅处理一个put请求：

def EntityHandover(node: Node) -> Generator:
    # 容量为1的资源作为互斥锁
    mutex = node.env.Resource(1)
    while True:
        with node.get() as item:
            entity = yield item

            # 获取锁，确保同一时间仅执行一次分配操作
            with mutex.request():
                success = False
                for receiver in node.next:
                    try:
                        # 无超时尝试put，节点可用则立即成功
                        yield receiver.put(entity, timeout=0)
                        success = True
                        break
                    except simpy.Timeout:
                        continue
                # 所有节点不可用时，等待任意节点可用后再尝试
                if not success:
                    yield node.env.any_of([receiver.put(entity) for receiver in node.next])

通过timeout=0检测节点可用性，结合互斥锁避免多节点同时被判定为可用的场景。

方案3：自定义带优先级的Put事件 #

若要保留“同时发起请求但仅一个生效”的逻辑，可给put请求添加优先级，利用SimPy的PriorityEvent确保同一时刻仅最高优先级请求成功：

1. 扩展Store类支持优先级put：

class PriorityStore(simpy.Store):
    def put(self, item, priority=0):
        event = simpy.PriorityEvent(self.env, priority)
        if len(self.items) < self.capacity:
            self.items.append(item)
            event.succeed()
        else:
            self.put_queue.append((priority, event, item))
            self.put_queue.sort()  # 按优先级排序队列
        return event

2. 修改Node使用PriorityStore，分配时给节点设置唯一优先级：

def EntityHandover(node: Node) -> Generator:
    while True:
        with node.get() as item:
            entity = yield item

            # 按节点索引设置优先级，索引越小优先级越高
            receivers = [
                receiver.put(entity, priority=i) 
                for i, receiver in enumerate(node.next)
            ]

            done, _ = yield node.env.any_of(receivers)
            # 取消未完成的请求
            for rec in receivers:
                if not rec.processed:
                    rec.cancel()

这种方式无需修改SimPy核心，仅通过扩展Store类实现优先级控制，确保同一时刻仅一个put生效。

内容的提问来源于stack exchange，提问作者C Sheahan