Python asyncio Kraft Timeout

Question

asyncio einen Koroutine verwenden, kann mit einem Timeout ausgeführt werden, so dass es nach dem Timeout abgebrochen wird:

@asyncio.coroutine 
def coro(): 
    yield from asyncio.sleep(10) 

loop = asyncio.get_event_loop() 
loop.run_until_complete(asyncio.wait_for(coro(), 5)) 

Das obige Beispiel wie erwartet funktioniert (es Timeout nach 5 Sekunden).

Wenn jedoch die Coroutine asyncio.sleep() (oder andere asyncio-Koroutinen) nicht verwendet, scheint es nicht zu Timeout. Beispiel:

@asyncio.coroutine 
def coro(): 
    import time 
    time.sleep(10) 

loop = asyncio.get_event_loop() 
loop.run_until_complete(asyncio.wait_for(coro(), 1)) 

Diese mehr als 10 Sekunden dauert, weil die time.sleep(10) laufen wird nicht abgebrochen. Ist es in einem solchen Fall möglich, die Löschung der Coroutine zu erzwingen?

Wenn asyncio verwendet werden soll, um dies zu lösen, wie könnte ich das tun?

Answer 1

Nein, Sie können eine Coroutine nicht unterbrechen, es sei denn, sie gibt die Kontrolle an die Ereignisschleife zurück, was bedeutet, dass sie innerhalb eines yield from Aufrufs sein muss. asyncio ist single-threaded. Wenn Sie also in Ihrem zweiten Beispiel den Aufruf time.sleep(10) blockieren, kann die Ereignisschleife nicht ausgeführt werden. Das bedeutet, wenn die Zeitüberschreitung, die Sie mit wait_for festgelegt haben, abläuft, kann die Ereignisschleife keine Maßnahmen ergreifen. Die Ereignisschleife erhält keine Möglichkeit erneut ausgeführt zu werden, bis coro beendet wird, zu diesem Zeitpunkt ist es zu spät.

Aus diesem Grund sollten Sie im Allgemeinen blockierende Aufrufe vermeiden, die nicht asynchron sind. Jedes Mal, wenn ein Anruf blockiert wird, ohne der Ereignisschleife zu folgen, kann nichts anderes in Ihrem Programm ausgeführt werden, was wahrscheinlich nicht das ist, was Sie wollen. Wenn Sie wirklich eine lange, Sperrung Betrieb tun müssen, sollten Sie versuchen, BaseEventLoop.run_in_executor zu verwenden, um es in einem Thread oder Prozess Pool ausgeführt werden, die vermeiden wird die Ereignisschleife blockiert:

import asyncio 
import time 
from concurrent.futures import ProcessPoolExecutor 

@asyncio.coroutine 
def coro(loop): 
    ex = ProcessPoolExecutor(2) 
    yield from loop.run_in_executor(ex, time.sleep, 10) # This can be interrupted. 

loop = asyncio.get_event_loop() 
loop.run_until_complete(asyncio.wait_for(coro(loop), 1)) 

Answer 2

Thx @dano für Ihre Antwort. Wenn Ausführen eines coroutine keine harte Anforderung ist, ist hier eine überarbeitete, kompaktere Version

import asyncio, time, concurrent 

timeout = 0.5 
loop = asyncio.get_event_loop() 
future = asyncio.wait_for(loop.run_in_executor(None, time.sleep, 2), timeout) 
try: 
    loop.run_until_complete(future) 
    print('Thx for letting me sleep') 
except concurrent.futures.TimeoutError: 
    print('I need more sleep !') 

Für die Neugierigen, ein wenig Debugging in meinem Python 3.5.2 zeigte, dass None als ein Vollstrecker führt zur Schaffung eines _default_executor vorbei, wie folgt:

Answer 3

Die Beispiele, die ich für die Timeout-Behandlung gesehen habe, sind sehr trivial. Angesichts der Realität ist meine App etwas komplexer. Die Reihenfolge ist:

1. Wenn ein Client zum Server verbindet, muss der Server eine andere Verbindung zum internen Server senden Daten
2. Wenn der interne Server-Verbindung in Ordnung ist, warten Sie auf den Client erstellen. Basierend auf diesen Daten können wir eine Anfrage an den internen Server stellen.
3. Wenn Daten an den internen Server gesendet werden sollen, senden Sie sie. Da der interne Server manchmal nicht schnell genug antwortet, wickeln Sie diese Anfrage in ein Timeout.
4. Wenn die Betriebszeiten aus, alle Verbindungen kollabieren den Kunden über Fehler

Um alle oben zu erreichen, um zu signalisieren, während die Ereignisschleife am Laufen zu halten, enthält der resultierende Code folgenden Code:

def connection_made(self, transport): 
    self.client_lock_coro = self.client_lock.acquire() 
    asyncio.ensure_future(self.client_lock_coro).add_done_callback(self._got_client_lock) 

def _got_client_lock(self, task): 
    task.result() # True at this point, but call there will trigger any exceptions 
    coro = self.loop.create_connection(lambda: ClientProtocol(self), 
              self.connect_info[0], self.connect_info[1]) 
    asyncio.ensure_future(asyncio.wait_for(coro, 
              self.client_connect_timeout 
              )).add_done_callback(self.connected_server) 

def connected_server(self, task): 
    transport, client_object = task.result() 
    self.client_transport = transport 
    self.client_lock.release() 

def data_received(self, data_in): 
    asyncio.ensure_future(self.send_to_real_server(message, self.client_send_timeout)) 

def send_to_real_server(self, message, timeout=5.0): 
    yield from self.client_lock.acquire() 
    asyncio.ensure_future(asyncio.wait_for(self._send_to_real_server(message), 
                timeout, loop=self.loop) 
           ).add_done_callback(self.sent_to_real_server) 

@asyncio.coroutine 
def _send_to_real_server(self, message): 
    self.client_transport.write(message) 

def sent_to_real_server(self, task): 
    task.result() 
    self.client_lock.release()