Écrire un décorateur TTL en Python

Question

J'essaye d'écrire un décorateur TTL en python. Fondamentalement, je lui donne lever une exception si la fonction ne répond pas dans l'heure sélectionnée.

Vous pouvez trouver les extraits de thead2 sur http://sebulba.wikispaces.com/recipe+thread2

from thread2 import Thread 

""" A TTL decorator. """ 
class Worker(Thread): 
    def __init__(self, q, f, args, kvargs): 
     Thread.__init__(self) 

     self.q = q 
     self.f = f 
     self.args = args 
     self.kvargs = kvargs 

    def run(self,): 
     try: 
      res = (True, self.f(*self.args, **self.kvargs)) 
      self.q.put(res) 
     except Exception, e: 
      self.q.put((False, e)) 

class Referee(Thread): 
    def __init__(self,q, ttl,exception_factory): 
     Thread.__init__(self) 

     self.exception_factory=exception_factory  
     self.q=q 
     self.ttl=ttl 

    def run(self): 
     time.sleep(self.ttl) 
     res = (False, self.exception_factory()) 
     self.q.put(res) 

def raise_if_too_long(ttl, exception_factory=lambda :RuntimeError("Timeout")): 
    def raise_if_too_long_aux(f): 
     def ritl(*args,**kvargs): 
      q = Queue.Queue(2) 

      referee = Referee(q, ttl, exception_factory) 
      worker = Worker(q,f,args,kvargs) 

      worker.start() 
      referee.start() 

      (valid, res)= q.get(1) 

      q.task_done() 

      referee.terminate() 
      worker.terminate() 

      if valid: 
       return res 
      else: 
       raise res 

     return ritl 

    return raise_if_too_long_aux 

Cependant, je reçois un résultat assez mauvais. Il semble que la fonction retourne parfois bien mais le décorateur ne retourne pas tant que le TTL n'est pas atteint et l'erreur augmente.

Un problème est survenu dans ce code? Existe-t-il un moyen/bibliothèque commun d'écrire une fonction avec un TTL en python?

Answer 1

Le code fourni est un peu difficile à suivre - comment va-t-il déclencher une exception au bon endroit au bon moment dans le bon thread?

Tenir compte de ce flux bruts:

fonction décorateur appelé avec la fonction cible. Retour une fonction:

2. Commence fil, appelant la fonction cible
3. Joint au fil en utilisant Thread.join ([timeout])
4. Si vous obtenez un délai d'attente, un lever exception, et ignorer le résultat du thread.
5. Si vous n'obtenez pas de délai, capturez le résultat du thread et renvoyez-le.

(Vous auriez besoin de trouver un moyen de capturer la sortie du fil ...)

Voir http://docs.python.org/library/threading.html pour plus d'informations sur le délai d'attente de thread ...

(Ou tout simplement commencer à utiliser erlang:)

Answer 2

Si vous voulez l'exécution de la fonction à mettre fin une fois le délai dépassé, vous voudrez peut-être essayer code ha s cette capacité. Pour utiliser le module, tout ce qui doit être fait est que votre fonction soit appelée en tant qu'argument à add_timeout, et que la valeur retournée puisse s'exécuter. Une fois appelée, la propriété ready de l'objet peut être interrogée et tout ce qui est renvoyé est accessible via la propriété value. La documentation du code devrait fournir une explication pour le reste de l'API disponible.

## {{{ http://code.activestate.com/recipes/577045/ (r2) 
#! /usr/bin/env python 
"""Provide way to add timeout specifications to arbitrary functions. 

There are many ways to add a timeout to a function, but no solution 
is both cross-platform and capable of terminating the procedure. This 
module use the multiprocessing module to solve both of those problems.""" 

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

__author__ = 'Stephen "Zero" Chappell <Noctis.Skytower@gmail.com>' 
__date__ = '11 February 2010' 
__version__ = '$Revision: 3 $' 

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

import inspect 
import sys 
import time 
import multiprocessing 

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

def add_timeout(function, limit=60): 
    """Add a timeout parameter to a function and return it. 

    It is illegal to pass anything other than a function as the first 
    parameter. If the limit is not given, it gets a default value equal 
    to one minute. The function is wrapped and returned to the caller.""" 
    assert inspect.isfunction(function) 
    if limit <= 0: 
     raise ValueError() 
    return _Timeout(function, limit) 

class NotReadyError(Exception): pass 

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

def _target(queue, function, *args, **kwargs): 
    """Run a function with arguments and return output via a queue. 

    This is a helper function for the Process created in _Timeout. It runs 
    the function with positional arguments and keyword arguments and then 
    returns the function's output by way of a queue. If an exception gets 
    raised, it is returned to _Timeout to be raised by the value property.""" 
    try: 
     queue.put((True, function(*args, **kwargs))) 
    except: 
     queue.put((False, sys.exc_info()[1])) 

class _Timeout: 

    """Wrap a function and add a timeout (limit) attribute to it. 

    Instances of this class are automatically generated by the add_timeout 
    function defined above. Wrapping a function allows asynchronous calls 
    to be made and termination of execution after a timeout has passed.""" 

    def __init__(self, function, limit): 
     """Initialize instance in preparation for being called.""" 
     self.__limit = limit 
     self.__function = function 
     self.__timeout = time.clock() 
     self.__process = multiprocessing.Process() 
     self.__queue = multiprocessing.Queue() 

    def __call__(self, *args, **kwargs): 
     """Execute the embedded function object asynchronously. 

     The function given to the constructor is transparently called and 
     requires that "ready" be intermittently polled. If and when it is 
     True, the "value" property may then be checked for returned data.""" 
     self.cancel() 
     self.__queue = multiprocessing.Queue(1) 
     args = (self.__queue, self.__function) + args 
     self.__process = multiprocessing.Process(target=_target, 
               args=args, 
               kwargs=kwargs) 
     self.__process.daemon = True 
     self.__process.start() 
     self.__timeout = self.__limit + time.clock() 

    def cancel(self): 
     """Terminate any possible execution of the embedded function.""" 
     if self.__process.is_alive(): 
      self.__process.terminate() 

    @property 
    def ready(self): 
     """Read-only property indicating status of "value" property.""" 
     if self.__queue.full(): 
      return True 
     elif not self.__queue.empty(): 
      return True 
     elif self.__timeout < time.clock(): 
      self.cancel() 
     else: 
      return False 

    @property 
    def value(self): 
     """Read-only property containing data returned from function.""" 
     if self.ready is True: 
      flag, load = self.__queue.get() 
      if flag: 
       return load 
      raise load 
     raise NotReadyError() 

    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, 
        doc="Property for controlling the value of the timeout.") 
## end of http://code.activestate.com/recipes/577045/ }}}