Problème avec multi threaded Python application et les connexions socket

Je me demande si l'écriture de cette aide tordu pourrait aider? Quelqu'un pourrait-il montrer ce que mon exemple ressemblerait écrite à l'aide de tordu?

Cette variante est beaucoup plus rapide que le code qui utilise gevent:

#!/usr/bin/env python
import sys
from timeit import default_timer as timer
from twisted.internet import defer, protocol, reactor, ssl, task
from twisted.python   import log
info = log.msg
class NoopProtocol(protocol.Protocol):
def makeConnection(self, transport):
transport.loseConnection()
def connect(host, port, contextFactory=ssl.ClientContextFactory(), timeout=30):
info("connecting %s" % host)
cc = protocol.ClientCreator(reactor, NoopProtocol)
d = cc.connectSSL(host, port, contextFactory, timeout)
d.addCallbacks(lambda _: info("done %s" % host),
lambda f: info("error %s reason: %s" % (host, f.value)))
return d
def n_at_a_time(it, n):
"""Iterate over `it` concurently `n` items at a time.
`it` - an iterator creating Deferreds
`n`  - number of concurrent iterations
return a deferred that fires on completion
"""
return defer.DeferredList([task.coiterate(it) for _ in xrange(n)])
def main():
try:
log.startLogging(sys.stderr, setStdout=False)
info("getting hostname list")
hosts_file = sys.argv[1] if len(sys.argv) > 1 else "hosts.txt"
hosts_list = open(hosts_file).read().splitlines()
info("spawning jobs")
start = timer()        
jobs = (connect(host, 443, timeout=2) for host in hosts_list)
d = n_at_a_time(jobs, n=20) # limit number of simultaneous connections
d.addCallback(lambda _: info("%d hosts took us %.2g seconds" % (
len(hosts_list), timer() - start)))
d.addBoth(lambda _: (info("the end"), reactor.stop()))
except:
log.err()
reactor.stop()
if __name__=="__main__":
reactor.callWhenRunning(main)
reactor.run()

Voici une variante qui utilise t.j'.d.inlineCallbacks. Il nécessite Python 2.5 ou plus récente. Il permet d'écrire du code asynchrone en mode synchrone (blocage):

#!/usr/bin/env python
import sys
from timeit import default_timer as timer
from twisted.internet import defer, protocol, reactor, ssl, task
from twisted.python   import log
info = log.msg
class NoopProtocol(protocol.Protocol):
def makeConnection(self, transport):
transport.loseConnection()
@defer.inlineCallbacks
def connect(host, port, contextFactory=ssl.ClientContextFactory(), timeout=30):
info("connecting %s" % host)
cc = protocol.ClientCreator(reactor, NoopProtocol)
try:
yield cc.connectSSL(host, port, contextFactory, timeout)
except Exception, e:
info("error %s reason: %s" % (host, e))
else:
info("done %s" % host)
def n_at_a_time(it, n):
"""Iterate over `it` concurently `n` items at a time.
`it` - an iterator creating Deferreds
`n`  - number of concurrent iterations
return a deferred that fires on completion
"""
return defer.DeferredList([task.coiterate(it) for _ in xrange(n)])
@defer.inlineCallbacks
def main():
try:
log.startLogging(sys.stderr, setStdout=False)
info("getting hostname list")
hosts_file = sys.argv[1] if len(sys.argv) > 1 else "hosts.txt"
hosts_list = open(hosts_file).read().splitlines()
info("spawning jobs")
start = timer()        
jobs = (connect(host, 443, timeout=2) for host in hosts_list)
yield n_at_a_time(jobs, n=20) # limit number of simultaneous connections
info("%d hosts took us %.2g seconds" % (len(hosts_list), timer()-start))
info("the end")
except:
log.err()
finally:
reactor.stop()
if __name__=="__main__":
reactor.callWhenRunning(main)
reactor.run()

OriginalL'auteur jfs

Python 3.4 présente les nouvelles provisoire de l'API asynchrones IO -- asyncio module.

Cette approche est similaire à twistedde réponse:

#!/usr/bin/env python3.4
import asyncio
import logging
from contextlib import closing
class NoopProtocol(asyncio.Protocol):
def connection_made(self, transport):
transport.close()
info = logging.getLogger().info
@asyncio.coroutine
def connect(loop, semaphor, host, port=443, ssl=True, timeout=15):
try:
with (yield from semaphor):
info("connecting %s" % host)
done, pending = yield from asyncio.wait(
[loop.create_connection(NoopProtocol, host, port, ssl=ssl)],
loop=loop, timeout=timeout)
if done:
next(iter(done)).result()
except Exception as e:
info("error %s reason: %s" % (host, e))
else:
if pending:
info("error %s reason: timeout" % (host,))
for ft in pending:
ft.cancel()
else:
info("done %s" % host)
@asyncio.coroutine
def main(loop):
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
limit, timeout, hosts = parse_cmdline()
# connect `limit` concurrent connections
sem = asyncio.BoundedSemaphore(limit)
coros = [connect(loop, sem, host, timeout=timeout) for host in hosts]
if coros:
yield from asyncio.wait(coros, loop=loop)
if __name__=="__main__":
with closing(asyncio.get_event_loop()) as loop:
loop.run_until_complete(main(loop))

Ainsi que twisted variante il utilise NoopProtocol qui ne fait rien mais se déconnecte immédiatement lors de la connexion réussie.

Nombre de connexions simultanées est limité à l'aide d'un sémaphore.

Le code est coroutine à base de.

Exemple

Pour savoir comment beaucoup de succès connexions ssl nous pouvons faire pour les 1000 premiers hôtes de haut millions Alexa liste:

$ curl -O http://s3.amazonaws.com/alexa-static/top-1m.csv.zip
$ unzip *.zip
$ /usr/bin/time perl -nE'say $1 if /\d+,([^\s,]+)$/' top-1m.csv | head -1000 |\
python3.4 asyncio_ssl.py - --timeout 60 |& tee asyncio.log

Le résultat est moins de la moitié de toutes les connexions sont couronnées de succès. En moyenne, il vérifie ~20 hôtes par seconde. De nombreux sites expiré au bout d'une minute. Si l'hôte ne correspond pas à des noms d'hôtes à partir de certificat du serveur la connexion échoue également. Il comprend example.com vs www.example.com -comme les comparaisons.

OriginalL'auteur jfs