Simpy离散事件仿真问题：下游队列满时服务器阻塞致请求无法完成

问题描述 #

我已为编写该离散事件仿真代码耗费数日，此前在本站获取过部分解答，但仿真仍无法正常运行。出现的问题为：首个客户（及所有后续客户）可通过首个服务的队列，但始终无法实际进入服务器1。我推测这可能是一处低级错误，希望能快速定位问题。客户可通过第0个服务器并进入服务器1的队列，我已在疑似错误行前后添加####标记。

原代码 #

import simpy
import numpy as np

class Entity(object):
    pass

def process0(env, entity, process_0_res, process_1_q):
    print(f' {env.now} customer {entity.id} is in system')
    with process_0_res.request() as res_req:
        yield res_req
        print(f'{env.now} customer {entity.id} is in queue 1')
        yield process_1_q.put(entity)

def process1_broker(env, process_1_q, process_1_res):
    while True:
        # is resource available?
        res_req = process_1_res.request()
        yield res_req
        # is customer available?
        entity = yield process_1_q.get()
        # save resource request to release later
        entity.res_req = res_req
        # start process
        env.process(process1(env,entity,process_1_res, process_2_q))

def process1(env, entity, process_1_res, process_2_q):
    with process_1_res.request() as res_req:
        print(f'{env.now} customer {entity.id} should now request server 1')
        #########
        yield res_req
        #########
        print(f' {env.now} customer {entity.id} in process 1')
        yield env.timeout(2)
        yield process_2_q.put(entity)

def process2_broker(env, process_2_q, process_2_res):
    while True:
        res_req = process_2_res.request()
        yield res_req
        entity = yield process_2_q.get()
        entity.res_req = res_req
        env.process(process2(env,entity,process_2_res, process_3_q))

def process2(env, entity, process_2_res, process_3_q):
    print(f' {env.now} customer {entity.id} in process 2')
    with process_2_res.request() as res_req:
        yield res_req
        yield env.timeout(np.random.exponential(mu[1]))
        yield process_3_q.put(entity)
        
def process3_broker(env, process_3_q, process_3_res):
    while True:
        res_req = process_3_res.request()
        yield res_req
        entity = yield process_3_q.get()
        entity.res_req = res_req
        env.process(process3(env,entity,process_3_res, process_4_q))

def process3(env, entity, process_3_res, process_4_q):
    print(f' {env.now} customer {entity.id} in process 3')
    with process_3_res.request() as res_req:
        yield res_req
        yield env.timeout(np.random.exponential(mu[2]))
        yield process_4_q.put(entity)

def process4_broker(env, process_4_q, process_4_res):
    while True:
        res_req = process_4_res.request()
        yield res_req
        entity = yield process_3_q.get()
        entity.res_req = res_req
        env.process(process4(env,entity,process_4_res))

def process4(env, entity, process_4_res):
    print(f' {env.now} customer {entity.id} in process 4')
    with process_4_res.request() as res_req:
        yield res_req
        yield env.timeout(np.random.exponential(mu[3]))
        yield process_4_res.release(entity.res_req)
        print(f' {env.now} customer {entity.id} leaves system')
        
def gen_entities(env, process_0_res, process_1_q):
    next_id = 1
    while True:
        yield env.timeout(np.random.exponential(labda))
        entity = Entity()
        entity.id = next_id
        next_id += 1
        env.process(process0(env, entity, process_0_res, process_1_q))

# 补充未定义的全局变量
labda = 0.5
mu = [1.0, 1.0, 1.0, 1.0]

env = simpy.Environment()
process_0_res = simpy.Resource(env, capacity = 1)
process_1_res = simpy.Resource(env, capacity = 1)
process_2_res = simpy.Resource(env, capacity = 1)
process_3_res = simpy.Resource(env, capacity = 1)
process_4_res = simpy.Resource(env, capacity = 1)

process_1_q = simpy.Store(env, capacity = 5)
process_2_q = simpy.Store(env, capacity = 4)
process_3_q = simpy.Store(env, capacity = 3)
process_4_q = simpy.Store(env, capacity = 2)

env.process(gen_entities(env, process_0_res, process_1_q))

env.process(process1_broker(env, process_1_q, process_1_res))
env.process(process2_broker(env, process_2_q, process_2_res))
env.process(process3_broker(env, process_3_q, process_3_res))
env.process(process4_broker(env, process_4_q, process_4_res)) 

env.run(10)

问题定位与修复 #

核心错误：重复请求资源导致死锁 #

process1_broker中已经提前请求并持有了process_1_res资源（通过res_req = process_1_res.request() + yield res_req），但process1函数里又再次通过with process_1_res.request() as res_req请求同一个资源。由于process_1_res容量为1，已被broker占用，process1永远无法获取资源，导致客户卡在yield res_req处，无法进入服务器1处理。

修复方案 #

删除process1中的重复资源请求逻辑，直接使用broker预获取的entity.res_req，处理完成后释放该资源：

def process1(env, entity, process_1_res, process_2_q):
    print(f'{env.now} customer {entity.id} should now request server 1')
    print(f' {env.now} customer {entity.id} in process 1')
    yield env.timeout(2)
    yield process_2_q.put(entity)
    # 释放broker提前获取的资源
    yield process_1_res.release(entity.res_req)

其他需修复的问题 #

1. process4_broker队列错误：entity = yield process_3_q.get()需改为entity = yield process_4_q.get()，否则会从错误队列取客户。
2. 后续process函数的重复请求问题：process2、process3、process4均存在相同的重复请求逻辑，需按process1的修复方式调整，以process2为例：

   def process2(env, entity, process_2_res, process_3_q):
       print(f' {env.now} customer {entity.id} in process 2')
       yield env.timeout(np.random.exponential(mu[1]))
       yield process_3_q.put(entity)
       yield process_2_res.release(entity.res_req)
   

内容的提问来源于stack exchange，提问作者Barend Kok