thread code in async umwandeln?

[Scholastik]

Hallo

ich fange gerade an mich mit "async/asyncio" zu beschäftigen, es scheint leider doch deutlich schwerer zu erlernen, als ich gedacht habe (in test-codes gibts ständig fehlermeldungen wie "nicht awaited" oder "das muss in async funktion rein", ohne dass ich verstehe warum das so sein muss)

Jedenfalls ist mein Hauptanliegen, in einem synchron laufendem code, async zu verwenden, um gleichzeitig x REST API calls zu machen.
Aktuell mache ich das simpel mit threads. Nur leider kommt es ab und zu auch vor, dass ich 1000+ API calls möglichst gleichzeitig machen muss. Und wenn ich das mit Threads versuche, gibts logischerweise den Fehler, dass keine neuen threads gestartet werden können, weil schon zuviele aktiv sind. Nun könnte ich stattdessen einfach immer nur 500 calls aufeinmal machen und das nacheinander... aber ich glaube async wäre hier die bessere Wahl, oder?

Nur wie verändere ich folgenden beispiel code so, dass die calls mit async funktionieren? (Ich habe aktuell den Eindruck, dass async nur geht, wenn der vollständige code auf async basiert, doch ich will den hauptcode eigentlich weiterhin synchron haben, da der ohne die REST Ergebnisse sowieso nicht weiterlaufen kann):

Code: Alles auswählen

import concurrent.futures
import time
import requests

class Test:
    
    def __init__(self):
        self.url = "https://api.github.com/events" # example url
        self.session = requests.session()
        self.Pool = concurrent.futures.ThreadPoolExecutor(500)
    
    def make_call(self,i): # requests/aiohttp code to do the REST calls
        return self.session.get(url=self.url,params={"param":i})
        
    def run(self):
        while True:
            # make stuff
            results = []
            futures = []
            for i in range(1000): # doing 1000 REST API calls as fast as possible
                futures.append( self.Pool.submit( self.make_call, i) )
            for future in futures:
                results.append(future.result())
            # do stuff with results
            time.sleep(2) # start over again in some seconds.
            
Test().run()

Die funktion "run" soll also mit while True so bleiben und soll synchron weiterlaufen. Aber die 1000 API calls sollen asynchron mit async abgefragt werden (sodass deren gesamt-abfrag-zeit möglichst gering ausfällt, vermutlich geringer als mit threads)

Außerdem kam ich mit aiohttp noch nicht ganz klar, denn jedes Beispiel was ich dazu finde, lässt sich glaube ich nicht auf meinen code anwenden. Denn da wird entweder mit context manager gearbeitet um eine session zu öffnen/schließen oder halt am Ende des programms "await session.close()" aufgerufen (welches nur innerhalb einer async funktion geht). Aber ich möchte so eine clientsession ja nur einmal zu start des skripts öffnen und dann soll diese offen bleiben, bis ich das ganze system wieder beende. Die session soll also nicht jeden "while True" Durchlauf neu geöffnet und geschlossen werden.

[__blackjack__]

@Scholastik: Ich sehe bei dem Code nicht warum der in einer Klasse stecken sollte. Das ist im Grunde nur eine Funktion die auf drei Methoden aufgeteilt wurde, warum auch immer. Wenn man das nicht macht, dann stellt sich auch das `session`-Problem nicht. Dann kann man das mit ``with`` verwenden. Also auch schon das normale `requests.Session`-Objekt. Und auch den Pool. Die `requests.session()`-Funktion sollte man auch nicht mehr verwenden. Die ist nur noch aus Gründen der Rückwärtskompatibilität da. Ich würde erwarten, das die irgendwann verschwinden wird.

Zwischenstand:

Code: Alles auswählen

#!/usr/bin/env python3
import concurrent.futures
import time

import requests


def make_call(session, url, i):
    return session.get(url=url, params={"param": i})


def run():
    url = "https://api.github.com/events"  # example url
    with requests.session() as session:
        with concurrent.futures.ThreadPoolExecutor(500) as pool:
            while True:
                futures = []
                #
                # Doing 1000 REST API calls as fast as possible.
                #
                for i in range(1000):
                    futures.append(pool.submit(make_call, session, url, i))

                results = [future.result() for future in futures]

                # Do stuff with results.

                time.sleep(2)  # Start over again in some seconds.


def main():
    run()


if __name__ == "__main__":
    main()

Das lässt sich jetzt vielleicht einfacher auf aiohttp umschreiben. Allerdings ist async nicht dazu da Programme schneller zu machen, sonder um mehr in einem Thread zu bedienen als das mit vielen Threads möglich wäre. Man sollte da jetzt keine Wunder erwarten.

Und der Code enthielt keine Ausnahmebehandlung. Da müsstest Du auch noch drüber nachdenken, egal mit welcher Technik Du die Nebenläufigkeit umsetzen möchtest.

[Scholastik]

Danke dir für deine Antwort

*siehe unten (der Übersicht halber)

Also zurück zur eigentlichen Frage:
Die Idee den context manager über die while schleife zu packen war schonmal gut.
Gehen wir also nun von folgendem code aus:

Code: Alles auswählen

#!/usr/bin/env python3
import concurrent.futures
import time
import requests
class Test:
    def __init__(self):
        self.url = "https://api.github.com/events" # example url

    def make_call(self,session,i): # requests/aiohttp code to do the REST calls
        session.get(url=self.url,params={"param":i})
        
    def run(self):
        with requests.session() as session:
            with concurrent.futures.ThreadPoolExecutor(500) as pool:
                while True:
                    futures = []
                    for i in range(1000): # doing 1000 REST API calls as fast as possible
                        futures.append( Pool.submit( self.make_call,session,i) )
                    results = [future.result() for future in futures]
                    time.sleep(2) # start over again in some seconds.
    def main(self):
        self.run()
        
if __name__ == "__main__":
    Test().main()

Wenn wir das nun auf async und aiohttp umändern wollen, dann muss "run" ja offenbar eine async funktion sein, da session laut doku nur in einer solchen Funktion geöffnet werden soll. Doch um eine async Funktion aufzurufen, muss man sie mit "await run()" aufrufen, da man sonst "not awaited" als Fehler bekommt. Doch "await" lässt sich ebenfalls nur in einer async Funktion verwenden -> da beißt sich die Katze doch in den Schwanz?!
Dies stellt mein Verständnisproblem von async da. Könnt ihr das entwirren?

Ich möchte nicht alles an code in async umwandeln, sondern nur den teil, der für die REST calls zuständig ist.



*[spoiler] (warum gibt es hier keine spoiler tags?)
1) requests.session() : wo/wie wird man denn darüber informiert, ob etwas noch verwendet werden sollte oder nicht? Hast du nen Link, was man stattdessen verwenden sollte? In der requests doku finde ich auf die schnelle nur den beispielcode welcher session verwendet.

2) Ob Klasse oder nicht, hängt noch vom Gesamtdesign des eigentlichen Programms ab denke ich (das hier ist ja nur beispielcode). Das würde aber viel zu weit führen, ich denke aber ich weiß mittlerweile genug um zu sagen, dass eine Klasse in meinem Fall notwendig ist, daher ist das etwas woran leider nicht gerüttelt werden kann und eine Diskussion darüber ist nicht zielführend.

3) async "code schneller machen": das hab ich nicht geschrieben. Mir ist durchaus bewusst, dass async code nicht schneller macht. Stattdessen nutzt es Wartezeiten effizient, indem es in der Zeit mit anderen Baustellen weiter macht. Die Wartezeit die ich sinnvoll nutzen möchte, ist die Zeit, die ich auf die Antwort der REST API warte. Das heißt ich stelle mir das so vor, dass: ein call abgeschickt wird und anstatt dass wir nun auf die antwort warten, schicken wir den nächsten call auch schon ab usw. bis alle 1000 calls abgeschickt wurden. und erst dann warten wir auf die Antworten. Genau das sollte ein Sinn und Zweck von aiohttp sein. Oder nicht?

4) Ausnahmebehanldungen: Wie gesagt, Beispielcode den ich mal eben schnell hingezimmert habe, um den Aufbau zu verdeutlichen. Aber danke für den Hinweis, wie Ausnahmebehandlung bei async/aiohttp funktioniert, weiß ich tatsächlich noch nicht.
[/spoiler]

[kbr]

Doch um eine async Funktion aufzurufen, muss man sie mit "await run()" aufrufen, da man sonst "not awaited" als Fehler bekommt. Doch "await" lässt sich ebenfalls nur in einer async Funktion verwenden -> da beißt sich die Katze doch in den Schwanz?!

Asynchroner Code muß in einer Event-Loop laufen, die über asyncio gestartet werden kann.

Code: Alles auswählen

import asyncio

async def main():
    # run more async stuff here ...
    
    
asyncio.run(main())

Vielleicht möchtest Du in dem Zusammenhang auch einen Blick auf 'httpx' statt 'requests' werfen.

[Scholastik]

Ich hab mal den async/aiohttp code geschrieben, wie er zumindest keine Fehler raised, aber dennoch nciht das ist was ich möchte:

Code: Alles auswählen

#!/usr/bin/env python3
import time
import aiohttp
import asyncio
class Test:
    def __init__(self):
        self.url = "https://api.github.com/events" # example url

    async def make_x_calls(self,num):
        async with aiohttp.ClientSession() as session:
            for i in range(num):
                async with session.get(self.url) as response:
                    return await response.read()
   
    def run(self):
        while True:
            results = asyncio.run(self.make_x_calls(1000))# doing 1000 REST API calls as fast as possible
            time.sleep(2) # start over again in some seconds.
        
if __name__ == "__main__":
    Test().run()

Ist das soweit korrekt?
Doch so wird aktuell für jeden "while True" Durchlauf eine ClienSession geöffnet, x calls gemacht, und wieder geschlossen. Doch es wäre doch deutlich effizienter, wenn diese Session die ganze Zeit offen bleiben könnte, solange die while True Schleife läuft. Dazu müsste man den context manager wieder über die Schleife packen, aber das geht ja nicht, ohne das die synchrone funktion asynchron werden muss, was ich ja aber nicht möchte.


@kbr:
du meinst also httpx ist besser als aiohttp (async) und requests (sync), warum? Auf der github seite sehe ich eine warnung bzgl Beta, vllt sollte ich da noch warten, bis es released ist.
Die Problemstellung bleibt aber auch bei httpx dieselbe, wie behalte ich ein und dieselbe session, ohne dass ich meinen synchronen code mit async versehen muss.

edit:
20 minuten nach abschicken des posts hab ich gemerkt, dass asyncio.run() tatsächlich etwas returned. Ich nehme an das sind die results der api calls in dem Fall?

[kbr]

Ich hatte nicht geschrieben, dass etwas besser oder schlechter ist. Aber 'httpx' kann async und 'requests' kann das nicht – vielleicht kommt Dir das ja gelegen. Der Rückgabewert von asyncio.run() ist das Ergebnis der aufgerufenen Coroutine; das steht auch so in der Doku.

[__blackjack__]

@Scholastik: Ob etwas veraltet ist steht üblicherweise in der Dokumentation. Und da wird nirgendwo `session()` verwendet. Die Funktion scheint da nicht mal mehr dokumentiert zu sein. Welche Dokumentation verwendest Du denn?

Einzige Erwähnung ist bei Migration zu Version 1.x (zweiter Punkt bei „API Changes“): https://2.python-requests.org/en/master ... ing-to-1-x

Die Dokumentation direkt am Funktionsobjekt lautet wie folgt:

Code: Alles auswählen

In [86]: requests.session?                                                      
Signature: requests.session()
Docstring:
Returns a :class:`Session` for context-management.

.. deprecated:: 1.0.0

    This method has been deprecated since version 1.0.0 and is only kept for
    backwards compatibility. New code should use :class:`~requests.sessions.Session`
    to create a session. This may be removed at a future date.

:rtype: Session

[Scholastik]

bzgl requests:
Also abgesehen von der Großschreibweise, sehe ich nichts bzgl deprecated. hier wird es als advanced usage beschrieben session zu verwenden und daher mach ich das halt auch so: https://2.python-requests.org/en/master/user/advanced/

Aber wäre es möglich, dass mir jemand bei der ursprünglichen Fragestellung hilft?

Ich habe beim Testen schon rausgefunden, dass ich wohl asyncio.gather() verwenden muss, da die calls ansonsten nicht korrekt gleichzeitig ausgeführt werden.
Aber die Problematik mit der Session, die ich gerne für die komplette Laufzeit haben würde, konnte ich nicht lösen.
Ich habe mir auch httpx angeguckt, aber dafür bekomme ich keinen funktionierenden code zum laufen... wenn ich einfach 1:1 aiotthp.ClientSession durch httpx.AsyncClient ersetze, bekomme ich eine Fehlermeldung, die mir was von asyncio stream.close() erzählt, damit kann ich leider nichts anfangen.


Also hier mal mein code mit httpx (wenn ich httpx durch aiohttp ersetze, funktioniert der code):

Code: Alles auswählen

#!/usr/bin/env python3
import time
import asyncio
import httpx # alternative zu requests und aiohttp, kann sowohl sync als auch async (client = httpx.AsyncClient() bzw client = httpx.Client())
class Test:
    def __init__(self):
        self.url = "https://example.com" # example url
        self.async_session = httpx.AsyncClient()

    async def async_make_call_httpx(self,session): # code to do the REST calls
        return await session.get(url=self.url)
    
    async def async_make_x_calls_gather_httpx(self,num):
        async with httpx.AsyncClient() as session:
            return await asyncio.gather(
                    # *[self.async_make_call_httpx(self.async_session) for i in range(num)]
                    *[self.async_make_call_httpx(session) for i in range(num)]
                )
   
    def run(self):
        while True:
            results = asyncio.run(self.async_make_x_calls_gather_httpx(5))# doing 1000 REST API calls as fast as possible
            time.sleep(2) # start over again in some seconds.
    
    def main(self):
        self.run()
        
if __name__ == "__main__":
    Test().main()

und das ist der Fehler, den ich nach 1 oder 2 while True durchläufen bekomme:

PS C:\Users\Serpens66\desktop> python ttt.py
An open stream object is being garbage collected; call "stream.close()" explicitly.
Exception ignored in: <bound method StreamReaderProtocol._on_reader_gc of <asyncio.streams.StreamReaderProtocol object at 0x044E5B38>>
Traceback (most recent call last):
File "C:\Users\Serpens66\AppData\Local\Programs\Python\Python38-32\lib\asyncio\streams.py", line 245, in _on_reader_gc
transport.abort()
File "C:\Users\Serpens66\AppData\Local\Programs\Python\Python38-32\lib\asyncio\sslproto.py", line 400, in abort
self._ssl_protocol._abort()
File "C:\Users\Serpens66\AppData\Local\Programs\Python\Python38-32\lib\asyncio\sslproto.py", line 732, in _abort
self._transport.abort()
File "C:\Users\Serpens66\AppData\Local\Programs\Python\Python38-32\lib\asyncio\proactor_events.py", line 417, in abort
self._force_close(None)
File "C:\Users\Serpens66\AppData\Local\Programs\Python\Python38-32\lib\asyncio\proactor_events.py", line 151, in _force_close
self._loop.call_soon(self._call_connection_lost, exc)
File "C:\Users\Serpens66\AppData\Local\Programs\Python\Python38-32\lib\asyncio\base_events.py", line 711, in call_soon
self._check_closed()
File "C:\Users\Serpens66\AppData\Local\Programs\Python\Python38-32\lib\asyncio\base_events.py", line 504, in _check_closed
raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

Wenn ich stattdessen die auskommentierte Zeile verwende, also die "self.async_session" (und dafür den context manager mit session auskommentiere), dann gibts auch einen Fehler:

PS C:\Users\Serpens66\desktop> python ttt.py
Fatal error on SSL transport
protocol: <asyncio.sslproto.SSLProtocol object at 0x037C5CB8>
transport: <_ProactorSocketTransport fd=796 read=<_OverlappedFuture cancelled>>
Traceback (most recent call last):
File "C:\Users\Serpens66\AppData\Local\Programs\Python\Python38-32\lib\asyncio\sslproto.py", line 685, in _process_write_backlog
self._transport.write(chunk)
File "C:\Users\Serpens66\AppData\Local\Programs\Python\Python38-32\lib\asyncio\proactor_events.py", line 359, in write
self._loop_writing(data=bytes(data))
File "C:\Users\Serpens66\AppData\Local\Programs\Python\Python38-32\lib\asyncio\proactor_events.py", line 395, in _loop_writing
self._write_fut = self._loop._proactor.send(self._sock, data)
AttributeError: 'NoneType' object has no attribute 'send'

Bei zweiterem Ansatz habe ich schon fast einen Fehler erwartet, da ich session außerhalb einer async funktion erstellt habe.
Aber der erste ansatz, also so wie ihr oben den code seht, der müsste doch eigentlich funktionieren, oder nicht?!

[kbr]

Das ist jetzt weder recherchiert noch getestet, aber meiner Erinnerung nach liefern die Aufrufe von session.get() bereits Coroutinen, die sich an gather übergeben lassen:

Code: Alles auswählen

async with httpx.AsyncClient() as session:
    responses = await asyncio.gather(*[session.get(url) for url in urls])

Wie Du asyncio.run() aufrufst, ist allerdings fehlerträchtig. Dieser Aufruf ist in einer Klasse implementiert, von der möglicherweise mehrere Instanzen angelegt werden könnten. Pro Thread darf aber nur eine asyncio Mainloop laufen, d.h. dieser Code wartet nur darauf, unerwartete Resultate zu liefern.

[Scholastik]

Das ist jetzt weder recherchiert noch getestet, aber meiner Erinnerung nach liefern die Aufrufe von session.get() bereits Coroutinen, die sich an gather übergeben lassen:

Code: Alles auswählen

async with httpx.AsyncClient() as session:
    responses = await asyncio.gather(*[session.get(url) for url in urls])

Wie Du asyncio.run() aufrufst, ist allerdings fehlerträchtig. Dieser Aufruf ist in einer Klasse implementiert, von der möglicherweise mehrere Instanzen angelegt werden könnten. Pro Thread darf aber nur eine asyncio Mainloop laufen, d.h. dieser Code wartet nur darauf, unerwartete Resultate zu liefern.

Danke, das vereinfacht den Code ein wenig, aber leider kommt noch immer dieselbe Fehlermeldung mit dem "stream.close()".

Auch danke für den Hinweis, dass das nur einmal pro Thread laufen darf, das war mir nicht bewusst... ja es sollen tatsächlich mehrere instanzen laufen, das allerdings gleichzeitig und das wird aktuell über threads umgesetzt. Daher sollte "einmal pro thread" noch funktionieren. Werde es mir aber aufjedenfall notieren, dass ich daran denke.

[__deets__]

Python kann eh keine mehreren threads wirklich gleichzeitig ausführen. Wenn du schon async gehst, dann kannst du es auch gleich 100%ig machen. Und dir den thread overhead sparen.

[Scholastik]

Python kann eh keine mehreren threads wirklich gleichzeitig ausführen. Wenn du schon async gehst, dann kannst du es auch gleich 100%ig machen. Und dir den thread overhead sparen.

abgesehen von dem enormen aufwand:
Wäre das denn sinnvoll, wenn der komplette code garnicht asynchron laufen soll?

Der Hauptteil besteht halt aus dieser while True Schleife. Innerhalb dieser gibt es mehrere Stellen an denen mehrere API Calls gleichzeitig gemacht werden sollen, dazwischen/danach dann immer wieder Berechnungen, die teilweise recht lange dauern. Diese Berechnungen können nicht ohne die vorangegangen API Calls gemacht werden, weshalb das sowieso nacheinander laufen muss.
Abgesehen von einem variablem sleep() am Ende der Schleife, gibt es also außer bei den API Calls keinerlei Wartezeit die sinnvoll mit async überbrückt werden kann, dennoch müssen alle instanzen zeitgleich laufen, nicht nacheinander.
Daher sehe ich hier wirklich keinen einzigen guten Grund, weshalb ich alles auf async umschreiben sollte?!

[__blackjack__]

@Scholastik: Die *Funktion* `session()` — kleines `s` — ist „deprecated“. Statt diese Funktion zu verwenden um ein `Session`-Objekt — grosses `S` weil keine Funktion sondern eine Klasse — zu erstellen, erstellt man direkt ein `Session`-Objekt.

Wo das mit dem „deprecated“ steht habe ich in meinen letzten Beitrag kopiert: Im Docstring der `session()`-Funktion. Die Funktion taucht wie schon gesagt mittlerweile nicht mal mehr in der Dokumentation auf, was wirklich ein Zeichen ist sie besser nicht mehr zu benutzen. Das hast Du aber gemacht, also nicht das was in der Dokumentation steht.

Zum letzte Satz im letzten Beitrag: Wenn Du keinen Grund siehst das mit `async` zu machen, dann lass es doch einfach. Die Initiative ging ja von Dir aus, das zu machen.

Es ist auf jeden Fall keine gute Idee davon auszugehen man könnte existierenden thread-basierten Code nehmen, und da die Threads raus und ein paar ``async``/``await`` reinzuwerfen ohne sich mit dem Konzept dahinter zu beschäftigen. Es ist keine „rocket science“, aber eben schon ein anderes vorgehen, eine andere Denkweise, in die man sich mal kurz einarbeiten muss.

[__deets__]

Das “alles” bezog sich auf die Mischung von threads und async. Die ist sinnlos. Das du ab einem bestimmten Moment mit allen Ergebnissen synchron rechnest ist davon doch völlig unberührt. Wenn zu dem Zeitpunkt keine Abfragen gemacht werden, denn zwingt dich weder threading noch async dazu.

[sparrow]

@Scholastik es nützt nichts, wenn du aus irgendwelchen herbeigedachten Performancegründen versucht krampfhaft async in dein Programm zu bringen. Du musst verstehen, was der Unterschied zwischen Multithreading und event-basiertem Programmfluss ist. Dann weißt du auch, warum es unsinnig ist, beides zu mischen.
Dafür muss man sich aber mit Dingen auseinandersetzen. Auffällig ist, dass du aber alle Vorschläge erst einmal als falsch abtust weil offensichtlich der Helfende nicht versteht?!

[Scholastik]

@Scholastik es nützt nichts, wenn du aus irgendwelchen herbeigedachten Performancegründen versucht krampfhaft async in dein Programm zu bringen. Du musst verstehen, was der Unterschied zwischen Multithreading und event-basiertem Programmfluss ist. Dann weißt du auch, warum es unsinnig ist, beides zu mischen.
Dafür muss man sich aber mit Dingen auseinandersetzen. Auffällig ist, dass du aber alle Vorschläge erst einmal als falsch abtust weil offensichtlich der Helfende nicht versteht?!

Offtopic:
ja das ist allgemein das Problem von diesem Forum, oder ein seit jeher existierendes Problem zwischen Experten und Anfängern. Sie verstehen sich gegenseitig nicht.
Ich kenne das selbst aus lua und modding von computerspielen. Mittlerweile kenne ich dort auch einige "Experten"-programmierweisen und wenn ich sehe wie ein Anfänger fragt wie er seinen code fixen kann, dann neige ich leider auch oft dazu sowas zu schreiben wie ihr "das ist alles unfug, du musst das völlig anders angehen".
Aber im Gegensatz zu euch, schreibe ich ihm dann auch fertigen Code und erkläre schritt für schritt und kleinteilig, welche Zusammenhänge der Anfänger wissen muss, damit er versteht, warum es zb sinnvoller ist einen loop zu verwenden, als jeden befehl 20 mal hinzuschreiben (simples beispiel) und versuche ihm die Angst vor diesem unbekanntem Hilfsmittel zu nehmen... eben weil ich selbst weiß, wie es ist Anfänger in einem Gebiet zu sein, man erkennt die Zusammenhänge nicht, und ohne augezeichnete Erklärung hilft es leider 0,garnichts, wenn man als Experte dann nur herabschaut und sagt "dein code ist scheiße, du bist dumm" -> so hört es sich hier für mich an.
Ich finde das wirklich sehr schade, bitte versucht euch daran zu erinnern, wie es war als ihr Anfänger mit etwas wart und um Hilfe bitten musstet. falla es so eine Zeit bei euch gab.

Topic:
Ich bin mir eigentlich ziemlich sicher, dass ich den unterscheid zwischen event und multithread und reihenbasietem Code sehr gut verstehe.
Dennoch verstehe ich nicht, warum ihr behauptet, dass es schlecht wäre beides zu komibinieren? async ud threads haben beide Vor- und Nachteile. Ist es nicht offentsichtlich, dass man versucht es zu kombinieren um die Vorteile aus beidem zu erhalten? Ich sehe nicht warum das falsch sein sollte?
Wie ich eindeutig beschrieben habe, geht es mir NUR darum, die Wartezeit zwischen den REST API Calls sinnvoll zu nutzen. Und dafür ist async bewiesenermaßen millionenmal besser als jede thread lösung die es geben kann -> logisch dass ich async verwenden will. Der Rest soll aber nicht asynchron/eventbasiert sein, ich habe mir darüber schon sehr oft Gedanken gemacht, eventbasiert funktioniert in meinem Fall nicht. Async hat also einen perfekten Anwendungsfall und für den Rest sind threads der perfekte anwendungfall (da die instanzen gleichzeitig laufen müssen, ungeachtet dessen, dass sie durhc die Aufteilung der Ressourcen länger brauchen).

So. schafft es nun irgendeiner von euch auf meinen Absatz in der Mitte einzugehen und mir schritt für schritt argumentativ zu erklären wo meine Denkfehler sind?

[__deets__]

Ich hab’s dir schon gesagt. Threads laufen in Python nicht parallel. Das Stichwort dazu ist das GIL. Python Code ist also IMMER komplett seriell. Ein Grund threads zu benutzen besteht darin, blockierende IO im und Berechnungen “quasi”-parallel zu machen - was man mit async eben auch macht. Der eine wird vom OS aus dem scheduling geworfen bis sich was getan hat, und konsumiert Stack und andere Systemresourcen. Der andere gibt die Zeit freiwillig ab und konsumiert weniger Ressourcen. Die beiden Ansätze zu mischen ist also sinnlos, wenn man schon in async investiert hat.

[DeaD_EyE]

Eins vorweg, die Lazenz ist bei asyncio generell höher, als bei synchronen Prozessen, die in Threads laufen.
Bitte keine Wunder erwarten.


Minimale Implementierung mit asyncio und aiohttp:

Code: Alles auswählen

import asyncio
import aiohttp


async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()


async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, "http://localhost:8000/")] * 1000
        return await asyncio.gather(*tasks, return_exceptions=True)


if __name__ == "__main__":
    results = asyncio.run(main())
    print(len(results))
    print(results[0])

Dann starte einen lokalen Webserver.
Kannst du auch mit Python: python -m http.server 8000

Den Code dann einmal mit Webserver laufen lassen und einmal ohne.
return_exceptions sorgt dafür, dass auch Exceptions mit in das Ergebnis einfließen und nicht abgebrochen wird.

Das hat mir weitergeholfen die Werkzeuge in asyncio zu verstehen: https://hynek.me/articles/waiting-in-asyncio/

[Sirius3]

Es ist schwierig, herauszufinden, was ein Fragender bereits weiß und was noch nicht. Die allgemeine Floskel "ich bin Anfänger" hat leider oft den Beiklang, "schreibt mir bitte mein Programm, denn ich habe keine Lust dazu, selbst was zu lernen".
Asynchrone Programmierung ist nicht einfach, und Thread-Programmierung (wobei Du wahrscheinlich sowieso eher Multiprocessing brauchst) auch nicht.
Beides kann man nicht mischen, sondern muß es sauber getrennt voneinander programmieren.
Also eine Hauptschleife, die asynchron arbeitet und noch Worker, die parallel arbeiten. Das ganze muß dann über Queues miteinander verknüpft werden.
Bei Dir kommt zusätzlich noch das Problem, dass Sockets keine unbegrenzte Resource sind und Webseitenanbieter es gar nicht gerne sehen, wenn tausende Anfragen von der selben IP gleichzeitig kommen.

Zurück zum Problem: wenn man sich in eine neue Technik einarbeitet, muß man erst die grundsätzliche Struktur verstehen. Bei asyncio ist das eben die EINE Hauptschleife, die die einzelnen Coroutinen steuert. Dass Du die einzelnen Teile auf mehrere Klassenmethoden aufgeteilt hast, macht es noch schwieriger zu verstehen.

Hier ist so ziemlich genau das Problem erklärt, das Du hast: https://pawelmhm.github.io/asyncio/pyth ... ohttp.html
Glück gehabt, dass jemand so freundlich war, mit viel Mühe so einen Artikel zu schreiben, denn in einem Forum findest Du kaum jemand, der sich Stundenlang hinsetzt, um die Problem anderer Leute zu lösen. Bedenke immer, hier wird unentgeltlich freiwillig geholfen.


@DeaD_EyE: warum sollte die Latenz immer höher sein? Intern werden die selben Mechanismen verwendet, "stoppe den Thread so lange, bis Daten am Socket ankommen", im einen Fall wird nur an einem Socket gelauscht, im im async-Fall an vielen. Da dies vom Betriebssystem gemacht wird, ist die Latenz aber in beiden Fällen exakt gleich, nur dass viele Threads viel mehr Resourcen belegen und falls tatsächlich etwas zu tun ist, sie sich auch noch gegenseitig die CPU-Zeit stehlen, statt die Daten brav nacheinander, so wie sie ankommen, zu verarbeiten.

[__blackjack__]

Wobei der Vergleich den der Artike von Paweł Miechl macht, nicht wirklich interessant ist. Das async hier effizienter als rein sequentiell ist, war ja irgendwie auch ohne den Vergleich sonnenklar. Interessanter wäre async mit Threads oder Prozessen zu vergleichen. Und sinnvolle Fehler- bzw. Ausnahmebehandlung fehlt auch.