如果需要超过 5 秒，则传递一个函数

Question

我在 for 循环中调用了一个函数，但是我想检查该函数的执行时间是否超过 5 秒，我想传递该迭代并继续进行下一次迭代。

我考虑过使用时间库，并启动一个时钟，但结束计时器只会在函数执行后执行，因此我无法在 5 秒内通过该特定迭代

Answer 1

我将使用 subprocess 模块发布替代解决方案。您需要使用您的函数创建一个 python 文件，将其作为子进程调用，然后调用 wait 方法。如果进程无法在期望的时间内完成，则会引发错误，因此您终止该进程并继续迭代。

例如，这是您要调用的函数：

from time import time
import sys

x = eval(sys.argv[1])   

t = time()
a = [i for i in range(int(x**5))]

#pipe to the main process the computaiton time
sys.stdout.write('%s'%(time()-t))

还有main函数，这里我在func.py文件中调用了前一个函数：

import subprocess as sp
from subprocess import Popen, PIPE


for i in range(1,50,1):
    #call the process
    process = Popen(['python','~func.py', '%i'%i],
                    stdout = PIPE,stdin = PIPE)

    try:
        #if it finish within 1 sec:
        process.wait(1)
        print('Finished in: %s s'%(process.stdout.read().decode()))

    except:
        #else kill the process. It is important to kill it,
        #otherwise it will keep running.
        print('Timeout')
        process.kill()
        

Answer 2

这是我一直在试验的一些代码，它有一个 task() 迭代它 params 参数并花费随机时间来完成每个参数。

我为每个task 启动一个thread，通过监视返回值队列等待线程完成。如果线程未能完成，则主循环放弃它，并开始下一个线程。

程序显示哪些任务失败或完成（每次都不同）。

完成的任务会打印出它们的结果（参数和睡眠时间）。

import threading, queue
import random
import time

def task(params, q):
    for p in params:
        s = random.randint(1,4)
        s = s * s
        s = s / 8
        time.sleep(s)
        q.put((p,s), False)
    q.put(None, False)  # None is sentinal value

def sampleQueue(q, ret, results):
    while not q.empty():
        item = q.get()
        if item:
            ret.append(item)
        else:
            # Found None sentinal
            results.append(ret)
            return True
    return False
    

old = []
results = []
for p in [1,2,3,4]:
    q = queue.SimpleQueue()
    t = threading.Thread(target=task, args=([p,p,p,p,p], q))
    t.start()
    end = time.time() + 5
    ret = []
    failed = True
    while time.time() < end:
        time.sleep(0.1)
        if sampleQueue(q, ret, results):
            failed = False
            break
    if failed:
        print(f'Task {p} failed!')
        old.append(t)
    else:
        print(f'Task {p} finished!')
        t.join()

print(results)
print(f'{len(old)} threads failed')
for t in old:
    t.join()
print('Done')

示例输出：

Task 1 finished!
Task 2 finished!
Task 3 failed!
Task 4 failed!
[[(1, 1.125), (1, 1.125), (1, 2.0), (1, 0.125), (1, 0.5)], [(2, 0.125), (2, 1.125), (2, 0.5), (2, 2.0), (2, 0.125)]]
2 threads failed
Done

Answer 3

我在下面附上一个例子。希望这对您有所帮助：

from threading import Timer 
class LoopStopper: 
 
    def __init__(self, seconds): 
        self._loop_stop = False 
        self._seconds = seconds 
  
    def _stop_loop(self): 
        self._loop_stop = True 
 
    def run( self, generator_expression, task): 
        """ Execute a task a number of times based on the generator_expression""" 
        t = Timer(self._seconds, self._stop_loop) 
        t.start() 
        for i in generator_expression: 
            task(i) 
            if self._loop_stop: 
                break 
        t.cancel() # Cancel the timer if the loop ends ok. 
 
ls = LoopStopper( 5) # 5 second timeout 
ls.run( range(1000000), print) # print numbers from 0 to 999999