0

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'.
10
  • What does mytimeout do? Commented Feb 6, 2015 at 11:24
  • "mytimeout" is the function I need : it terminate the script in "arg" seconds if the script is not terminated. I edit my question. Commented Feb 6, 2015 at 11:28
  • I still don't understand your question. What does not work with your code? Commented Feb 6, 2015 at 11:30
  • My code is an example of what I want, I don't have "mytimeout" function working. Commented Feb 6, 2015 at 11:32
  • Would you be executing the script from python using something like subprocess? Commented Feb 6, 2015 at 11:34

3 Answers 3

2

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)
Sign up to request clarification or add additional context in comments.

1 Comment

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.
1

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.

1 Comment

Reported for promoting suicide.
0

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.

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.