How to execute a function under limited time (eg under 5 minutes) in Python 3?

Question

I have this line of code ("model.predict(image=img)") that I want to stop executing if it is taking longer than 5 minutes. Is there an easy way or library in python I can use to achieve this?

for filename in os.listdir():
    if filename.endswith(".jpg") or filename.endswith(".png"):
        print(filename)
        img = DataURI.from_file(filename)
        #This should run for no longer than 5 minutes
        output = model.predict(image=img)
        #if this took more than 5 minutes to run, use "continue"
        #continue

You may find answers to this problem useful: https://stackoverflow.com/questions/492519/timeout-on-a-function-call — qzhong, Nov 03 '22 at 05:57

score 1 · Answer 1 · answered Nov 04 '22 at 09:42

The solution that I know is by using thread worker and kill it. one worker for running the function, and we have another funciton to maintain the running time.

I think you can use python-worker (link)

import time
from worker import worker

def run_with_timeout(worker_object, timeout):
    while worker_object.work_time < timeout:
        time.sleep(0.01) # give interval for checking
    worker_object.abort()

@worker
def my_controlled_function(a, b, c):
   ...

## then you can run it

run_with_timeout(my_controlled_function(1, 2, 3), timeout=10)

user19723070 · Answer 2 · 2022-11-03T06:28:35.467

You can use something like this:

for filename in os.listdir():
if filename.endswith(".jpg") or filename.endswith(".png"):
    print(filename)

    for x in range(0, 300)  #Converting Minutes to seconds so 300
        try:
            img = DataURI.from_file(filename)
            output = model.predict(image=img)
            break
                          
        except:
            time.sleep(1)
                    

    #if this took more than 5 minutes to run, use "continue"
    #continue

What this will do is like it will essentially try to check for it every 1 second until it is completely 5 minutes.

Or If this does't work you may refer this post which might help Link: Timeout on a function call

I don't think your code will have the desired effect... If "output" takes forever to run, the code will take forever running the "try" part of the code — plshelpmeout, Nov 03 '22 at 06:25

Veysel Olgun · Answer 3 · 2022-11-04T15:29:55.077

Your model.predict() method is CPU bound operation probably. In Linux, you can use signal mechanism that will interrupt your current operation, but Windows OS has no all POSIX signal features, so we can use thread or multiprocess to interrupt/terminate immediately some operation. Python gives a good option to terminate a thread using ctypes.pythonapi.PyThreadState_SetAsyncExc, but this will not terminate the thread before the I/O(if thread blocked) operation is finished

Here is a simple example :

import threading
import ctypes
import time

class TimeoutException(Exception):
    pass


def f(arg):
    while 1:
        time.sleep(1) 
        print("hello")       


class Timeout(threading.Thread):
    def __init__(self,func,arg,timeout):
        self.func = func
        self.arg = arg

        self.thread_ident = None
        self.output = None

        self.wait_for_finish = threading.Event()
        threading.Thread.__init__(self)

        self.killer = threading.Timer(timeout, self.raise_exception)
        
        self.start()
        self.wait_for_finish.wait()

    def raise_exception(self):
        ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(self.thread_ident),
                                                       ctypes.py_object(TimeoutException))
    def run(self):
        self.thread_ident = threading.get_ident()
        self.killer.start()
        try:
            self.output = self.func(self.arg)

        except TimeoutException:
            self.output = "Timeout exception" 

        finally:
            self.wait_for_finish.set()
            self.killer.cancel()

output = Timeout(func=f, arg=5, timeout = 5).output
print(output)

You can also look at this

How to execute a function under limited time (eg under 5 minutes) in Python 3?

3 Answers3