批处理系统中中断式服务器启动与延迟关机的仿真问题

需要构建一个带启动时间的服务器系统，规则如下：

• 队列累计满2个客户时启动服务器
• 一批（2个客户）处理完成后，若仍有待处理批次，服务器保持活跃并立即处理新批次，同时中断正在启动的服务器
• 若无待处理批次，服务器进入延迟关机模式；若此期间有新批次到达，中断延迟过程并立即处理该批次

基于Michael R. Gibbs的无延迟关机版本代码，添加延迟关机逻辑后出现问题：被中断启动的服务器未按预期立即关闭，反而进入延迟关机流程。

import simpy
import random
# import numpy as np

class param:
    def __init__(self, x):
        #self.FILE = 'Setup_time.csv'
        self.MEAN_INTERARRIVAL = 1     # arrival rate
        self.MEAN_SERVICE_TIME = 1      # service rate
        self.MEAN_DELAY_TIME = 3        # delayoff 
        self.MEAN_SWITCH_TIME =  x      # setup rate
        self.NUM_OF_SERVER = 20          # maximum number of servers can be setup
        self.MAX_SYS_SIZE = 5000000           # maximum number of customers in the system
        self.BATCH_SIZE = 2  
        self.RANDOM_SEED = 0 
        self.DURATION_TIME = 12
        

# there is no wating so normal lists are good enough

class Server():
    """
    Server that process batches

    Has two states: starting up, and batch processing
    """
    
    def __init__(self, id, env, processing_q, server_q, delayoff_q, param):

        self.id = id
        self.env = env
        self.processing_q = processing_q
        self.server_q = server_q

        self.start_process = self.env.process(self.start_up(param))
        # self.delay_process = self.env.process(self.delayoff(param))

        global num_servers
        # sever has started, but not active 
        num_servers +=1
        
    def delayoff(self, param):
        """
        keep server in idle mode before turning it off (delayoff)

        delayoff can be interrupted if any batch arrives, then reactive to serve a batch.
        """
        global num_servers 
        
        try:   
            print(f'{env.now} server {self.id} start delayoff, delayoff servers:{len(delayoff_q)}, setup servers:{len(server_q)}')
            yield self.env.timeout(param.MEAN_DELAY_TIME)   
            # delayoff_q.remove(self)
            print(f'{env.now} server {self.id} shutdown after delayoff, delayoff servers:{len(delayoff_q)}, setup servers:{len(server_q)}')
        except simpy.Interrupt:

            print(f'{self.env.now} server{self.id} is interupted (idle server steal batch), # batches in queue: {len(processing_q)}')

        

    def start_up(self, param):
        """
        starts up the server, then start processing batches

        start up can be interrupted, stoping the server
        """
        global num_servers,num_active_server
        
         # start up
        state_time_trans.append(self.env.now)
        num_server_state.append(len(server_q))
                
        try:  
            print(f'{self.env.now} server {self.id} start to setup, num server setting up: {len(server_q)}')
            
            
            yield self.env.timeout(1/param.MEAN_SWITCH_TIME)
            # yield self.env.timeout(np.random.exponential(1/param.MEAN_SWITCH_TIME))
            
            # server has started, need to remove from startup q
            self.server_q.remove(self)
            print(f'{self.env.now} server {self.id} active and start processing, num server setting up: {len(server_q)}')
            state_time_trans.append(env.now)
            num_server_state.append(len(server_q))
                        
            
            self.env.process(self.process(param))
                    
        except simpy.Interrupt:
            
            print(f'{env.now} server {self.id} has been interupted, num server setting up: {len(server_q)}-------------------')
            state_time_trans.append(env.now)
            num_server_state.append(len(server_q))
            
            # server is stoping, need to adjust server count
            num_servers -= 1
            

    def process(self, param):
        """
        process batches
        keeps going as long as there are batches in queue

        If starts second batch, also interupts starting up server
        """
        global num_servers, num_active_server
        # count server becoming ative only once
        num_active_server += 1
        
        #Save the time a server active
        state_server_active.append(num_active_server)
        time_active.append(env.now)
        #print(f'{self.env.now} server {self.id} active, num server active: {num_active_server}')

        while True:
            # server becomes active only once, this is counting batches processed
            # num_active_server += 1

            b = processing_q.pop(0)
            
               
            yield self.env.timeout(param.MEAN_SERVICE_TIME)
            # yield env.timeout(np.random.exponential(1/param.MEAN_SERVICE_TIME))
            
            for i in range(0, len(b)):
                b[i].serve_time = env.now
                dic_serve_time.update({b[i].name: b[i].serve_time})
            print(f'{env.now} server {self.id} finish to serve customer {b[1].name},  {b[0].name}')
            # Server is still running

            if len(self.processing_q) > 0:
                # more processes to do,
                # steal batch from starting up server

                #s = self.server_q.pop() # lifo
                #s.stop()

                # need to check the length, not if it exists
                if len(self.server_q) >= len(self.processing_q):
                #if len(self.server_q) > 0:
                    s = self.server_q.pop()   #fifo
                    s.stop()
                    print(f'{env.now} server {self.id} continue to take a new batch')
                    

            # else:
            #     print(f'{env.now} server {self.id} no more batches, shut down')
            #     break
            
            else:
                # now the server is idle
                num_active_server -= 1
                delayoff_q.append(self)
                self.env.process(self.delayoff(param))
                print(f'{env.now} server {self.id} delayoff, have {len(delayoff_q)} idle servers')
                break
        
        # now the server is shutting down
        #num_active_server -= 1
        num_servers -= 1
        
        state_server_active.append(num_active_server)
        time_active.append(env.now)
        #print(f'{self.env.now} server {self.id} active, num server active: {num_active_server}')


    def stop(self):
        """
        Interrupts server start up, stoping server
        """
        try:
            self.start_process.interrupt()
        except:
            pass

def gen_arrivals(env, batching_q, processing_q, server_q, delayoff_q, param):
    """
    Generate arring customers

    If queues are too big customer will abort

    If have enough customers, create a batch and start a server
    """
    global num_servers, num_balk, num_cumulative_customer, num_active_server 
    id = 1
    name = 1
    while True:
        yield env.timeout(param.MEAN_INTERARRIVAL)
        # yield env.timeout(np.random.exponential(1/param.MEAN_INTERARRIVAL))
        num_cumulative_customer += 1

        if name < 10000000000:
            customer = Customer(name)
            batching_q.append(customer) 
            name += 1

        q_size  = len(batching_q) + (param.BATCH_SIZE * len(processing_q)) 
        sys_size = q_size + (num_active_server * param.BATCH_SIZE) 

        #if q_size > max_q_size:
        if  sys_size > param.MAX_SYS_SIZE:    
            num_balk += 1

            batching_q.pop(-1)                
            #print(f'{env.now} customer {customer.name} arrived and aborted, sys len: {sys_size }')

        else:
            #customer = object()                        
            #batching_q.append(customer)

            print(f'{env.now} customer{customer.name} has arrived, q len: {q_size}, sys len: {sys_size}')
            customer.arrival_time = env.now
            dic_arrival_time.update({customer.name: customer.arrival_time})
            
            # check if a batch can be creatd
            while len(batching_q) >= param.BATCH_SIZE:
                batch = list()
                while len(batch) < param.BATCH_SIZE:
                    batch.append(batching_q.pop(0))

                # put batch in processing q
                processing_q.append(batch)
                
                print(f'{env.now} new batch arrived, #delay: {len(delayoff_q)}')
   
                if len(delayoff_q) == 0:   #there is no idle server, need to setup a server
                    if num_servers < param.NUM_OF_SERVER:   
                        server = Server(id, env, processing_q, server_q, delayoff_q, param)
                        id += 1
                        server_q.append(server)
                        # print(f'{env.now} setup a new server {server.id}')
                else: 
                    s_delay = delayoff_q.pop(0)
                    env.process(s_delay.delayoff(param)).interrupt()
                    env.process(s_delay.process(param))

        #Calculate balking probability
        prob_balk = num_balk/num_cumulative_customer
        #print(f'{env.now} prob_balk {prob_balk}')
        list_prob_balk.append(prob_balk)
        
        
class Customer:                        
    def __init__(self, name):
        self.name = name
        self.arrival_time = 0
        self.serve_time = 0
        self.leave_time = 0
        

# boot up sim

for x in [1]:
    
    paramtest1 = param(x)
    random.seed(paramtest1.RANDOM_SEED)
    
    #Save waiting time
    dic_serve_time = {}
    dic_arrival_time = {}
    list_waiting_time =[]

    batching_q = list()
    processing_q = list()
    server_q = list()                   # servers that are still starting up
    delayoff_q = list()
    num_servers = 0                     # number of server in system (both starting and serving server)
    num_active_server = 0               # number of servers serving customers
    
    #Save balking customers
    num_balk = 0                        # number of balking customers        
    num_cumulative_customer = 0         # total arriving customers  
    list_prob_balk = []    #list balk prob each trial
    num_start =0
    
    #Save average start up server
    num_server_state = []
    state_time_trans = []
    pi_ij = []
    
    #Save average active server E[A]
    state_server_active = []  #number server active each time
    time_active = []   #the time a server active
    delta = []   #the time between each setting up or distance between term in time_active list
    
    # create and start the model
    env = simpy.Environment()
    env.process(gen_arrivals(env, batching_q, processing_q, server_q, delayoff_q, paramtest1))
    env.run(paramtest1.DURATION_TIME)

核心问题根源 #

被中断启动的服务器未从server_q中移除，导致后续可能被错误流转到延迟关机流程；同时队列状态与服务器计数的同步逻辑存在错误。

具体修复点 #

1. 启动中断时的队列清理
   在start_up方法的异常处理块中添加self.server_q.remove(self)，确保被中断的服务器立即从启动队列中移除，避免后续被错误处理。
2. 延迟关机流程的状态同步
   • 在Server类初始化时新增self.delayoff_q引用，统一队列操作的上下文
   • 在delayoff方法中，无论正常完成还是被中断，都从delayoff_q中移除当前服务器，保证队列状态准确
   • 将num_servers -=1移至delayoff方法的正常完成分支中，修正原代码中提前减少服务器计数的错误
3. 激活延迟服务器的逻辑优化
   在gen_arrivals中激活延迟服务器时，检查延迟流程是否存活并中断，避免重复启动处理流程
4. 时间引用统一
   替换代码中直接使用env.now的部分为self.env.now，保证时间获取的一致性

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