Python-сообщество

from datetime import datetime
import sys
import threading
import time
def timed_run(func, args=(), kwargs={}, timeout=None):
    class TimedThread(threading.Thread):
        """An abortable thread, by merly raising an exception inside its
        context.
        """
        def __init__(self):
            super(TimedThread, self).__init__()
            self.exc_info = (None, None, None)
        def run(self):
            self.started_at = datetime.now()
            try:
                self.result = func(*args, **kwargs)
            except:
                # save the exception as an object attribute
                self.exc_info = sys.exc_info()
                self.result = None
            self.ended_at = datetime.now()
        def abort(self):
            self.ended_at = datetime.now()
            raise RuntimeError("Thread aborted as requested.")
    t = TimedThread()
    t.start()
    t.join(timeout)
    if t.exc_info[0] is not None:  # if there were any exceptions
        t, v, tb = t.exc_info
        raise t, v, tb  # Raise the exception/traceback inside the caller
    if t.is_alive():
        t.abort()
        diff = t.ended_at - t.started_at
        raise RuntimeError("%(f)s timed out after %(d)r seconds" %
                {'f': func, 'd': diff.seconds})
    return t.result
def func():
    # Unstoppable func
    while True:
        print "I am an unstoppable"
        time.sleep(5)
    return "done"
if __name__ == '__main__':
    print timed_run(func=func,timeout=10)

from threading import Timer
import thread
import time
 
 
class TimeoutError(Exception):
    """Function timed out"""
 
 
def timed_run(func, args=(), kwargs={}, timeout=None):
    start = time.time()
    try:
        Timer(timeout, thread.interrupt_main).start()
        return func(*args, **kwargs)
    except KeyboardInterrupt:
        raise TimeoutError("%s timed out (%.2f seconds)" % 
                            (func, time.time() - start))
 
 
def func():
    # Unstoppable func
    while True:
        print "I am an unstoppable"
        time.sleep(5)
     return "done"
if __name__ == '__main__':
    print timed_run(func=func,timeout=1.0)

from multiprocessing.process import Process
import time
    
def func():
    while True:
        print "I am an unstoppable"
        time.sleep(1)
    
result = None
timeout = 15 
p = Process(target=func)
p.start()
    
# waiting for timeout
p.join(timeout)
    
print "timeout has elapsed or process has finished"
    
if p.is_alive():
    p.terminate()
    print "terminating the process"
    
time.sleep(10)
print "p.is_alive = %r" % p.is_alive()

import multiprocessing
import time
    
resultq = multiprocessing.Queue(1)
    
def func1():
    while True:
        print "I am an unstoppable"
        time.sleep(1)
    #time.sleep(5)
    return True
    
def func2():
    time.sleep(5)
    return "Result of func2"
    
def _func(func=None, func_args=()):
    result = func(*func_args)
    resultq.put(result)       
    
def timedoutRun(func=None, f_args=()):
    timeout = 10
    result = None
        
    p = multiprocessing.Process(target=_func, args=(func,)+f_args)
    p.start()
        
    # waiting for timeout
    p.join(timeout)
    print "timeout has elapsed or process has finished"
    
    if resultq.empty():
        print "We do not have a result of func because of the func gets stuck"
    else:
        print "process has competed"        
        result = resultq.get()
        print "result is %r" % result
        
    if p.is_alive():
        p.terminate()
        print "terminating the process"
         
    time.sleep(10)
    print "p.is_alive = %r" % p.is_alive()
    
    return result
    
    
if __name__ == '__main__':
    print "start"
    print timedoutRun(func1)
    print timedoutRun(func2)
    print "end"