Appel de méthode asynchrone en Python?

Je me demandais s'il existe une bibliothèque pour les appels de méthodes asynchrones en Python . Ce serait génial si vous pouviez faire quelque chose comme

@async
def longComputation():
    <code>


token = longComputation()
token.registerCallback(callback_function)
# alternative, polling
while not token.finished():
    doSomethingElse()
    if token.finished():
        result = token.result()

Ou pour appeler une routine non asynchrone de manière asynchrone

def longComputation()
    <code>

token = asynccall(longComputation())

Ce serait formidable d'avoir une stratégie plus raffinée en tant que native dans le noyau de la langue. Cela a-t-il été envisagé?

Vous pouvez utiliser le module multiprocesseur ajouté dans Python 2.6. Vous pouvez utiliser des pools de processus, puis obtenir des résultats de manière asynchrone avec:

apply_async(func[, args[, kwds[, callback]]])

Par exemple:

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    pool = Pool(processes=1)              # Start a worker processes.
    result = pool.apply_async(f, [10], callback) # Evaluate "f(10)" asynchronously calling callback when finished.

Ce n'est qu'une alternative. Ce module fournit de nombreuses fonctionnalités pour réaliser ce que vous voulez. De plus, il sera très facile d'en faire un décorateur.

Quelque chose comme:

import threading

thr = threading.Thread(target=foo, args=(), kwargs={})
thr.start() # Will run "foo"
....
thr.is_alive() # Will return whether foo is running currently
....
thr.join() # Will wait till "foo" is done

Consultez la documentation sur https://docs.python.org/library/threading.html pour plus de détails.

À partir de Python 3.5, vous pouvez utiliser des générateurs améliorés pour les fonctions asynchrones.

import asyncio
import datetime

Syntaxe du générateur améliorée:

@asyncio.coroutine
def display_date(loop):
    end_time = loop.time() + 5.0
    while True:
        print(datetime.datetime.now())
        if (loop.time() + 1.0) >= end_time:
            break
        yield from asyncio.sleep(1)


loop = asyncio.get_event_loop()
# Blocking call which returns when the display_date() coroutine is done
loop.run_until_complete(display_date(loop))
loop.close()

Nouvelle async/awaitsyntaxe:

async def display_date(loop):
    end_time = loop.time() + 5.0
    while True:
        print(datetime.datetime.now())
        if (loop.time() + 1.0) >= end_time:
            break
        await asyncio.sleep(1)


loop = asyncio.get_event_loop()
# Blocking call which returns when the display_date() coroutine is done
loop.run_until_complete(display_date(loop))
loop.close()

Ce n'est pas dans le noyau du langage, mais une bibliothèque très mature qui fait ce que vous voulez est Twisted . Il introduit l'objet Deferred, auquel vous pouvez attacher des rappels ou des gestionnaires d'erreurs ("errbacks"). Un différé est essentiellement une «promesse» qu'une fonction aura un résultat éventuellement.

Vous pouvez implémenter un décorateur pour rendre vos fonctions asynchrones, bien que ce soit un peu délicat. Le multiprocessingmodule est plein de petites bizarreries et de restrictions apparemment arbitraires - raison de plus pour l'encapsuler derrière une interface conviviale, cependant.

from inspect import getmodule
from multiprocessing import Pool


def async(decorated):
    r'''Wraps a top-level function around an asynchronous dispatcher.

        when the decorated function is called, a task is submitted to a
        process pool, and a future object is returned, providing access to an
        eventual return value.

        The future object has a blocking get() method to access the task
        result: it will return immediately if the job is already done, or block
        until it completes.

        This decorator won't work on methods, due to limitations in Python's
        pickling machinery (in principle methods could be made pickleable, but
        good luck on that).
    '''
    # Keeps the original function visible from the module global namespace,
    # under a name consistent to its __name__ attribute. This is necessary for
    # the multiprocessing pickling machinery to work properly.
    module = getmodule(decorated)
    decorated.__name__ += '_original'
    setattr(module, decorated.__name__, decorated)

    def send(*args, **opts):
        return async.pool.apply_async(decorated, args, opts)

    return send

Le code ci-dessous illustre l'utilisation du décorateur:

@async
def printsum(uid, values):
    summed = 0
    for value in values:
        summed += value

    print("Worker %i: sum value is %i" % (uid, summed))

    return (uid, summed)


if __name__ == '__main__':
    from random import sample

    # The process pool must be created inside __main__.
    async.pool = Pool(4)

    p = range(0, 1000)
    results = []
    for i in range(4):
        result = printsum(i, sample(p, 100))
        results.append(result)

    for result in results:
        print("Worker %i: sum value is %i" % result.get())

Dans un cas réel, j'aborderais un peu plus le décorateur, en fournissant un moyen de le désactiver pour le débogage (tout en gardant la future interface en place), ou peut-être une facilité pour gérer les exceptions; mais je pense que cela démontre assez bien le principe.

Juste

import threading, time

def f():
    print "f started"
    time.sleep(3)
    print "f finished"

threading.Thread(target=f).start()

Vous pouvez utiliser eventlet. Il vous permet d'écrire ce qui semble être du code synchrone, mais de le faire fonctionner de manière asynchrone sur le réseau.

Voici un exemple de robot d'exploration super minimal:

urls = ["http://www.google.com/intl/en_ALL/images/logo.gif",
     "https://wiki.secondlife.com/w/images/secondlife.jpg",
     "http://us.i1.yimg.com/us.yimg.com/i/ww/beta/y3.gif"]

import eventlet
from eventlet.green import urllib2

def fetch(url):

  return urllib2.urlopen(url).read()

pool = eventlet.GreenPool()

for body in pool.imap(fetch, urls):
  print "got body", len(body)

Ma solution est:

import threading

class TimeoutError(RuntimeError):
    pass

class AsyncCall(object):
    def __init__(self, fnc, callback = None):
        self.Callable = fnc
        self.Callback = callback

    def __call__(self, *args, **kwargs):
        self.Thread = threading.Thread(target = self.run, name = self.Callable.__name__, args = args, kwargs = kwargs)
        self.Thread.start()
        return self

    def wait(self, timeout = None):
        self.Thread.join(timeout)
        if self.Thread.isAlive():
            raise TimeoutError()
        else:
            return self.Result

    def run(self, *args, **kwargs):
        self.Result = self.Callable(*args, **kwargs)
        if self.Callback:
            self.Callback(self.Result)

class AsyncMethod(object):
    def __init__(self, fnc, callback=None):
        self.Callable = fnc
        self.Callback = callback

    def __call__(self, *args, **kwargs):
        return AsyncCall(self.Callable, self.Callback)(*args, **kwargs)

def Async(fnc = None, callback = None):
    if fnc == None:
        def AddAsyncCallback(fnc):
            return AsyncMethod(fnc, callback)
        return AddAsyncCallback
    else:
        return AsyncMethod(fnc, callback)

Et fonctionne exactement comme demandé:

@Async
def fnc():
    pass

Quelque chose comme ça fonctionne pour moi, vous pouvez alors appeler la fonction, et elle se répartira sur un nouveau thread.

from thread import start_new_thread

def dowork(asynchronous=True):
    if asynchronous:
        args = (False)
        start_new_thread(dowork,args) #Call itself on a new thread.
    else:
        while True:
            #do something...
            time.sleep(60) #sleep for a minute
    return

Y a-t-il une raison de ne pas utiliser de threads? Vous pouvez utiliser la threadingclasse. Au lieu de la finished()fonction, utilisez le isAlive(). La result()fonction pourrait join()le thread et récupérer le résultat. Et, si vous le pouvez, remplacez les fonctions run()et __init__pour appeler la fonction spécifiée dans le constructeur et enregistrez la valeur quelque part dans l'instance de la classe.

Vous pouvez utiliser concurrent.futures (ajouté dans Python 3.2).

import time
from concurrent.futures import ThreadPoolExecutor


def long_computation(duration):
    for x in range(0, duration):
        print(x)
        time.sleep(1)
    return duration * 2


print('Use polling')
with ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(long_computation, 5)
    while not future.done():
        print('waiting...')
        time.sleep(0.5)

    print(future.result())

print('Use callback')
executor = ThreadPoolExecutor(max_workers=1)
future = executor.submit(long_computation, 5)
future.add_done_callback(lambda f: print(f.result()))

print('waiting for callback')

executor.shutdown(False)  # non-blocking

print('shutdown invoked')

Vous pouvez utiliser process. Si vous voulez l'exécuter pour toujours, utilisez-le (comme le réseau) dans votre fonction:

from multiprocessing import Process
def foo():
    while 1:
        # Do something

p = Process(target = foo)
p.start()

si vous voulez juste l'exécuter une fois, faites comme ça:

from multiprocessing import Process
def foo():
    # Do something

p = Process(target = foo)
p.start()
p.join()