简单的红绿灯模拟

Question

我正在尝试用一系列交通信号灯依次模拟一条圆形道路。车辆通过泊松过程进入系统。一旦进入系统，他们就会在每个灯前排队。他们通过每个队列需要一个时间单位。当他们行过与其行程长度相等的队列数时，他们会退出系统。队列仅在绿色阶段运行。汽车用整数表示。

问题是我不断收到错误“从空双端队列弹出”，即使这部分代码只能通过检查双端队列中是否有汽车的 if 语句访问。我对 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，让调用进程运行，这可能会导致其他事件导致队列为空并引发你的异常。这有点像协作式多任务处理，但没有锁，所以你必须小心你放置yields 的位置。

将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)