Timeout function in Python

Question

I want to have a function, in Python (3.x), which force to the script itself to terminate, like :

i_time_value = 10
mytimeout(i_time_value )   # Terminate the script if not in i_time_value seconds 
for i in range(10):
   print("go")
   time.sleep(2)

Where "mytimeout" is the function I need : it terminate the script in "arg" seconds if the script is not terminated.

I have seen good solutions for put a timeout to a function here or here, but I don't want a timeout for a function but for the script.

Also :

I know that I can put my script in a function or using something like subprocess and use it with a timeout, I tried it and it works, but I want something more simple.
It must be Unix & Windows compatible.
The function must be universal i.e. : it may be add to any script in one line (except import)
I need a function not a 'how to put a timeout in a script'.

"mytimeout" is the function I need : it terminate the script in "arg" seconds if the script is not terminated. I edit my question. — philnext, Feb 06 '15 at 11:28
I still don't understand your question. What does not work with your code? — , Feb 06 '15 at 11:30
My code is an example of what I want, I don't have "mytimeout" function working. — philnext, Feb 06 '15 at 11:32
Would you be executing the script from python using something like subprocess? — Harman, Feb 06 '15 at 11:34
@Harman Sure I could, but it is more or less the same solution as putting it in a function : it works, but having a function to do it would be more simple (even if the function is not simple to write). I edit my post to be more clear. — philnext, Feb 06 '15 at 11:40
Which python version are you using ? A few are available 2.7 onwards — Harman, Feb 06 '15 at 11:47

Himal · Answer 1 · 2015-02-06T13:15:16.297

1

I would use something like this.

import sys
import time
import threading

def set_timeout(event):
    event.set()

event = threading.Event()
i_time_value = 2

t = threading.Timer(i_time_value, set_timeout, [event])
t.start()

for i in range(10):

    print("go")

    if event.is_set():
        print('Timed Out!')
        sys.exit()

    time.sleep(2)

edited Feb 06 '15 at 13:15

answered Feb 06 '15 at 12:08

Himal

1,351
3
13
28

Clearly not a good answer for me : you didn't make an universal function, you put an event in a loop in the main script. – philnext Feb 06 '15 at 13:35

score 1 · Accepted Answer · edited May 23 '17 at 12:06

signal is not Windows compatible.

You can send some signals on Windows e.g.:

os.kill(os.getpid(), signal.CTRL_C_EVENT) # send Ctrl+C to itself

You could use threading.Timer to call a function at a later time:

from threading import Timer

def kill_yourself(delay):
    t = Timer(delay, kill_yourself_now)
    t.daemon = True # no need to kill yourself if we're already dead
    t.start()

where kill_yourself_now():

import os
import signal
import sys

def kill_yourself_now():
    sig = signal.CTRL_C_EVENT if sys.platform == 'win32' else signal.SIGINT
    os.kill(os.getpid(), sig) # raise KeyboardInterrupt in the main thread

If your scripts starts other processes then see: how to kill child process(es) when parent dies? See also, How to terminate a python subprocess launched with shell=True -- it demonstrates how to kill a process tree.

Reported for promoting suicide. – NoName Mar 02 '20 at 21:01 — NoName, Mar 02 '20 at 21:01

score 0 · Answer 3 · answered Feb 06 '15 at 13:50

A little bit of googling turned this answer up:

import multiprocessing as MP
from sys import exc_info
from time import clock

DEFAULT_TIMEOUT = 60

################################################################################

def timeout(limit=None):
    if limit is None:
        limit = DEFAULT_TIMEOUT
    if limit <= 0:
        raise ValueError()
    def wrapper(function):
        return _Timeout(function, limit)
    return wrapper

class TimeoutError(Exception): pass

################################################################################

def _target(queue, function, *args, **kwargs):
    try:
        queue.put((True, function(*args, **kwargs)))
    except:
        queue.put((False, exc_info()[1]))

class _Timeout:

    def __init__(self, function, limit):
        self.__limit = limit
        self.__function = function
        self.__timeout = clock()
        self.__process = MP.Process()
        self.__queue = MP.Queue()

    def __call__(self, *args, **kwargs):
        self.cancel()
        self.__queue = MP.Queue(1)
        args = (self.__queue, self.__function) + args
        self.__process = MP.Process(target=_target, args=args, kwargs=kwargs)
        self.__process.daemon = True
        self.__process.start()
        self.__timeout = self.__limit + clock()

    def cancel(self):
        if self.__process.is_alive():
            self.__process.terminate()

    @property
    def ready(self):
        if self.__queue.full():
            return True
        elif not self.__queue.empty():
            return True
        elif self.__timeout < clock():
            self.cancel()
        else:
            return False

    @property
    def value(self):
        if self.ready is True:
            flag, load = self.__queue.get()
            if flag:
                return load
            raise load
        raise TimeoutError()

    def __get_limit(self):
        return self.__limit

    def __set_limit(self, value):
        if value <= 0:
            raise ValueError()
        self.__limit = value

    limit = property(__get_limit, __set_limit)

It might be Python 2.x, but it shouldn't be terribly hard to convert.

Timeout function in Python

3 Answers3