Simpy资源时段不可用设置失效及AttributeError报错求助

问题定位 #

• 报错AttributeError: 'int' object has no attribute 'callbacks'的根因：delay函数工作时段分支直接yield了整数数值，SimPy要求所有yield的对象必须是事件实例，需要用env.timeout()包裹时间数值。
• 资源在非工作时段仍可处理请求的根因：
  1. 采用普通Resource类无法强制抢占容量，你发起的占满容量的请求需要等待现有任务全部完成才能生效，无法实现到点立刻锁定资源的效果
  2. 工作时段判断逻辑存在疏漏，第二个条件没有对1440取模，跨天后时间判断会失效
  3. 占满资源的请求未设置优先级，无法比用户请求优先抢占容量

修复后的完整代码 #

import numpy as np
import simpy
from simpy import PreemptiveResource

def interarrival():
    return np.random.exponential(10)

def servicetime():
    return np.random.exponential(20)


def servicing(env, servers_array):
    i = 0
    while True:
        i += 1
        yield env.timeout(interarrival())
        print(f"Customer {i} arrived in the process at {env.now}")
        state = 0
        env.process(items(env, i, servers_array, state))

def items(env, customer_id, servers_array, state):
    # 用户请求设置低优先级，确保非工作时段锁资源的请求可以优先抢占
    with servers_array[state].request(priority=1) as request:
        yield request
        t_arrival = env.now
        print(f"Customer {customer_id} arrived in {state} at {t_arrival}")
        try:
            yield env.timeout(servicetime())
            t_depart = env.now
            print(f"Customer {customer_id} departed from {state} at {t_depart}")
            if state == 1:
                print(f"Customer {customer_id} exits")
            else:
                state = 1
                env.process(items(env, customer_id, servers_array, state))
        except simpy.Interrupt:
            # 被非工作时段抢占的任务，可自行补充重试逻辑，比如放回队列等待下一工作日
            print(f"Customer {customer_id} in node {state} is interrupted at {env.now}, will resume next workday")

def delay(env, servers_array):
    while True:
        now_mod = env.now % 1440
        if 540 <= now_mod <= 1080:
            # 工作时段，等到18点触发锁资源逻辑
            yield env.timeout(1080 - now_mod)
        else:
            print(f"{env.now}: resources will be blocked")
            resource_unavailability_dict = {0: [], 1: []}
            # 锁资源的请求设置最高优先级，优先占满所有容量
            for node in resource_unavailability_dict:
                for _ in range(servers_array[node].capacity):
                    resource_unavailability_dict[node].append(servers_array[node].request(priority=0, preempt=True))
            # 等待所有锁资源的请求都拿到容量
            for node in resource_unavailability_dict:
                yield env.all_of(resource_unavailability_dict[node])
            # 锁定到下一个工作日9点
            if now_mod < 540:
                lock_duration = 540 - now_mod
            else:
                lock_duration = (1440 - now_mod) + 540
            yield env.timeout(lock_duration)
            # 释放所有锁资源的请求
            for node in resource_unavailability_dict:
                for req in resource_unavailability_dict[node]:
                    servers_array[node].release(req)
            print(f"{env.now}: resources are released")
            

env = simpy.Environment()
servers_array = []
# 替换为支持抢占的PreemptiveResource
servers_array.append(PreemptiveResource(env, capacity=5))
servers_array.append(PreemptiveResource(env, capacity=7))
env.process(servicing(env, servers_array))
env.process(delay(env, servers_array))
env.run(until=2880)

核心修改说明 #

• 将普通Resource替换为PreemptiveResource，支持高优先级请求抢占资源，实现到点立刻锁定的效果
• 给锁资源的请求设置更高优先级（priority=0，数值越小优先级越高），用户请求设置低优先级priority=1，确保锁资源的请求可以优先占满所有容量
• 修复时间判断逻辑，统一先计算now_mod = env.now%1440，避免跨天判断错误
• 修复报错问题，所有yield的时间值都用env.timeout()包裹
• 新增任务中断异常处理，非工作时段被抢占的任务可以根据业务需求补充重试逻辑

内容的提问来源于stack exchange，提问作者Coding Practice