Python多处理 - 是否可以在各个进程之间引入固定的时间延迟？

Question

我已经搜索过，但在其他地方找不到这个问题的答案。希望我没有错过任何东西。

我正在尝试使用 Python 多处理来并行批量运行一些专有模型。比如说，我有 200 次模拟，我想一次批量运行 10-20 次。我的问题是，如果两个模型碰巧在相同/相似的时间启动，专有软件就会崩溃。我需要在多处理产生的进程之间引入延迟，以便每个新模型运行在开始前等待一点。

到目前为止，我的解决方案是在子进程启动模型运行之前引入随机时间延迟。但是，这只会降低任意两次运行同时开始的概率，因此在尝试处理大量模型时仍然会遇到问题。因此，我认为时间延迟需要内置到代码的多处理部分中，但我无法找到任何文档或示例。

编辑：我使用的是 Python 2.7

这是我目前的代码：

from time import sleep
import numpy as np
import subprocess
import multiprocessing

def runmodels(arg):
    sleep(np.random.rand(1,1)*120) # this is my interim solution to reduce the probability that any two runs start at the same time, but it isn't a guaranteed solution
    subprocess.call(arg) # this line actually fires off the model run

if __name__ == '__main__':    

    arguments =     [big list of runs in here
                    ]    

    count = 12
    pool = multiprocessing.Pool(processes = count)
    r = pool.imap_unordered(runmodels, arguments)      
    pool.close()
    pool.join()

Answer 1

multiprocessing.Pool() 已经限制了同时运行的进程数。

您可以使用锁来分隔进程的开始时间（未测试）：

import threading
import multiprocessing

def init(lock):
    global starting
    starting = lock

def run_model(arg):
    starting.acquire() # no other process can get it until it is released
    threading.Timer(1, starting.release).start() # release in a second
    # ... start your simulation here

if __name__=="__main__":
   arguments = ...
   pool = Pool(processes=12, 
               initializer=init, initargs=[multiprocessing.Lock()])
   for _ in pool.imap_unordered(run_model, arguments):
       pass

Answer 2

使用线程和信号量的一种方法：

from time import sleep
import subprocess
import threading


def runmodels(arg):
    subprocess.call(arg)
    sGlobal.release() # release for next launch


if __name__ == '__main__':
    threads = []
    global sGlobal
    sGlobal = threading.Semaphore(12) #Semaphore for max 12 Thread
    arguments =  [big list of runs in here
                ]
    for arg in arguments :
        sGlobal.acquire() # Block if more than 12 thread
        t = threading.Thread(target=runmodels, args=(arg,))
        threads.append(t)
        t.start()
        sleep(1)

    for t in threads :
        t.join()

Answer 3

jfs 建议的答案给我带来了问题，因为我用threading.Timer 启动了一个新线程。如果工人恰好在计时器完成之前完成，则计时器将被终止并且永远不会释放锁。

我提出了一条替代路线，其中每个连续的工人将等待，直到自前一个工人开始以来已经过去了足够的时间。这似乎具有相同的预期效果，但不必依赖另一个子进程。

import multiprocessing as mp
import time

def init(shared_val):
    global start_time
    start_time = shared_val

def run_model(arg):
    with start_time.get_lock():
        wait_time = max(0, start_time.value - time.time())
        time.sleep(wait_time)
        start_time.value = time.time() + 1.0 # Specify interval here
    # ... start your simulation here

if __name__=="__main__":
   arguments = ...
   pool = mp.Pool(processes=12, 
                  initializer=init, initargs=[mp.Value('d')])
   for _ in pool.imap_unordered(run_model, arguments):
       pass