An improvement on @rik.the.vik's answer would be to use the with statement to give the timeout function some syntactic sugar:

import signal
from contextlib import contextmanager

class TimeoutException(Exception): pass

@contextmanager
def time_limit(seconds):
    def signal_handler(signum, frame):
        raise TimeoutException("Timed out!")
    signal.signal(signal.SIGALRM, signal_handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)


try:
    with time_limit(10):
        long_function_call()
except TimeoutException as e:
    print("Timed out!")

I'm not sure how cross-platform this might be, but using signals and alarm might be a good way of looking at this. With a little work you could make this completely generic as well and usable in any situation.

http://docs.python.org/library/signal.html

So your code is going to look something like this.

import signal

def signal_handler(signum, frame):
    raise Exception("Timed out!")

signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(10)   # Ten seconds
try:
    long_function_call()
except Exception, msg:
    print "Timed out!"

Here: a simple way of getting the desired effect:

https://pypi.org/project/func-timeout

This saved my life.

And now an example on how it works: lets say you have a huge list of items to be processed and you are iterating your function over those items. However, for some strange reason, your function get stuck on item n, without raising an exception. You need to other items to be processed, the more the better. In this case, you can set a timeout for processing each item:

import time
import func_timeout


def my_function(n):
    """Sleep for n seconds and return n squared."""
    print(f'Processing {n}')
    time.sleep(n)
    return n**2


def main_controller(max_wait_time, all_data):
    """
    Feed my_function with a list of itens to process (all_data).

    However, if max_wait_time is exceeded, return the item and a fail info.
    """
    res = []
    for data in all_data:
        try:
            my_square = func_timeout.func_timeout(
                max_wait_time, my_function, args=[data]
                )
            res.append((my_square, 'processed'))
        except func_timeout.FunctionTimedOut:
            print('error')
            res.append((data, 'fail'))
            continue

    return res


timeout_time = 2.1  # my time limit
all_data = range(1, 10)  # the data to be processed

res = main_controller(timeout_time, all_data)
print(res)

Doing this from within a signal handler is dangerous: you might be inside an exception handler at the time the exception is raised, and leave things in a broken state. For example,

def function_with_enforced_timeout():
  f = open_temporary_file()
  try:
   ...
  finally:
   here()
   unlink(f.filename)

If your exception is raised here(), the temporary file will never be deleted.

The solution here is for asynchronous exceptions to be postponed until the code is not inside exception-handling code (an except or finally block), but Python doesn't do that.

Note that this won't interrupt anything while executing native code; it'll only interrupt it when the function returns, so this may not help this particular case. (SIGALRM itself might interrupt the call that's blocking--but socket code typically simply retries after an EINTR.)

Doing this with threads is a better idea, since it's more portable than signals. Since you're starting a worker thread and blocking until it finishes, there are none of the usual concurrency worries. Unfortunately, there's no way to deliver an exception asynchronously to another thread in Python (other thread APIs can do this). It'll also have the same issue with sending an exception during an exception handler, and require the same fix.

You don't have to use threads. You can use another process to do the blocking work, for instance, maybe using the subprocess module. If you want to share data structures between different parts of your program then Twisted is a great library for giving yourself control of this, and I'd recommend it if you care about blocking and expect to have this trouble a lot. The bad news with Twisted is you have to rewrite your code to avoid any blocking, and there is a fair learning curve.

You can use threads to avoid blocking, but I'd regard this as a last resort, since it exposes you to a whole world of pain. Read a good book on concurrency before even thinking about using threads in production, e.g. Jean Bacon's "Concurrent Systems". I work with a bunch of people who do really cool high performance stuff with threads, and we don't introduce threads into projects unless we really need them.

The only "safe" way to do this, in any language, is to use a secondary process to do that timeout-thing, otherwise you need to build your code in such a way that it will time out safely by itself, for instance by checking the time elapsed in a loop or similar. If changing the method isn't an option, a thread will not suffice.

Why? Because you're risking leaving things in a bad state when you do. If the thread is simply killed mid-method, locks being held, etc. will just be held, and cannot be released.

So look at the process way, do not look at the thread way.

Using simple decorator

Here's the version I made after studying above answers. Pretty straight forward.

def function_timeout(seconds: int):
    """Wrapper of Decorator to pass arguments"""

    def decorator(func):
        @contextmanager
        def time_limit(seconds_):
            def signal_handler(signum, frame):  # noqa
                raise TimeoutException(f"Timed out in {seconds_} seconds!")

            signal.signal(signal.SIGALRM, signal_handler)
            signal.alarm(seconds_)
            try:
                yield
            finally:
                signal.alarm(0)

        @wraps(func)
        def wrapper(*args, **kwargs):
            with time_limit(seconds):
                return func(*args, **kwargs)

        return wrapper

    return decorator

How to use?

@function_timeout(seconds=5)
def my_naughty_function():
    while True:
        print("Try to stop me ;-p")

Well of course, don't forget to import the function if it is in a separate file.

Here's a timeout function I think I found via google and it works for me.

From: http://code.activestate.com/recipes/473878/

def timeout(func, args=(), kwargs={}, timeout_duration=1, default=None):
    '''This function will spwan a thread and run the given function using the args, kwargs and 
    return the given default value if the timeout_duration is exceeded 
    ''' 
    import threading
    class InterruptableThread(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)
            self.result = default
        def run(self):
            try:
                self.result = func(*args, **kwargs)
            except:
                self.result = default
    it = InterruptableThread()
    it.start()
    it.join(timeout_duration)
    if it.isAlive():
        return it.result
    else:
        return it.result   

The method from @user2283347 is tested working, but we want to get rid of the traceback messages. Use pass trick from Remove traceback in Python on Ctrl-C, the modified code is:

from contextlib import contextmanager
import threading
import _thread

class TimeoutException(Exception): pass

@contextmanager
def time_limit(seconds):
    timer = threading.Timer(seconds, lambda: _thread.interrupt_main())
    timer.start()
    try:
        yield
    except KeyboardInterrupt:
        pass     
    finally:
        # if the action ends in specified time, timer is canceled
        timer.cancel()

def timeout_svm_score(i):
     #from sklearn import svm
     #import numpy as np
     #from IPython.core.display import display
     #%store -r names X Y
     clf = svm.SVC(kernel='linear', C=1).fit(np.nan_to_num(X[[names[i]]]), Y)
     score = clf.score(np.nan_to_num(X[[names[i]]]),Y)
     #scoressvm.append((score, names[i]))
     display((score, names[i])) 
     
%%time
with time_limit(5):
    i=0
    timeout_svm_score(i)
#Wall time: 14.2 s

%%time
with time_limit(20):
    i=0
    timeout_svm_score(i)
#(0.04541284403669725, '计划飞行时间')
#Wall time: 16.1 s

%%time
with time_limit(5):
    i=14
    timeout_svm_score(i)
#Wall time: 5h 43min 41s

We can see that this method may need far long time to interrupt the calculation, we asked for 5 seconds, but it work out in 5 hours.

This code works for Windows Server Datacenter 2016 with python 3.7.3 and I didn't tested on Unix, after mixing some answers from Google and StackOverflow, it finally worked for me like this:

from multiprocessing import Process, Lock
import time
import os

def f(lock,id,sleepTime):
    lock.acquire()
    print("I'm P"+str(id)+" Process ID: "+str(os.getpid()))
    lock.release()
    time.sleep(sleepTime)   #sleeps for some time
    print("Process: "+str(id)+" took this much time:"+str(sleepTime))
    time.sleep(sleepTime)
    print("Process: "+str(id)+" took this much time:"+str(sleepTime*2))

if __name__ == '__main__':
    timeout_function=float(9) # 9 seconds for max function time
    print("Main Process ID: "+str(os.getpid()))
    lock=Lock()
    p1=Process(target=f, args=(lock,1,6,))   #Here you can change from 6 to 3 for instance, so you can watch the behavior
    start=time.time()
    print(type(start))
    p1.start()
    if p1.is_alive():
        print("process running a")
    else:
        print("process not running a")
    while p1.is_alive():
        timeout=time.time()
        if timeout-start > timeout_function:
            p1.terminate()
            print("process terminated")
        print("watching, time passed: "+str(timeout-start) )
        time.sleep(1)
    if p1.is_alive():
        print("process running b")
    else:
        print("process not running b")
    p1.join()
    if p1.is_alive():
        print("process running c")
    else:
        print("process not running c")
    end=time.time()
    print("I am the main process, the two processes are done")
    print("Time taken:- "+str(end-start)+" secs")   #MainProcess terminates at approx ~ 5 secs.
    time.sleep(5) # To see if on Task Manager the child process is really being terminated, and it is
    print("finishing")

The main code is from this link: Create two child process using python(windows)

Then I used .terminate() to kill the child process. You can see that the function f calls 2 prints, one after 5 seconds and another after 10 seconds. However, with a 7 seconds sleep and the terminate(), it does not show the last print.

It worked for me, hope it helps!