Asynchrone Funktionen (asyncio)

[mechanicalStore]

Hallo,

Code: Alles auswählen

#!/usr/bin/env python

import time

def main():

    bits = [1,1,0,0,0,1,1,0,1,0,1,1,1,0,0,0,0,0,1,0,0,1,1,0,0,1]

    for n in bits:
        print(n, end = ',')
        # Do anything more with the value
        time.sleep(0.2)

    print()


if __name__ == '__main__':
    main()

Angenommen, ich will die 'bits' nicht in dieser Schleife behandeln, sondern die sollen vielmehr in einer asynchronen Funktion kontinuierlich (z.B. durch ständig erneutes Durchlaufen des Arrays) erzeugt werden und dann parallel dazu in einer anderen Funktion verarbeitet werden (nach Prinzip Producer, Consumer).
Da ich bisher nichts mit nebenläufigen Funktionen zu tun hatte, versuche ich mich da gerade einzulesen (was wäre hier die beste Quelle?), bekomme das aber noch nicht zusammen gereimit.

Danke und Gruß

[sparrow]

Es sind ein paar Formulierungen in deinem Text, die nicht klar sind.

Du sprichst von einem "Array" - ich nehme an, du meinst die Liste (Arrays sind in Python in der Regel etwas Anderes).
Dann schreibst du, dass die "bits" durch das durchlaufen des Arrays (wahrscheinlich Liste) neu erzeugt werden sollen - aber gleichzeitig sollen sie an anderer Stellte verarbeitet werden‽ Das klingt in dieser Formulierung nicht sinnvoll.

Man verändert keine Objekte, die in verschiedenen Threads benötigt werden.
Um eine Ahnung davon zu erhalten, empfehle ich heute gerne mit der Programmiersprache Rust zu beschäftigen.

Ich rate mal, was du eigentlich machen möchtest:
Du schreibst Daten in eine Liste, und dann packst du diese Liste eine Queue. Eine Arbeitsthread holt sich die Liste dort heraus und verabeitet sie.
Dein "Producer" erstellt für die nächste Runde eine neue Liste und auch der Consumer erstellt für das Ergebnis eine neue Liste und ändert die vorhandene nicht.

[mechanicalStore]

Ja, ich habe Array gesagt und meine Liste. Ich weiß, was Listen sind und da Arrays einen ähnlichen Sinn haben, habe ich fälschlicherweise Array gesagt. Ich benutze diese Begriffe synonym. Die "bits" sind extra in Anführungszeichen gewesen. Natürlich sind das keine echten Bits. Und die Liste soll auch nicht verändert werden, das habe ich so nicht geschrieben. Ich will sie nur fortlaufend erneut von Vorne durchlaufen, ohne sie zu ändern. Was ich eigentlich möchte, war auch nicht eine Queue, in die ich immer neue Daten rein schiebe und abhole.

Vielleicht hätte ich sagen sollen, dass ich einen Datenfluss quasi simulieren will, als kämen Bits von Aussen (z.B. über eine serielle Schnittstelle) Bitweise an, auf die ich dann in unterschiedlicher Weise reagieren will. Um das zu simulieren, stehen in der Liste halt willkürliche Werte, die fortwährend "gesendet" werden.

[Sirius3]

So ganz klar wird mir nicht, was Du eigentlich machen möchtest, und was das dann mit asynchroner Verarbeitung zu tun hat. Das ist nämlich nur sinnvoll, wenn man viele Quellen gleichzeitig verarbeiten will.
Du hast ja nur einen Strom an Daten, und da ist das einfach ein Generator.

[grubenfox]

Und die Liste soll auch nicht verändert werden, das habe ich so nicht geschrieben. Ich will sie nur fortlaufend erneut von Vorne durchlaufen, ohne sie zu ändern.

Das wäre dann wohl cycle

[DeaD_EyE]

So?

Code: Alles auswählen

import asyncio
from itertools import cycle


class AsyncCycle:
    def __init__(self, iterable, delay):
        self.cycle = cycle(iterable)
        self.delay = delay

    async def __aiter__(self):
        for element in self.cycle:
            await asyncio.sleep(self.delay)
            yield element


async def worker1(async_iterable):
    async for element in async_iterable:
        print(element)


async def worker2(async_iterable):
    async for element in async_iterable:
        print(element)


async def main():
    bits = [
        1,
        1,
        0,
        0,
        0,
        1,
        1,
        0,
        1,
        0,
        1,
        1,
        1,
        0,
        0,
        0,
        0,
        0,
        1,
        0,
        0,
        1,
        1,
        0,
        0,
        1,
    ]
    async_iterator1 = AsyncCycle(bits, 1.0)
    async_iterator2 = AsyncCycle(bits, 1.5)

    task1 = asyncio.create_task(worker1(async_iterator1))
    task2 = asyncio.create_task(worker2(async_iterator2))

    await asyncio.gather(task1, task2)


if __name__ == "__main__":
    asyncio.run(main())

[noisefloor]

Hallo,

mir ist auch nicht klar, worauf du hinaus willst... vielleicht solltest du das Problem nicht versuchen, zu abstrahieren, weil dich das eher weiter von einer Lösung weg führt...

Python kennt in der Standardbibliothek drei Module für Nebenläufigkeit: threading, multiprocessing und asyncio. Plus concurrent.futures - was "nur" eine zusätzlich Schicht über threading und multiprocessing legt, die aber kann praktisch sein kann. Dann gibt es noch diverse externe Bibliotheken, die darauf aufsetzen oder eigene Sachen implementieren oder asyncrone Task Queue implementieren (wie z.B. Celery).

Was für deinen Fall am besten ist: keine Ahnung, dazu fehlen deinerseits Infos. asyncio nimmt man in der Regel dann, wenn man eine größerer Zahl externer Datenquellen hat, von den man nicht weiß, wann Daten ankommen. Wie z.B. bei einem Chatserver.

Echte Parallelität bekommst du z.Zt. mit (C)Python nur mit Multiprocessing und mehreren CPU-Kernen. Noch ist in CPython der GIL aktiv, d.h. aus Python heraus erzeugte Threads kann CPython nicht parallel. Zumindest nicht, bis free-threaded CPython der Standard ist.

Ich benutze diese Begriffe synonym.

Solltest du in Python tunlichst vermeiden, weil array ein eigenes Modul in Python ist und array und list eben weder das gleiche noch das selber sind.

Gruß, noisefloor

[mechanicalStore]

Hallo,

mir ist auch nicht klar, worauf du hinaus willst... vielleicht solltest du das Problem nicht versuchen, zu abstrahieren, weil dich das eher weiter von einer Lösung weg führt...

Das Problem ist, dass ich keinen konkreten Anwenungsfall habe und daher in dem Sinne auch nichts abstrahiere. D.h. es handelt sich hier nicht um was Konkretes, um keinerlei Projekt, sondern ich will einfach nur just for fun wissen, wie man sowas angeht, da ich mit Nebenläufigkeit bisher nichts zu tun hatte.

Einfach gesagt, könnte ich jetzt den Lötkolben schwingen und einen Signalgenerator zusammen löten und damit dann irgendwas an die serielle Schnittstelle liefern. Da mir das Löten aber zuviel Aufwand ist, wollte ich das möglichst realitätsnah simulieren. Und gelesen habe ich bisher, dass asyncio für sowas gut ist, damit das Programm nicht 'warten' muss, sondern derweil weiter läuft, bis die Schnittstelle halt irgendwas liefert.
Was Dead_Eye gezeigt hat, kommt dem sehr nahe. Nur wäre das in meinem Fall so, dass nicht die worker Funktion selbst was printet, sondern die Werte entsprechend an die aufrufende Funktion zurück liefert.

Sorry, wenn das alles Verwirrung schafft, ich weiß halt nicht, wie ich es sonst beschreiben soll, da es halt um nichts Konkretes geht.

Danke für die Antworten und Gruß

[noisefloor]

Hallo,

ok. Dein konstruiertes Beispiel macht das aber eben _nicht_, weil die "Daten" in Form der Liste permanent zur Verfügung stehen - da muss auf nix gewartet werden. Grundproblem beim Lernen von asyncio: wenn man keine konkrete Anwendung hat, ist das echt schwierig. Viele Beispiele im Netz arbeiten mit `asyncio.sleep`, um das "Warten" zu simulieren. Was IMHO den Transfer auf die Realität nicht einfacher macht.

Wenn du mit Daten einer seriellen Schnittstelle arbeitest, dann benutzt du doch i.d.R. ein externes Python-Modul zur Abfrage der selbigen. Wenn das Modul asyncio unterstützt, dann sollte in der Doku auch was dazu zu finden sein. Und asynchrones Warten macht nur Sinn, wenn man entweder mindesten zwei Schnittstellen hat oder auf andere Daten, die irgendwann kommen, wartet. Das hat nicht direkt was damit zu tun, was du mit den Daten machst. Wenn jetzt Daten kommen und die Verarbeitung dauert ein paar Sekunden, dann willst du die Daten über eine Queue an einen Worker-Prozess schicken, damit die Verarbeitung deinen Event-Loop nicht blockiert.

Vereinfachtes Beispiel:
Du steht in einem Raum mit 4 Türen. Die gehst pausenlos die vier Türen der Reihe nach ab und schaust, ob wer dahinter steht. Wenn ja, lässt du ihn rein und gehst direkt weiter. Wenn nein gehst du auch direkt weiter. Wenn du eine Tür öffnest und dich mit der Person in der Tür erstmal 5 Minuten unterhältst, bevor sie rein kommt, dann kannst du in den fünf Minuten nicht die anderen drei Türen öffnen. Wodurch sich dahinter ein (massiver) Rückstau bilden könnte.

Zu asyncio findet man auch reichlich Lesestoff im Netz. Bisschen älter, aber zum Einlesen IMHO immer noch ok: https://pymotw.com/3/asyncio/. Wo IMHO das async-Zeugs auch gut erklärt wird: https://trio.readthedocs.io/en/stable/. Trio ist ein Python-Modul mit einer alternativen Implementierung von asyncio aus der Standardbibliothek.

Gruß, noisefloor

[DeaD_EyE]

In meinem Beispiel wäre an der Stelle, wo await asyncio.sleep() aufgerufen wird, ein Ereignis.
Das kann z.B. ein Netzwerkpaket sein, auf das reagiert wird. Wie der Name des Moduls hindeutet, geht es hier hauptsächlich um IO (Daten, die ankommen und gesendet werden sollen).

Die Nebenläufigkeit ist im Endeffekt synchron und die Tasks werden durch die Eventloop ausgeführt, wenn für die entsprechenden Tasks Daten vorhanden sind. Meist programmiert man __aiter__ und die anderen Methoden nicht selbst, sondern nutzt meistens Bibliotheken, die das abstrahieren.

Hier ein Beispiel mit websockets, wo async for genutzt werden kann: https://websockets.readthedocs.io/en/st ... connection

[__blackjack__]

`__aiter__()` implementiert man doch eigentlich genau so häufig wie man `__iter__()` tatsächlich implementiert. Also eher selten und im Beispiel trifft „eine Klasse mit einer `__init__()` und nur einer zusätzlichen Methode ist oft eine als Klasse verkleidete Funktion“ zu:

Code: Alles auswählen

# Die Klasse…

class AsyncCycle:
    def __init__(self, iterable, delay):
        self.cycle = cycle(iterable)
        self.delay = delay

    async def __aiter__(self):
        for element in self.cycle:
            await asyncio.sleep(self.delay)
            yield element

# …lässt sich auch als Funktion schreiben.

async def async_cycle(iterable, delay):
    for element in cycle(iterable):
        await asyncio.sleep(delay)
        yield element