使用 Container 或 store 的简单事件同步

Question

我有一个代码可以简单地实现调度。模拟以间隔并行运行处理器。此外，对于每个间隔，都有一个同步屏障等待所有处理器执行任务，然后移动到下一个间隔。 以下代码取自https://wso2.com/blog/research/modeling-closed-system-performance-of-a-server-with-discrete-event-simulation/ 该代码由一个客户端组成，该客户端将请求发送到由服务器（处理器）检查的输出队列。然后，服务器检查它们的队列并执行队列中的作业。这段代码的问题是没有同步；处理器不会互相等待。我需要一个统一的消息发送给所有的处理器，这样他们就可以互相等待。我正在考虑使用容器或商店，但似乎无法绕过它们。

例如，如果我运行 4 个处理器，每个处理器执行不同执行时间的作业（P1：4s，P2：3s，P3：2s，P4：1s）；处理器 1 (P1) 正在执行一个时长为 4 秒的作业。我如何添加同步屏障，以便它会中断处理器 P2:P4 直到 4 秒过去？

import random
import simpy 
SEED=42
average_processing_time=0.025
response_times=[]
queue_lengths=[]
waiting_times=[]

concurrenncy=4
num_cores=4



def client(env,out_pipe,in_pipe,i):
    global response_times
    while True:
        processing_time=random.expovariate(1/average_processing_time)
        arrival_time=env.now
        d={1:processing_time, 2:i , 3:arrival_time}
        out_pipe[i].put(d)
        #print('cliuent is processing the request %d' % i)
        response=yield in_pipe[i].get(filter=lambda x: True if x[2] == i else False)
        response_time=env.now-arrival_time
        response_times.append(response_time)
        

        
def server (env,in_pipe, out_pipe,i,channel):
    global queue_lengths 
    global waiting_times
    times=[]
    
        
    while True:
        request=yield in_pipe[i].get()
        #request_all=yield in_pipe.get()
        processing_time=request[1]
        arrival_time=request[3]
        waiting_time=env.now-arrival_time
        waiting_times.append(waiting_time)
        #for j in range(num_cores): 
         #  request_all=yield in_pipe[j].get()
            #times.append(request_all[1])
            
        queue_length=len(in_pipe[i].items)
        queue_lengths.append(queue_length)
        print('server %d is processing the request at time %f' % (i,env.now))
        #if max(times) > processing_time:
         #   new_t=max(times)
        #else:
         #   new_t=processing_time
        yield env.timeout(processing_time)
        channel.put(1)
        out_pipe[i].put(request)
        
random.seed(SEED)
in_pipe=[]
out_pipe=[]
p=[]
enviornment=simpy.Environment()
channel=simpy.Store(enviornment)
for i in range(num_cores):
    in_pipe.append(simpy.Store(enviornment))
    out_pipe.append(simpy.FilterStore(enviornment))
for i in range(concurrenncy):
    enviornment.process(client(enviornment,in_pipe,out_pipe,i))
    
for i in range(num_cores):
    t=enviornment.process(server(enviornment,in_pipe,out_pipe,i,channel))
    p.append(t)
enviornment.run(until=enviornment.all_of(p))

response_times=[x*100 for x in response_times]
waiting_times=[x*100 for x in waiting_times]
#print(waiting_times)
        

Answer 1

这是一个关于如何同步多个工作人员的快速示例，其中每个工作人员在开始下一个任务之前等待所有工作人员完成

"""
    Simple demo where workers wait for all workers to finish before starting next task

    Programer: Michael R. Gibbs
"""

import simpy
import random

class Worker():
    """
        simple worker
        waits for a start event
        does some work
        repeat
    """

    def __init__(self, env, id, controler):
        self.env = env
        self.id = id
        self.controler = controler
        self.next_done_event = simpy.Event(self.env)

        self.env.process(self.do_work())

    def do_work(self):
        """
            wait for controler's start event
            do some work
            trigger done event
            repeat
        """

        while True:

            yield self.controler.get_start_event()

            print(f'{self.env.now:.2f} worker {self.id} has started')

            yield self.env.timeout(random.uniform(1,4))

            print(f'{self.env.now:.2f} worker {self.id} has finish')

            self.next_done_event.succeed()

            self.next_done_event = simpy.Event(self.env)

    def get_next_done_event(self):
        """
            workers next done event
        """

        return self.next_done_event

class Controler():
    """
        tracks when all the workers have finished
        and broadcasts a done event
    """

    def __init__(self, env):
        
        self.env = env
        self.start_event = simpy.Event(self.env)

        self.workers = []

        self.env.process(self.monitor_workers())

    def add_worker(self, worker):
        """
            workers need to be addeded before sim is started
        """

        self.workers.append(worker)

    def get_start_event(self):
        """
            returns the controler next start event

            Note: that all the workers are gettng the same event
        """

        return self.start_event

    def monitor_workers(self):
        """
            trigger start event
            wait for all workers to finish
            repeat
        """

        # need this to insure all workers are set up
        yield self.env.timeout(1)

        while True:

            self.start_event.succeed()
            self.start_event = simpy.Event(self.env)

            done_events = [worker.get_next_done_event() for worker in self.workers]

            yield self.env.all_of(done_events)

            print(f'{self.env.now:.2f} all workers have finished')

# boot it up
env = simpy.Environment()

controler = Controler(env)

for i in range(3):
    worker = Worker(env, i+1, controler)
    controler.add_worker(worker)

env.run(100)
    
print('done')