simpy红绿灯模拟

Question

我正在尝试模拟一条环形道路，其中包含一系列按顺序排列的红绿灯。车辆通过泊松过程进入系统。一旦进入系统，他们就会在每个红绿灯处排队。它们需要一个时间单位来遍历每个队列。当他们经过的队列数量与他们的行程长度相等时，他们就会离开系统。队列仅在绿色阶段运行。汽车是用整数表示的。

问题是，我一直收到错误"pop from a only deque“，即使这部分代码只能通过一个if语句来检查deque中是否有汽车。我对simpy不是很熟悉，所以我认为问题与超时有关。如果我在弹出操作后移动超时，代码将正常工作。但这并不是我想要的。

import simpy
from simpy.util import start_delayed
# import numpy.random
from collections import deque,namedtuple
from numpy import random
NUM_INT = 3
ARRIVAL_TIME_MEAN = 1.1
TRIP_LENGTH = 4
GREEN_TIME = 3.0
RED_TIME = 3.0

class Simulation(object):
    def __init__(self,env):
        self.env = env
        self.intersections = [Intersection(env,i) for i in range(NUM_INT)]
        for (i,intersection) in enumerate(self.intersections):
            intersection.set_next_intersection(self.intersections[(i+1)%NUM_INT])
        self.env.process(self.light())
        self.env.process(self.arrivals())

    def arrivals(self):
        while True:
            yield self.env.timeout(random.exponential(ARRIVAL_TIME_MEAN))
            intersection = random.choice(self.intersections)
            intersection.receive(TRIP_LENGTH)

    def light(self):
        while True:
            for intersection in self.intersections:
                intersection.start_departing()
            yield self.env.timeout(GREEN_TIME)
            for intersection in self.intersections:
                intersection.turn_red()
            yield env.timeout(RED_TIME)


class Intersection(object):
    def __init__(self,env,index):
        self.index = index
        self.queue = deque()
        self.env = env
        self.start_departing()

    def set_next_intersection(self,intersection):
        self.next_intersection = intersection

    def start_departing(self):
        self.is_departing = True
        self.action = env.process(self.departure())

    def turn_red(self):
        if self.is_departing:
            self.is_departing = False
            self.action.interrupt('red light')

    def receive(self,car):
        self.queue.append(car)
        if not self.is_departing:
            self.start_departing()

    def departure(self):
        while True:
            try:
                if len(self.queue)==0:
                    self.is_departing = False
                    self.env.exit('no more cars in %d'%self.index)
                else:
                    yield self.env.timeout(1.0)
                    car = self.queue.popleft()
                    car = car - 1
                    if car > 0:
                        self.next_intersection.receive(car)
            except simpy.Interrupt as i:
                print('interrupted by',i.cause)

env = simpy.Environment()
sim = Simulation(env)
env.run(until=15.0)

Answer 1

与其说是修复，不如说是变通。

import simpy
from simpy.util import start_delayed
# import numpy.random
from collections import deque,namedtuple
from numpy import random
NUM_INT = 3
ARRIVAL_TIME_MEAN = 1.1
TRIP_LENGTH = 4
GREEN_TIME = 3.0
RED_TIME = 3.0

class Simulation(object):
    def __init__(self,env):
        self.env = env
        self.intersections = [Intersection(env,i) for i in range(NUM_INT)]
        for (i,intersection) in enumerate(self.intersections):
            intersection.set_next_intersection(self.intersections[(i+1)%NUM_INT])
        self.env.process(self.light())
        self.env.process(self.arrivals())

    def arrivals(self):
        while True:
            yield self.env.timeout(random.exponential(ARRIVAL_TIME_MEAN))
            intersection = random.choice(self.intersections)
            intersection.receive(TRIP_LENGTH)

    def light(self):
        while True:
            for intersection in self.intersections:
                intersection.start_departing()
            yield self.env.timeout(GREEN_TIME)
            for intersection in self.intersections:
                intersection.turn_red()
            yield env.timeout(RED_TIME)


class Intersection(object):
    def __init__(self,env,index):
        self.index = index
        self.queue = deque()
        self.env = env
        self.start_departing()

    def set_next_intersection(self,intersection):
        self.next_intersection = intersection

    def start_departing(self):
        self.is_departing = True
        self.action = env.process(self.departure())

    def turn_red(self):
        if self.is_departing:
            self.is_departing = False
            self.action.interrupt('red light')

    def receive(self,car):
        self.queue.append(car)
        if not self.is_departing:
            self.start_departing()

    def departure(self):
        while True:
            try:
                if len(self.queue)==0:
                    self.is_departing = False
                    self.env.exit('no more cars in %d'%self.index)
                else:
                    yield self.env.timeout(1.0)
                    if len(self.queue)>0:
                        if len(self.queue)==1:
                            car=self.queue[0]
                            self.queue.clear()
                        else:
                            car = self.queue.popleft()
                        car = car - 1
                        if car > 0:
                            self.next_intersection.receive(car)
            except simpy.Interrupt as i:
                print('interrupted by',i.cause)

env = simpy.Environment()
sim = Simulation(env)
env.run(until=15.0)

告诉我这对你是否有效

Answer 2

在departure中，在队列不为空的情况下，您会立即放弃，这会让调用进程运行，这可能会导致其他事件导致队列为空并引发异常。这有点像协作多任务，但没有锁，所以你必须小心放置yield的位置。

将yield移到该if的末尾可以帮我修复它。

    def departure(self):
        while True:
            try:
                if len(self.queue)==0:
                    self.is_departing = False
                    self.env.exit('no more cars in %d'%self.index)
                else:
                    car = self.queue.popleft()
                    car = car - 1
                    if car > 0:
                        self.next_intersection.receive(car)
                    yield self.env.timeout(1.0)
            except simpy.Interrupt as i:
                print('interrupted by',i.cause)
                yield self.env.timeout(1.0)