programming - Crear una devolución de llamada de temporizador asíncrono temporal a un método vinculado con python-asyncio

python async api

Question

asyncio crear una especie de devolución de llamada de temporizador a un método de asyncio async utilizando el asyncio de eventos de asyncio . El problema ahora es que el método de sincronización asíncrono no debe contener una referencia fuerte a la instancia, de lo contrario, este último nunca será eliminado. La devolución de llamada del temporizador solo debe vivir tanto tiempo como la instancia principal. Encontré una solución, pero no creo que sea bonita:

import asyncio
import functools
import weakref
class ClassWithTimer:
def __init__(self):
asyncio.ensure_future(
functools.partial(
ClassWithTimer.update, weakref.ref(self)
)()
)
def __del__(self):
print("deleted ClassWithTimer!")
async def update(self):
while True:
await asyncio.sleep(1)
if self() is None: break
print("IN update of object " + repr(self()))
async def run():
foo = ClassWithTimer()
await asyncio.sleep(5)
del foo
loop = asyncio.get_event_loop()
loop.run_until_complete(run())

¿Hay una manera mejor y más pitonica de hacer esto? La devolución de llamada del temporizador realmente necesita ser asincrónica. Sin asyncio , weakref.WeakMethod probablemente sea el camino a seguir. Pero asyncio.ensure_future requiere un objeto coroutine, por lo que no funcionará en este caso.

Answer 1

1. Python 3.5 maneja muy bien las referencias de ciclo (ver PEP 442 https://www.python.org/dev/peps/pep-0442/ )

2. Empujaré el administrador de contexto asincrónico de tiempo de espera a Asyncio pronto, vea el borrador:

`

import asyncio
import warnings
class Timeout:
def __init__(self, timeout, *, raise_error=False, loop=None):
self._timeout = timeout
if loop is None:
loop = asyncio.get_event_loop()
self._loop = loop
self._raise_error = raise_error
self._task = None
self._cancelled = False
self._cancel_handler = None
async def __aenter__(self):
self._task = asyncio.Task.current_task(loop=loop)
self._cancel_handler = self._loop.call_later(
self._cancel, self._timeout)
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self._cancelled:
if self._raise_error:
raise asyncio.TimeoutError
else:
# suppress
return True
else:
self._cancel_handler.cancel()
# raise all other errors
def __del__(self):
if self._task:
# just for preventing improper usage
warnings.warn("Use async with")
def _cancel(self):
self._cancelled = self._task.cancel()
async def long_running_task():
while True:
asyncio.sleep(5)
async def run():
async with Timeout(1):
await long_running_task()
loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Answer 2

Lo siento, no estoy seguro si entendí tu pregunta correctamente. ¿Esta es la solución que estás buscando?

import asyncio
class ClassWithTimer:
async def __aenter__(self):
self.task = asyncio.ensure_future(self.update())
async def __aexit__(self, *args):
try:
self.task.cancel() # Just cancel updating when we don''t need it.
await self.task
except asyncio.CancelledError: # Ignore CancelledError rised by cancelled task.
pass
del self # I think you don''t need this in real life: instance should be normally deleted by GC.
def __del__(self):
print("deleted ClassWithTimer!")
async def update(self):
while True:
await asyncio.sleep(1)
print("IN update of object " + repr(self))
async def run():
async with ClassWithTimer(): # Use context manager to handle when we need updating.
await asyncio.sleep(5)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Salida:

IN update of object <__main__.ClassWithTimer object at 0x000000468D0FB208>
IN update of object <__main__.ClassWithTimer object at 0x000000468D0FB208>
IN update of object <__main__.ClassWithTimer object at 0x000000468D0FB208>
IN update of object <__main__.ClassWithTimer object at 0x000000468D0FB208>
deleted ClassWithTimer!
[Finished in 5.2s]

Una forma más sin el administrador de contexto:

import asyncio
class ClassWithTimer:
def __init__(self):
self.task = asyncio.ensure_future(self.update())
async def release(self):
try:
self.task.cancel()
await self.task
except asyncio.CancelledError:
pass
del self
def __del__(self):
print("deleted ClassWithTimer!")
async def update(self):
while True:
await asyncio.sleep(1)
print("IN update of object " + repr(self))
async def run():
foo = ClassWithTimer()
await asyncio.sleep(5)
await foo.release()
loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Answer 3

Me encontré con casi el mismo problema hace unos meses, y escribí un decorador para evitarlo:

def weakmethod(f):
@property
def get(self):
return f.__get__(weakref.proxy(self))
# self = weakref.proxy(self)
# if hasattr(f, ''__get__''):
# raise RuntimeWarning(
# ''weakref may not work unless you implement ''
# ''the property protocol carefully by youself!''
# )
# return f.__get__(self)
# if asyncio.iscoroutinefunction(f):
# #Make the returned method a coroutine function, optional
# async def g(*arg, **kwarg):
# return await f(self, *arg, **kwarg)
# else:
# def g(*arg, **kwarg):
# return f(self, *arg, **kwarg)
# return g
# #Still some situations not taken into account?
return get

Su código podría ser reescrito de una manera muy natural:

class ClassWithTimer:
def __init__(self):
asyncio.ensure_future(self.update())
def __del__(self):
print("deleted ClassWithTimer!")
@weakmethod
async def update(self):
while True:
await asyncio.sleep(1)
print("IN update of object ", self)

Importante:

• weakref.proxy no evita que adquieras una referencia fuerte. Además, no se garantiza que se comporte exactamente igual que el objeto original.

• Mi implementación no ha cubierto todas las posibilidades.

Answer 4

En base a las respuestas de Huazuo Gao y germn, implementé ensure_weakly_binding_future que es básicamente lo mismo que ensure_future pero no guarda una referencia fuerte a la instancia del método bound. No modifica el enlace general (como la solución basada en el decorador) y cancela correctamente el futuro cuando se elimina la instancia principal:

import asyncio
import weakref
def ensure_weakly_binding_future(method):
class Canceller:
def __call__(self, proxy):
self.future.cancel()
canceller = Canceller()
proxy_object = weakref.proxy(method.__self__, canceller)
weakly_bound_method = method.__func__.__get__(proxy_object)
future = asyncio.ensure_future(weakly_bound_method())
canceller.future = future
class ClassWithTimer:
def __init__(self):
ensure_weakly_binding_future(self.update)
def __del__(self):
print("deleted ClassWithTimer!", flush=True)
async def update(self):
while True:
await asyncio.sleep(1)
print("IN update of object " + repr(self), flush=True)
async def run():
foo = ClassWithTimer()
await asyncio.sleep(5.5)
del foo
await asyncio.sleep(2.5)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())