Simpy：如何实现一次处理多个事件的资源
Simpy: How to implement a resource that handles multiple events at once
我正在模拟人的移动和他们使用电梯的情况。一部电梯在移动到另一层楼之前可以乘坐多人。默认进程有一个容量参数，但是，这些参数表示进程数，而不是同时使用电梯的人数。
我尝试使用多种可用资源，例如 Container、Store 和 Base。应请求电梯，但这些对象不具备请求的功能。因此，唯一合适的解决方案是从base.Resource继承class。我试图创建一个 subclass 电梯，从 base.Resource 实施并调整函数 _do_get 以从队列中获取多个元素。我非常有信心这不是实现它的正确方法，它也会给出一个错误：RuntimeError: <Request() object at 0x1ffb4474be0> has already been triggered。我不知道要调整哪些文件才能让 Simpy 满意。有人能给我指出正确的方向吗？
@dataclass
class Elevator(simpy.Resource):
current_floor: int = 0
available_floors: List[int] = field(default_factory=lambda: [0, 1])
capacity: int = 3
# load_carriers: List[LoadCarrier] = field(default_factory=list)
move_time: int = 5
def __init__(self, env: Environment, capacity: int = 1, elevator_capacity: int = 1):
self.elevator_capacity = elevator_capacity
if capacity <= 0:
raise ValueError('"capacity" must be > 0.')
super().__init__(env, capacity)
self.users: List[Request] = []
"""List of :class:`Request` events for the processes that are currently
using the resource."""
self.queue = self.put_queue
"""Queue of pending :class:`Request` events. Alias of
:attr:`~simpy.resources.base.BaseResource.put_queue`.
"""
@property
def count(self) -> int:
"""Number of users currently using the resource."""
return len(self.users)
if TYPE_CHECKING:
def request(self) -> Request:
"""Request a usage slot."""
return Request(self)
def release(self, request: Request) -> Release:
"""Release a usage slot."""
return Release(self, request)
else:
request = BoundClass(Request)
release = BoundClass(Release)
def _do_put(self, event: Request) -> None:
if len(self.users) < self.capacity:
self.users.append(event)
event.usage_since = self._env.now
event.succeed()
def _do_get(self, event: Release) -> None:
for i in range(min(self.elevator_capacity, len(self.users))):
try:
event = self.users.pop(0)
event.succeed()
# self.users.remove(event.request) # type: ignore
except ValueError:
pass
# event.succeed()
这是我想出的解决方案。棘手的一点是我将两个事件链接在一起。当您排队等待电梯时，您会收到一个事件，该事件会在电梯到达时触发。此事件也是 returns 当您到达目的地楼层时触发的第二个事件。这第二个事件是电梯上去往同一楼层的所有乘客共享的公共事件。触发这个事件会通知一群乘客。这种订阅广播模式可以大大减少模型需要处理的事件数量，从而提高性能。我使用链式事件是因为如果你在排队，而你前面的人上车而你没有上车，那么那个人也会在你之前下车，这需要一个不同的目的地到达事件。换句话说，我不知道你什么时候下车直到你上电梯，所以我需要把那部分推迟到你真正上电梯。
"""
Simple elevator demo using events to implements a subscribe, broadcast pattern to let passengers know when
they have reached there floor. All the passengers getting off on the same floor are waiting on the
same one event.
Programmer: Michael R. Gibbs
"""
import simpy
import random
class Passenger():
"""
simple class with unique id per passenger
"""
next_id = 1
@classmethod
def get_next_id(cls):
id = cls.next_id
cls.next_id += 1
return id
def __init__(self):
self.id = self.get_next_id()
class Elevator():
""""
Elevator that move people from floor to floor
Has a max compatity
Uses a event to notifiy passengers when they can get on the elevator
and when they arrive at their destination floor
"""
class Move_Goal():
"""
wrapps passengers so we can track where they are going to
"""
def __init__(self, passenger, start_floor, dest_floor, onboard_event):
self.passenger = passenger
self.start_floor = start_floor
self.dest_floor = dest_floor
self.onboard_event = onboard_event
self.arrive_event = None
def __init__(self,env, passenger_cap, floors):
self.env = env
self.passenger_cap = passenger_cap
self.floors = floors
self.on_floor = 0
self.move_inc = 1
# list of passengers on elevator, one per floor
self.on_board = {f:[] for f in range(1,floors + 1)}
# queue for passengers waitting to get on elevator, one queue per floor
self.boarding_queues = {f:[] for f in range(1,floors + 1)}
# events to notify passengers when they have arrived at their floor, one per floor
self.arrive_events = {f: simpy.Event(env) for f in range(1, floors + 1)}
# start elevator
env.process(self._move_next_floor())
def _move_next_floor(self):
"""
Moves the elevator up and down
Elevator stops at every floor
"""
while True:
# move time to next floor
yield self.env.timeout(5)
# update floor elevator is at
self.on_floor = self.on_floor + self.move_inc
# check if elevator needs to change direction
if self.on_floor == self.floors:
self.move_inc = -1
elif self.on_floor == 1:
self.move_inc = 1
# unload and notify passengers that want to get of at this floor
arrive_event = self.arrive_events[self.on_floor]
self.arrive_events[self.on_floor] = simpy.Event(self.env)
arrive_event.succeed()
self.on_board[self.on_floor] = []
# load new passengers
# get open capacity
used_cap = 0
for p in self.on_board.values():
used_cap += len(p)
open_cap = self.passenger_cap - used_cap
# get boarding passengers
boarding = self.boarding_queues[self.on_floor][:open_cap]
self.boarding_queues[self.on_floor] = self.boarding_queues[self.on_floor][open_cap:]
# sort bording into dest floors
for p in boarding:
# give passenger common event for arriving at destination floor
p.arrive_event = self.arrive_events[p.dest_floor]
# notify passeger that they are onboard the elevator
p.onboard_event.succeed()
self.on_board[p.dest_floor].append(p)
def move_to(self, passenger, from_floor, to_floor):
"""
Return a event that fires when the passenger gets on the elevator
The event returns another event that fires when the passager
arrives at their destination floor
(uses the env.process() to convert a process to a event)
"""
return self.env.process(self._move_to(passenger, from_floor, to_floor))
def _move_to(self, passenger, from_floor, to_floor):
"""
Puts the passenger into a queue for the elevator
"""
# creat event to notify passenger when they can get onto the elemator
onboard_event = simpy.Event(self.env)
# save move data in a wrapper and put passenger into queue
move_goal = self.Move_Goal(passenger, from_floor, to_floor, onboard_event)
self.boarding_queues[from_floor].append(move_goal)
# wait for elevator to arrive, and have space for passenger
yield onboard_event
# get destination arrival event
dest_event = self.arrive_events[to_floor]
move_goal.arrive_event = dest_event
return dest_event
def use_elevator(env, elevator, passenger, start_floor, end_floor):
"""
process for using a elevator to move from one floor to another
"""
print(f'{env.now:.2f} passenger {passenger.id} has queued on floor {start_floor}')
arrive_event = yield elevator.move_to(passenger, start_floor, end_floor)
print(f'{env.now:.2f} passenger {passenger.id} has boarded on floor {start_floor}')
yield arrive_event
print(f'{env.now:.2f} passenger {passenger.id} has arrived on floor {end_floor}')
def gen_passengers(env, elevator):
"""
creates passengers to use a elevatore
"""
floor_set = {f for f in range(1, elevator.floors + 1)}
while True:
# time between arrivals
yield env.timeout(random.uniform(0,5))
# get passenger and where they want to go
passenger = Passenger()
start_floor, end_floor = random.sample(floor_set, 2)
# use the elevator to get there
env.process(use_elevator(env, elevator, passenger, start_floor, end_floor))
# boot up
env = simpy.Environment()
elevator = Elevator(env, 20, 3)
env.process(gen_passengers(env, elevator))
env.run(100)
我正在模拟人的移动和他们使用电梯的情况。一部电梯在移动到另一层楼之前可以乘坐多人。默认进程有一个容量参数，但是，这些参数表示进程数，而不是同时使用电梯的人数。
我尝试使用多种可用资源，例如 Container、Store 和 Base。应请求电梯，但这些对象不具备请求的功能。因此，唯一合适的解决方案是从base.Resource继承class。我试图创建一个 subclass 电梯，从 base.Resource 实施并调整函数 _do_get 以从队列中获取多个元素。我非常有信心这不是实现它的正确方法，它也会给出一个错误：RuntimeError: <Request() object at 0x1ffb4474be0> has already been triggered。我不知道要调整哪些文件才能让 Simpy 满意。有人能给我指出正确的方向吗？
@dataclass
class Elevator(simpy.Resource):
current_floor: int = 0
available_floors: List[int] = field(default_factory=lambda: [0, 1])
capacity: int = 3
# load_carriers: List[LoadCarrier] = field(default_factory=list)
move_time: int = 5
def __init__(self, env: Environment, capacity: int = 1, elevator_capacity: int = 1):
self.elevator_capacity = elevator_capacity
if capacity <= 0:
raise ValueError('"capacity" must be > 0.')
super().__init__(env, capacity)
self.users: List[Request] = []
"""List of :class:`Request` events for the processes that are currently
using the resource."""
self.queue = self.put_queue
"""Queue of pending :class:`Request` events. Alias of
:attr:`~simpy.resources.base.BaseResource.put_queue`.
"""
@property
def count(self) -> int:
"""Number of users currently using the resource."""
return len(self.users)
if TYPE_CHECKING:
def request(self) -> Request:
"""Request a usage slot."""
return Request(self)
def release(self, request: Request) -> Release:
"""Release a usage slot."""
return Release(self, request)
else:
request = BoundClass(Request)
release = BoundClass(Release)
def _do_put(self, event: Request) -> None:
if len(self.users) < self.capacity:
self.users.append(event)
event.usage_since = self._env.now
event.succeed()
def _do_get(self, event: Release) -> None:
for i in range(min(self.elevator_capacity, len(self.users))):
try:
event = self.users.pop(0)
event.succeed()
# self.users.remove(event.request) # type: ignore
except ValueError:
pass
# event.succeed()
这是我想出的解决方案。棘手的一点是我将两个事件链接在一起。当您排队等待电梯时，您会收到一个事件，该事件会在电梯到达时触发。此事件也是 returns 当您到达目的地楼层时触发的第二个事件。这第二个事件是电梯上去往同一楼层的所有乘客共享的公共事件。触发这个事件会通知一群乘客。这种订阅广播模式可以大大减少模型需要处理的事件数量，从而提高性能。我使用链式事件是因为如果你在排队，而你前面的人上车而你没有上车，那么那个人也会在你之前下车，这需要一个不同的目的地到达事件。换句话说，我不知道你什么时候下车直到你上电梯，所以我需要把那部分推迟到你真正上电梯。
"""
Simple elevator demo using events to implements a subscribe, broadcast pattern to let passengers know when
they have reached there floor. All the passengers getting off on the same floor are waiting on the
same one event.
Programmer: Michael R. Gibbs
"""
import simpy
import random
class Passenger():
"""
simple class with unique id per passenger
"""
next_id = 1
@classmethod
def get_next_id(cls):
id = cls.next_id
cls.next_id += 1
return id
def __init__(self):
self.id = self.get_next_id()
class Elevator():
""""
Elevator that move people from floor to floor
Has a max compatity
Uses a event to notifiy passengers when they can get on the elevator
and when they arrive at their destination floor
"""
class Move_Goal():
"""
wrapps passengers so we can track where they are going to
"""
def __init__(self, passenger, start_floor, dest_floor, onboard_event):
self.passenger = passenger
self.start_floor = start_floor
self.dest_floor = dest_floor
self.onboard_event = onboard_event
self.arrive_event = None
def __init__(self,env, passenger_cap, floors):
self.env = env
self.passenger_cap = passenger_cap
self.floors = floors
self.on_floor = 0
self.move_inc = 1
# list of passengers on elevator, one per floor
self.on_board = {f:[] for f in range(1,floors + 1)}
# queue for passengers waitting to get on elevator, one queue per floor
self.boarding_queues = {f:[] for f in range(1,floors + 1)}
# events to notify passengers when they have arrived at their floor, one per floor
self.arrive_events = {f: simpy.Event(env) for f in range(1, floors + 1)}
# start elevator
env.process(self._move_next_floor())
def _move_next_floor(self):
"""
Moves the elevator up and down
Elevator stops at every floor
"""
while True:
# move time to next floor
yield self.env.timeout(5)
# update floor elevator is at
self.on_floor = self.on_floor + self.move_inc
# check if elevator needs to change direction
if self.on_floor == self.floors:
self.move_inc = -1
elif self.on_floor == 1:
self.move_inc = 1
# unload and notify passengers that want to get of at this floor
arrive_event = self.arrive_events[self.on_floor]
self.arrive_events[self.on_floor] = simpy.Event(self.env)
arrive_event.succeed()
self.on_board[self.on_floor] = []
# load new passengers
# get open capacity
used_cap = 0
for p in self.on_board.values():
used_cap += len(p)
open_cap = self.passenger_cap - used_cap
# get boarding passengers
boarding = self.boarding_queues[self.on_floor][:open_cap]
self.boarding_queues[self.on_floor] = self.boarding_queues[self.on_floor][open_cap:]
# sort bording into dest floors
for p in boarding:
# give passenger common event for arriving at destination floor
p.arrive_event = self.arrive_events[p.dest_floor]
# notify passeger that they are onboard the elevator
p.onboard_event.succeed()
self.on_board[p.dest_floor].append(p)
def move_to(self, passenger, from_floor, to_floor):
"""
Return a event that fires when the passenger gets on the elevator
The event returns another event that fires when the passager
arrives at their destination floor
(uses the env.process() to convert a process to a event)
"""
return self.env.process(self._move_to(passenger, from_floor, to_floor))
def _move_to(self, passenger, from_floor, to_floor):
"""
Puts the passenger into a queue for the elevator
"""
# creat event to notify passenger when they can get onto the elemator
onboard_event = simpy.Event(self.env)
# save move data in a wrapper and put passenger into queue
move_goal = self.Move_Goal(passenger, from_floor, to_floor, onboard_event)
self.boarding_queues[from_floor].append(move_goal)
# wait for elevator to arrive, and have space for passenger
yield onboard_event
# get destination arrival event
dest_event = self.arrive_events[to_floor]
move_goal.arrive_event = dest_event
return dest_event
def use_elevator(env, elevator, passenger, start_floor, end_floor):
"""
process for using a elevator to move from one floor to another
"""
print(f'{env.now:.2f} passenger {passenger.id} has queued on floor {start_floor}')
arrive_event = yield elevator.move_to(passenger, start_floor, end_floor)
print(f'{env.now:.2f} passenger {passenger.id} has boarded on floor {start_floor}')
yield arrive_event
print(f'{env.now:.2f} passenger {passenger.id} has arrived on floor {end_floor}')
def gen_passengers(env, elevator):
"""
creates passengers to use a elevatore
"""
floor_set = {f for f in range(1, elevator.floors + 1)}
while True:
# time between arrivals
yield env.timeout(random.uniform(0,5))
# get passenger and where they want to go
passenger = Passenger()
start_floor, end_floor = random.sample(floor_set, 2)
# use the elevator to get there
env.process(use_elevator(env, elevator, passenger, start_floor, end_floor))
# boot up
env = simpy.Environment()
elevator = Elevator(env, 20, 3)
env.process(gen_passengers(env, elevator))
env.run(100)