這次文章是關(guān)于如何用 SimPy 來(lái)解決兩個(gè)仿真需求:

• 如何隨時(shí)中斷恢復(fù) Process (進(jìn)程)
• 如何動(dòng)態(tài)設(shè)置 Resource (資源)的數(shù)量

相應(yīng)地這兩個(gè)需求滿(mǎn)足的場(chǎng)景是:

• 仿真過(guò)程中, 某一工序被中斷, 中斷可以依據(jù)一個(gè)預(yù)先設(shè)定的時(shí)間或者是不確定時(shí)間
• 仿真過(guò)程中, 人力資源也是依據(jù)時(shí)間變化, 模擬現(xiàn)實(shí)中工人的排班安排

回顧資源和進(jìn)程的概念

Resource 和 Process 是 SimPy 對(duì)人力資源和進(jìn)程進(jìn)行抽象的構(gòu)造. Resource 好比一個(gè)隊(duì)列, 其長(zhǎng)度就是提前設(shè)置好的資源數(shù), 不同的工序就按照時(shí)間先后和賦予的優(yōu)先級(jí)進(jìn)入隊(duì)列. Process 從構(gòu)造上來(lái)說(shuō)就是一個(gè)生成器, 我們可以通過(guò) send 方法傳入 Exception 對(duì) Process 進(jìn)行打斷.

比如某個(gè)工序需要占用一個(gè)工人, 耗時(shí) 30 min 來(lái)完成一個(gè)進(jìn)程, 當(dāng)前所有可以調(diào)用的工人數(shù)是 10, 代碼形式如下:

import simpy
import random


WORKERNUM = 10 # 工人數(shù)
PROCESS_TIME = 30 * 60 # 工序耗時(shí), 使用秒作為單位
MEAN_  = 4 * 60 # 平均物件生成時(shí)間

def process(env, workers, store):
    """工序"""
    while True:
        with workers.request() as req:
            yield req
            item = yield store.get()
            print(f"{env.now} - {item} - start")
            yield env.timeout(PROCESS_TIME)
            print(f"{env.now} - {item} - done")

def put_item(env, store):
    """每隔一段時(shí)間生成一個(gè)物品"""
    for i in range(100):
        item = f"{i}_item"
        store.put(item)
        yield env.timeout(random.expovariate(1 / MEAN_))

env = simpy.Environment()
workers = simpy.Resource(env, 10)
store = simpy.Store(env)

env.process(process(env, workers, store))
env.process(put_item(env, store))
env.run()

更詳細(xì)的介紹和資料可以回顧之前的文章 Python SimPy 仿真系列 (1)

Process 進(jìn)程的動(dòng)態(tài)調(diào)整

存在以下兩種情景:

• 進(jìn)程隨時(shí)中斷以及恢復(fù)
• 按照時(shí)間表對(duì)進(jìn)程進(jìn)行啟動(dòng)或者終止

要區(qū)分一件事情, 中斷的時(shí)候是讓當(dāng)前進(jìn)程完成后再中斷, 還是立即中斷. 具體場(chǎng)景可以想象為一個(gè)工人被調(diào)離當(dāng)前崗位, 他應(yīng)該是先完成手頭上的工序, 或者他需要停下手頭的工作離開(kāi)工位.

如果是必須實(shí)現(xiàn)進(jìn)程的隨時(shí)中斷, 只能通過(guò) process.interrupt() 中斷 process, 即第一種場(chǎng)景; 假若中斷是按照時(shí)間表進(jìn)行, 就可以通過(guò)第二種場(chǎng)景, 構(gòu)建多個(gè)不同時(shí)間開(kāi)啟的進(jìn)程來(lái)進(jìn)行模擬.

進(jìn)程中斷的實(shí)現(xiàn)

from simpy import interrupt, Environment

env = Environment()

def interrupter(env, victim_proc):
    yield env.timeout(1)
    victim_proc.interrupt('Spam')

def victim(env):
    try:
        yield env.timeout(10)
     except Interrupt as interrupt:
         cause = interrupt.cause

多段進(jìn)程模擬按時(shí)間安排的開(kāi)關(guān)

import simpy

PROCESS_TIME = 3

def put_item(env, store):
    for i in range(20):
        yield env.timeout(0.5)
        store.put(f"{i}_item")

def process(i, env, store, start, end):

    yield env.timeout(start)
    while True:
        item = yield store.get()
        # 判斷 item 到達(dá)時(shí)間是否超出本進(jìn)程關(guān)閉時(shí)間
        if env.now > end:
            print(f"{env.now} - process {i} - end")
            store.put(item)
            env.exit()
        else:
            print(f"{env.now} - {item} - start")
            yield env.timeout(PROCESS_TIME)
            print(f"{env.now} - {item} - end")

env = simpy.Environment()
store = simpy.Store(env)

env.process(put_item(env, store))

for i, (start, end) in enumerate([(20, 30), (40, 50), (60, 90)]):
    env.process(process(i, env, store, start, end))

env.run()

Resource 資源的動(dòng)態(tài)調(diào)整

• 資源人數(shù)按指定的排版表調(diào)配

由于Resource 在實(shí)例化后, 就沒(méi)辦法修改了. 為了滿(mǎn)足在仿真過(guò)程中對(duì)資源進(jìn)行修改, 使用了一個(gè)反向的思路. 首先所有資源使用 PriorityResource 實(shí)例, 預(yù)先設(shè)置一個(gè)可以調(diào)節(jié)的最大資源數(shù), 當(dāng)需要調(diào)節(jié)資源數(shù)的時(shí)候, 使用一個(gè)優(yōu)先級(jí)為 -1 的 request 去占用資源, 而正常的進(jìn)程默認(rèn)優(yōu)先級(jí)是 0.

通過(guò)這樣的操作會(huì)使得, 我們調(diào)節(jié)資源的占用進(jìn)程優(yōu)先級(jí)更高, 正常進(jìn)程可以調(diào)用的資源數(shù)會(huì)變成
:

可以調(diào)用資源 = 最大資源 - 占用資源

import simpy

PROCESS_TIME = 2

def put_item(env, store):
    for i in range(20):
        yield env.timeout(0.5)
        store.put(f"{i}_item")

def process(env, store, resource):
    while True:
        item = yield store.get()
        with resource.request() as req:
            yield req
            yield env.timeout(PROCESS_TIME)

def set_resource(env, resource, start_time, end_time):
    """占用資源，模擬資源減少的情況，
    end_time 會(huì)出現(xiàn) np.inf 無(wú)窮大，
    simpy 只會(huì)用作為排序，可以放在timeout事件里。
    """
    duration = end_time - start_time
    yield env.timeout(start_time)
    with resource.request(priority=-1) as req:
        yield req
        yield env.timeout(duration)

env = simpy.Environment()
store = simpy.Store(env)
res = simpy.PriorityResource(env, 10)

res_time_table = [(10, 20, 5), (20, 30, 6)]
env.process(put_item(env, store))
env.process(process(env, store, res))

for start, end, target_num in res_time_table:
    place_holder = 10 - target_num
    for _ in range(place_holder):
        env.process(set_resource(env, res, start, end))

env.run()