Python SimPy 模擬系列 (2)

04-02

這次文章是關於如何用SimPy來解決兩個模擬需求:

如何隨時中斷恢復Process(進程)
如何動態設置Resource(資源)的數量

相應地這兩個需求滿足的場景是:

模擬過程中, 某一工序被中斷, 中斷可以依據一個預先設定的時間或者是不確定時間
模擬過程中, 人力資源也是依據時間變化, 模擬現實中工人的排班安排

回顧資源和進程的概念

Resource和Process是 SimPy 對人力資源和進程進行抽象的構造.Resource好比一個隊列, 其長度就是提前設置好的資源數, 不同的工序就按照時間先後和賦予的優先順序進入隊列.Process從構造上來說就是一個生成器, 我們可以通過 send 方法傳入Exception對Process進行打斷.

比如某個工序需要佔用一個工人, 耗時 30 min 來完成一個進程, 當前所有可以調用的工人數是 10, 代碼形式如下:

import simpyimport randomWORKERNUM = 10 # 工人數PROCESS_TIME = 30 * 60 # 工序耗時, 使用秒作為單位MEAN_ = 4 * 60 # 平均物件生成時間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): """每隔一段時間生成一個物品""" 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()

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

`Process`進程的動態調整

存在以下兩種情景:

進程隨時中斷以及恢復
按照時間表對進程進行啟動或者終止

要區分一件事情, 中斷的時候是讓當前進程完成後再中斷, 還是立即中斷. 具體場景可以想像為一個工人被調離當前崗位, 他應該是先完成手頭上的工序, 或者他需要停下手頭的工作離開工位.

如果是必須實現進程的隨時中斷, 只能通過process.interrupt()中斷process, 即第一種場景; 假若中斷是按照時間表進行, 就可以通過第二種場景, 構建多個不同時間開啟的進程來進行模擬.

進程中斷的實現

from simpy import interrupt, Environmentenv = 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

多段進程模擬按時間安排的開關

import simpyPROCESS_TIME = 3def 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 到達時間是否超出本進程關閉時間 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`資源的動態調整

資源人數按指定的排版表調配

由於Resource在實例化後, 就沒辦法修改了. 為了滿足在模擬過程中對資源進行修改, 使用了一個反向的思路. 首先所有資源使用PriorityResource進行實例, 預先設置一個可以調用的最大資源數, 當需要調節資源數的時候, 使用一個優先順序為-1的request去佔用資源, 由於正常的進程默認優先順序是0, 通過這樣的操作使得,佔用資源的佔用進程優先順序更高, 正常進程可以調用的資源數會變成:

可以調用資源=最大資源-佔用資源

import simpyPROCESS_TIME = 2def 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 會出現 np.inf 無窮大， simpy 只會用作為排序，可以放在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()

本文作者：

趙鵬 @Pong