La meilleure façon d'effectuer le traitement multiple dans les demandes avec le python Tornade serveur?

Je suis en utilisant des e/S non bloquantes python serveur Tornade. J'ai une classe de GET les demandes qui peuvent prendre beaucoup de temps pour le terminer (pense que dans la gamme de 5 à 10 secondes). Le problème est que la Tornade blocs sur ces demandes suivantes rapide des demandes sont maintenus en place jusqu'à ce que la lenteur de la demande complète.

J'ai regardé: https://github.com/facebook/tornado/wiki/Threading-and-concurrency et est venu à la conclusion que je voulais une combinaison de #3 (d'autres processus) et #4 (les autres threads). #4 sur son propre eu des problèmes et j'ai été incapable d'obtenir un contrôle fiable de retour à la ioloop quand il y avait un autre thread à faire le "heavy_lifting". (Je suppose que c'était dû à la GIL et le fait que le heavy_lifting tâche a charge élevée de l'UC et continue tirant de contrôle à distance de la principale ioloop, mais c'est une supposition).

J'ai donc été prototypage comment résoudre ce problème en faisant des "gros travaux" tâches au sein de ces lentes GET demandes dans un processus séparé et un rappel de retour dans la Tornade ioloop lorsque le processus est terminé pour terminer la demande. Cela libère de la ioloop pour traiter d'autres demandes.

J'ai créé un exemple simple illustrant une solution possible, mais je suis curieux d'avoir des retours de la communauté sur elle.

Ma question est double: Comment cela peut-il l'approche actuelle d'être simplifié? Quels sont les pièges potentiellement exister avec elle?

L'Approche

Utiliser Tornade builtin asynchronous décorateur qui permet à une demande de rester ouvert et de la ioloop pour continuer.
Frayer un processus distinct pour les "gros travaux" tâches à l'aide de python multiprocessing module. J'ai d'abord tenté d'utiliser le threading module, mais n'a pas pu obtenir toute la fiabilité de renoncer à de contrôle le ioloop. Il apparaît également que mutliprocessing permettront de profiter des multicores.
Commencer un "observateur" fil dans les principales ioloop processus à l'aide de la threading module dont le travail est de regarder un multiprocessing.Queue pour les résultats du "gros œuvre" de la tâche quand il se termine. Cela était nécessaire parce que j'avais besoin de savoir que la heavy_lifting tâche avait terminé tout en étant en mesure de toujours aviser le ioloop que cette demande est maintenant terminé.
Assurez-vous que le "guetteur" fil abandonne le contrôle de la principale ioloop boucle souvent avec time.sleep(0) appels afin que d'autres demandes continuent d'obtenir facilement traitées.
Quand il y a un résultat dans la file d'attente puis ajouter un rappel de la "watcher" thread à l'aide de tornado.ioloop.IOLoop.instance().add_callback() qui est documenté pour être le seul moyen sûr d'appel ioloop cas des autres threads.
Assurez-vous d'appeler ensuite finish() dans le rappel pour remplir la demande et à la main sur une réponse.

Ci-dessous est un exemple de code montrant cette approche. multi_tornado.py est le serveur de mise en œuvre de ce qui précède, et call_multi.py est un exemple de script qui appelle le serveur de deux façons différentes pour tester le serveur. Les deux tests d'appeler le serveur avec 3 lent GET demandes de suivi de 20 rapide GET demandes. Les résultats sont présentés pour les deux cours d'exécution avec et sans le threading.

Dans le cas de l'exécution avec "pas de filetage" les 3 demandes lentes (bloc de chaque de prendre un peu plus d'une seconde pour terminer). Quelques-uns des 20 rapide des demandes faufiler entre certains de la lenteur de la demande dans le ioloop (pas totalement sûr de comment cela se produit - mais il pourrait être un artefact que je suis en cours d'exécution à la fois le serveur et le client script de test sur la même machine). Le point ici est que tous les rapide, les demandes sont à des degrés divers.

Dans le cas de l'exécution avec threading activé le 20 rapide des requêtes de la première immédiatement et les trois demandes lentes complète à peu près au même moment par la suite qu'ils ont été exécutés en parallèle. C'est le comportement souhaité. Les trois lent demandes de 2,5 secondes pour terminer en parallèle, alors que dans le non filetée cas, les trois demandes lentes prendre environ 3,5 secondes au total. Donc, il y a environ 35% de la vitesse maximum de l'ensemble (je suppose en raison de multicœur de partage). Mais le plus important - de la rapide des demandes ont été immédiatement traitées en leu de les lentes.

Je n'ai pas beaucoup d'expérience avec la programmation multithread - alors que cette apparence travaille ici, je suis curieux de savoir:

Est-il un moyen plus simple pour accomplir cette? Ce monstre peuvent se cacher à l'intérieur de cette approche?

(Note: Un avenir compromis peut être de simplement exécuter plusieurs instances de Tornade avec un reverse proxy comme nginx faire de l'équilibrage de la charge. Peu importe ce que je vais être l'exécution de plusieurs instances avec un équilibreur de charge - mais je suis inquiet au sujet de simplement jeter le matériel à ce problème, puisqu'il semble que le matériel est directement couplé au problème en termes de blocage.)

Exemple De Code

multi_tornado.py (serveur exemple):

import time
import threading
import multiprocessing
import math
from tornado.web import RequestHandler, Application, asynchronous
from tornado.ioloop import IOLoop
# run in some other process - put result in q
def heavy_lifting(q):
t0 = time.time()
for k in range(2000):
math.factorial(k)
t = time.time()
q.put(t - t0)  # report time to compute in queue
class FastHandler(RequestHandler):
def get(self):
res = 'fast result ' + self.get_argument('id')
print res
self.write(res)
self.flush()
class MultiThreadedHandler(RequestHandler):
# Note:  This handler can be called with threaded = True or False
def initialize(self, threaded=True):
self._threaded = threaded
self._q = multiprocessing.Queue()
def start_process(self, worker, callback):
# method to start process and watcher thread
self._callback = callback
if self._threaded:
# launch process
multiprocessing.Process(target=worker, args=(self._q,)).start()
# start watching for process to finish
threading.Thread(target=self._watcher).start()
else:
# threaded = False just call directly and block
worker(self._q)
self._watcher()
def _watcher(self):
# watches the queue for process result
while self._q.empty():
time.sleep(0)  # relinquish control if not ready
# put callback back into the ioloop so we can finish request
response = self._q.get(False)
IOLoop.instance().add_callback(lambda: self._callback(response))
class SlowHandler(MultiThreadedHandler):
@asynchronous
def get(self):
# start a thread to watch for
self.start_process(heavy_lifting, self._on_response)
def _on_response(self, delta):
_id = self.get_argument('id')
res = 'slow result {} <--- {:0.3f} s'.format(_id, delta)
print res
self.write(res)
self.flush()
self.finish()   # be sure to finish request
application = Application([
(r"/fast", FastHandler),
(r"/slow", SlowHandler, dict(threaded=False)),
(r"/slow_threaded", SlowHandler, dict(threaded=True)),
])
if __name__ == "__main__":
application.listen(8888)
IOLoop.instance().start()

call_multi.py (client testeur):

import sys
from tornado.ioloop import IOLoop
from tornado import httpclient
def run(slow):
def show_response(res):
print res.body
# make 3 "slow" requests on server
requests = []
for k in xrange(3):
uri = 'http://localhost:8888/{}?id={}'
requests.append(uri.format(slow, str(k + 1)))
# followed by 20 "fast" requests
for k in xrange(20):
uri = 'http://localhost:8888/fast?id={}'
requests.append(uri.format(k + 1))
# show results as they return
http_client = httpclient.AsyncHTTPClient()
print 'Scheduling Get Requests:'
print '------------------------'
for req in requests:
print req
http_client.fetch(req, show_response)
# execute requests on server
print '\nStart sending requests....'
IOLoop.instance().start()
if __name__ == '__main__':
scenario = sys.argv[1]
if scenario == 'slow' or scenario == 'slow_threaded':
run(scenario)

Résultats De Test

En exécutant python call_multi.py slow (le comportement de blocage):

Scheduling Get Requests:
------------------------
http://localhost:8888/slow?id=1
http://localhost:8888/slow?id=2
http://localhost:8888/slow?id=3
http://localhost:8888/fast?id=1
http://localhost:8888/fast?id=2
http://localhost:8888/fast?id=3
http://localhost:8888/fast?id=4
http://localhost:8888/fast?id=5
http://localhost:8888/fast?id=6
http://localhost:8888/fast?id=7
http://localhost:8888/fast?id=8
http://localhost:8888/fast?id=9
http://localhost:8888/fast?id=10
http://localhost:8888/fast?id=11
http://localhost:8888/fast?id=12
http://localhost:8888/fast?id=13
http://localhost:8888/fast?id=14
http://localhost:8888/fast?id=15
http://localhost:8888/fast?id=16
http://localhost:8888/fast?id=17
http://localhost:8888/fast?id=18
http://localhost:8888/fast?id=19
http://localhost:8888/fast?id=20
Start sending requests....
slow result 1 <--- 1.338 s
fast result 1
fast result 2
fast result 3
fast result 4
fast result 5
fast result 6
fast result 7
slow result 2 <--- 1.169 s
slow result 3 <--- 1.130 s
fast result 8
fast result 9
fast result 10
fast result 11
fast result 13
fast result 12
fast result 14
fast result 15
fast result 16
fast result 18
fast result 17
fast result 19
fast result 20

En exécutant python call_multi.py slow_threaded (le comportement désiré):

Scheduling Get Requests:
------------------------
http://localhost:8888/slow_threaded?id=1
http://localhost:8888/slow_threaded?id=2
http://localhost:8888/slow_threaded?id=3
http://localhost:8888/fast?id=1
http://localhost:8888/fast?id=2
http://localhost:8888/fast?id=3
http://localhost:8888/fast?id=4
http://localhost:8888/fast?id=5
http://localhost:8888/fast?id=6
http://localhost:8888/fast?id=7
http://localhost:8888/fast?id=8
http://localhost:8888/fast?id=9
http://localhost:8888/fast?id=10
http://localhost:8888/fast?id=11
http://localhost:8888/fast?id=12
http://localhost:8888/fast?id=13
http://localhost:8888/fast?id=14
http://localhost:8888/fast?id=15
http://localhost:8888/fast?id=16
http://localhost:8888/fast?id=17
http://localhost:8888/fast?id=18
http://localhost:8888/fast?id=19
http://localhost:8888/fast?id=20
Start sending requests....
fast result 1
fast result 2
fast result 3
fast result 4
fast result 5
fast result 6
fast result 7
fast result 8
fast result 9
fast result 10
fast result 11
fast result 12
fast result 13
fast result 14
fast result 15
fast result 19
fast result 20
fast result 17
fast result 16
fast result 18
slow result 2 <--- 2.485 s
slow result 3 <--- 2.491 s
slow result 1 <--- 2.517 s

Recommandation: attention pour le mur de texte.
OK. Des Suggestions? N'est-ce pas clair pour moi de communiquer tous les détails de ce qui se passe beaucoup plus concise.
Elle est généralement préférable de poser des questions comme celle-ci dans plusieurs petites. Mais, j'ai peut-être tort. Alors... est-ce votre seule question, comment simplifier cela? J'en ai mis que dans le haut - plus intéressant.
Je suis à la recherche de la simplification ou de l'autre approche. J'ai édité la question légèrement afin de mettre un peu plus de ce que je suis à la recherche d'avant.
Il me semble que vous pouvez simplifier ce à l'aide d'une file d'attente de demandes qui alimente un processus de piscine tels que celui trouvé dans le module multiprocessing. Voir docs.python.org/2/library/... pour plus d'info.
curieux, comment avez-vous résoudre ce problème? J'ai le même problème 🙂 Merci!
J'ai été très occupé avec d'autres aspects du projet - de sorte que je n'ai pas fait plus encore - mais le plan sur l'exécution ci-dessus MultithreadedHandler. J'ai vraiment pas vu d'autres commentaires sur d'autres approches qui utilisent la Tornade.

InformationsquelleAutor Rocketman | 2013-03-13

31

Si vous êtes prêt à utiliser simultanées.les contrats à terme.ProcessPoolExecutor au lieu de multiprocessing, c'est en fait très simple. Tornade ioloop prend déjà en charge concurrent.futures.Future, afin qu'ils jouent bien ensemble la sortie de la boîte. concurrent.futures est inclus dans Python 3.2+, et a été intégré à Python 2.x.

Voici un exemple:
```
import time
from concurrent.futures import ProcessPoolExecutor
from tornado.ioloop import IOLoop
from tornado import gen
def f(a, b, c, blah=None):
print "got %s %s %s and %s" % (a, b, c, blah)
time.sleep(5)
return "hey there"
@gen.coroutine
def test_it():
pool = ProcessPoolExecutor(max_workers=1)
fut = pool.submit(f, 1, 2, 3, blah="ok")  # This returns a concurrent.futures.Future
print("running it asynchronously")
ret = yield fut
print("it returned %s" % ret)
pool.shutdown()
IOLoop.instance().run_sync(test_it)
```
De sortie:
```
running it asynchronously
got 1 2 3 and ok
it returned hey there
```
ProcessPoolExecutor a un nombre plus limité d'API que multiprocessing.Pool, mais si vous n'avez pas besoin de fonctionnalités plus avancées de multiprocessing.Pool, c'est utile parce que l'intégration est beaucoup plus simple.
- Serait-ce de travailler pour asynchrones accès à Mongodb ainsi par exemple? Je ne semble pas être en mesure d'obtenir à Moteur de travail avec ssl donc cela pourrait être une solution alternative (j'ai des requêtes qui peut prendre plusieurs secondes). Liés à la question: dois-je aller avec ThreadPoolExecutor à la place? Peut-être plus léger?
- Oui, vous pouvez utiliser un ThreadPoolExecutor et suivent le même modèle. Vous n'avez pas besoin ProcessPoolExecutor parce que l'accès à un DB de blocage I/O, ce qui devrait libérer le GIL.
- Façon la plus simple pour envoyer le code python dans la Tornade contexte à un autre processus. Merci!
InformationsquelleAutor dano

multiprocessing.Pool peut être intégré dans le tornado I/O loop, mais c'est un peu brouillon. Un beaucoup plus propre d'intégration peut être fait en utilisant concurrent.futures (voir mon autre réponse pour plus de détails), mais si vous êtes coincé sur Python 2.x et ne peut pas installer les concurrent.futures backport, voici comment vous pouvez le faire strictement à l'aide de multiprocessing:

La multiprocessing.Pool.apply_async et multiprocessing.Pool.map_async méthodes, les deux ont une option de callback paramètre, ce qui signifie que les deux peuvent potentiellement être branché dans une tornado.gen.Task. Donc, dans la plupart des cas, l'exécution de code asynchrone dans un sous-processus est aussi simple que cela:

import multiprocessing
import contextlib
from tornado import gen
from tornado.gen import Return
from tornado.ioloop import IOLoop
from functools import partial
def worker():
print "async work here"
@gen.coroutine
def async_run(func, *args, **kwargs):
result = yield gen.Task(pool.apply_async, func, args, kwargs)
raise Return(result)
if __name__ == "__main__":
pool = multiprocessing.Pool(multiprocessing.cpu_count())
func = partial(async_run, worker)
IOLoop().run_sync(func)

Comme je l'ai mentionné, cela fonctionne bien dans le plus cas. Mais si worker() déclenche une exception, callback n'est jamais appelée, ce qui signifie que le gen.Task ne se termine jamais, et vous accrocher à jamais. Maintenant, si vous savez que votre travail ne sera jamais lever une exception (parce que vous avez enveloppé le tout dans un try/except, par exemple), vous vous ferez un plaisir d'utiliser cette approche. Toutefois, si vous voulez laisser les exceptions s'échapper du travailleur, la seule solution que j'ai trouvé était à la sous-classe certains multitraitement composants, et de faire appel callback même si le travailleur sous-processus a soulevé une exception:

from multiprocessing.pool import ApplyResult, Pool, RUN
import multiprocessing
class TornadoApplyResult(ApplyResult):
def _set(self, i, obj):
self._success, self._value = obj 
if self._callback:
self._callback(self._value)
self._cond.acquire()
try:
self._ready = True
self._cond.notify()
finally:
self._cond.release()
del self._cache[self._job]
class TornadoPool(Pool):
def apply_async(self, func, args=(), kwds={}, callback=None):
''' Asynchronous equivalent of `apply()` builtin
This version will call `callback` even if an exception is
raised by `func`.
'''
assert self._state == RUN
result = TornadoApplyResult(self._cache, callback)
self._taskqueue.put(([(result._job, None, func, args, kwds)], None))
return result
...
if __name__ == "__main__":
pool = TornadoPool(multiprocessing.cpu_count())
...

Avec ces changements, l'objet de l'exception sera retournée par la gen.Task, plutôt que de la gen.Task suspendu indéfiniment. J'ai aussi mis à jour mon async_run méthode de ré-augmenter l'exception lors de son retour, avaient fait d'autres modifications à fournir de meilleurs retraçage pour les exceptions levées dans le travailleur sous-processus. Voici le code complet:

import multiprocessing
from multiprocessing.pool import Pool, ApplyResult, RUN
from functools import wraps
import tornado.web
from tornado.ioloop import IOLoop
from tornado.gen import Return
from tornado import gen
class WrapException(Exception):
def __init__(self):
exc_type, exc_value, exc_tb = sys.exc_info()
self.exception = exc_value
self.formatted = ''.join(traceback.format_exception(exc_type, exc_value, exc_tb))
def __str__(self):
return '\n%s\nOriginal traceback:\n%s' % (Exception.__str__(self), self.formatted)
class TornadoApplyResult(ApplyResult):
def _set(self, i, obj):
self._success, self._value = obj 
if self._callback:
self._callback(self._value)
self._cond.acquire()
try:
self._ready = True
self._cond.notify()
finally:
self._cond.release()
del self._cache[self._job]   
class TornadoPool(Pool):
def apply_async(self, func, args=(), kwds={}, callback=None):
''' Asynchronous equivalent of `apply()` builtin
This version will call `callback` even if an exception is
raised by `func`.
'''
assert self._state == RUN
result = TornadoApplyResult(self._cache, callback)
self._taskqueue.put(([(result._job, None, func, args, kwds)], None))
return result
@gen.coroutine
def async_run(func, *args, **kwargs):
""" Runs the given function in a subprocess.
This wraps the given function in a gen.Task and runs it
in a multiprocessing.Pool. It is meant to be used as a
Tornado co-routine. Note that if func returns an Exception 
(or an Exception sub-class), this function will raise the 
Exception, rather than return it.
"""
result = yield gen.Task(pool.apply_async, func, args, kwargs)
if isinstance(result, Exception):
raise result
raise Return(result)
def handle_exceptions(func):
""" Raise a WrapException so we get a more meaningful traceback"""
@wraps(func)
def inner(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception:
raise WrapException()
return inner
# Test worker functions
@handle_exceptions
def test2(x):
raise Exception("eeee")
@handle_exceptions
def test(x):
print x
time.sleep(2)
return "done"
class TestHandler(tornado.web.RequestHandler):
@gen.coroutine
def get(self):
try:
result = yield async_run(test, "inside get")
self.write("%s\n" % result)
result = yield async_run(test2, "hi2")
except Exception as e:
print("caught exception in get")
self.write("Caught an exception: %s" % e)
finally:
self.finish()
app = tornado.web.Application([
(r"/test", TestHandler),
])
if __name__ == "__main__":
pool = TornadoPool(4)
app.listen(8888)
IOLoop.instance().start()

Voici comment il se comporte pour le client:

dan@dan:~$ curl localhost:8888/test
done
Caught an exception: 
Original traceback:
Traceback (most recent call last):
File "./mutli.py", line 123, in inner
return func(*args, **kwargs)
File "./mutli.py", line 131, in test2
raise Exception("eeee")
Exception: eeee

Et si j'envoie simultanément deux curl demandes, nous pouvons voir qu'ils sont manipulés de façon asynchrone sur le côté serveur:

dan@dan:~$ ./mutli.py 
inside get
inside get
caught exception inside get
caught exception inside get

Edit:

Notez que ce code devient plus simple avec Python 3, parce qu'elle introduit une error_callback argument mot-clé à tous asynchrone multiprocessing.Pool méthodes. Cela le rend beaucoup plus facile à intégrer avec Tornade:

class TornadoPool(Pool):
def apply_async(self, func, args=(), kwds={}, callback=None):
''' Asynchronous equivalent of `apply()` builtin
This version will call `callback` even if an exception is
raised by `func`.
'''
super().apply_async(func, args, kwds, callback=callback,
error_callback=callback)
@gen.coroutine
def async_run(func, *args, **kwargs):
""" Runs the given function in a subprocess.
This wraps the given function in a gen.Task and runs it
in a multiprocessing.Pool. It is meant to be used as a
Tornado co-routine. Note that if func returns an Exception
(or an Exception sub-class), this function will raise the
Exception, rather than return it.
"""
result = yield gen.Task(pool.apply_async, func, args, kwargs)
raise Return(result)

Tout ce que nous devons faire dans notre substituée apply_async est d'appeler le parent le error_callback argument mot-clé, en plus de la callback kwarg. Pas besoin de surcharger ApplyResult.

Nous pouvons obtenir même éleveur en utilisant une Métaclasse dans notre TornadoPool, pour permettre à son *_async méthodes à appeler directement comme s'ils étaient des coroutines:

import time
from functools import wraps
from multiprocessing.pool import Pool
import tornado.web
from tornado import gen
from tornado.gen import Return
from tornado import stack_context
from tornado.ioloop import IOLoop
from tornado.concurrent import Future
def _argument_adapter(callback):
def wrapper(*args, **kwargs):
if kwargs or len(args) > 1:
callback(Arguments(args, kwargs))
elif args:
callback(args[0])
else:
callback(None)
return wrapper
def PoolTask(func, *args, **kwargs):
""" Task function for use with multiprocessing.Pool methods.
This is very similar to tornado.gen.Task, except it sets the
error_callback kwarg in addition to the callback kwarg. This
way exceptions raised in pool worker methods get raised in the
parent when the Task is yielded from.
"""
future = Future()
def handle_exception(typ, value, tb):
if future.done():
return False
future.set_exc_info((typ, value, tb))
return True
def set_result(result):
if future.done():
return
if isinstance(result, Exception):
future.set_exception(result)
else:
future.set_result(result)
with stack_context.ExceptionStackContext(handle_exception):
cb = _argument_adapter(set_result)
func(*args, callback=cb, error_callback=cb)
return future
def coro_runner(func):
""" Wraps the given func in a PoolTask and returns it. """
@wraps(func)
def wrapper(*args, **kwargs):
return PoolTask(func, *args, **kwargs)
return wrapper
class MetaPool(type):
""" Wrap all *_async methods in Pool with coro_runner. """
def __new__(cls, clsname, bases, dct):
pdct = bases[0].__dict__
for attr in pdct:
if attr.endswith("async") and not attr.startswith('_'):
setattr(bases[0], attr, coro_runner(pdct[attr]))
return super().__new__(cls, clsname, bases, dct)
class TornadoPool(Pool, metaclass=MetaPool):
pass
# Test worker functions
def test2(x):
print("hi2")
raise Exception("eeee")
def test(x):
print(x)
time.sleep(2)
return "done"
class TestHandler(tornado.web.RequestHandler):
@gen.coroutine
def get(self):
try:
result = yield pool.apply_async(test, ("inside get",))
self.write("%s\n" % result)
result = yield pool.apply_async(test2, ("hi2",))
self.write("%s\n" % result)
except Exception as e:
print("caught exception in get")
self.write("Caught an exception: %s" % e)
raise
finally:
self.finish()
app = tornado.web.Application([
(r"/test", TestHandler),
])
if __name__ == "__main__":
pool = TornadoPool()
app.listen(8888)
IOLoop.instance().start()

InformationsquelleAutor dano

1

Si vos demandes sont à prendre qu'une tornade est le bon cadre.

Je vous suggère d'utiliser nginx pour acheminer le jeûne devient à la tornade et les plus lents, vers un autre serveur.

PeterBe a un article intéressant où il dirige plusieurs Tornade des serveurs et des jeux de l'un d'entre eux d'être "le lent" pour gérer les demandes en cours d'exécution, voir: inquiétant-sur-io-blocage je voudrais essayer cette méthode.
- Je peux facilement imaginer qu'il utilise Tornade de faire une API pour entamer le long demandes en cours d'exécution et de retourner le résultat. Aussi, si lente et rapide, le travail est lié?
- non ,les demandes sont plus lents car ils sont occupés à calculer les factorielles des grands nombres .
InformationsquelleAutor andy boot

Vous devez vous connecter pour publier un commentaire.