オブジェクトが状態を変更するのを待つ方法

Question

私のasyncハンドラの中で、タスクの状態が変更されるまで待つ必要があります。今のところ、私は無限ループで状態をチェックして待ちます。ここでwait_until_done機能、例です。

import asyncio 


class LongTask: 
    state = 'PENDING' 

my_task = LongTask() 


def done(): 
    my_task.state = 'DONE' 

async def wait_until_done(): 
    while True: 
     if my_task.state == 'PENDING': 
      await asyncio.sleep(2) 
     else: 
      break 
    print("Finally, the task is done") 


def main(loop, *args, **kwargs): 
    asyncio.ensure_future(wait_until_done()) 
    loop.call_later(delay=5, callback=done) 

loop = asyncio.get_event_loop() 
main(loop) 
loop.run_forever() 

はそれを行うためのより良い方法はありますか？

Answer 1

ちょうど混乱を避けるために：私はあなたがasyncio.Taskについて話していないと思いますが、代わりにいくつかの可変状態、右か？

Futureとsynchronization primitivesを使用すると、非同期的に変更されたものを待つことができます。

2つの状態を切り替える必要がある場合は、おそらくasyncio.Eventが必要です。ここで少しexamleです：

import asyncio 


my_task = asyncio.Event() 


def done(): 
    my_task.set() 



async def wait_until_done(): 
    await my_task.wait() # await until event would be .set() 
    print("Finally, the task is done") 


async def main(): 
    loop.call_later(delay=5, callback=done) 
    await wait_until_done() 


loop = asyncio.get_event_loop() 
try: 
    loop.run_until_complete(main()) 
finally: 
    loop.run_until_complete(loop.shutdown_asyncgens()) 
    loop.close() 

UPD：LongTaskインタフェース続け

より複雑な例：

import asyncio 



class LongTask: 
    _event = asyncio.Event() 

    @property 
    def state(self): 
     return 'PENDING' if not type(self)._event.is_set() else 'DONE' 

    @state.setter 
    def state(self, val): 
     if val == 'PENDING': 
      type(self)._event.clear() 
     elif val == 'DONE': 
      type(self)._event.set() 
     else: 
      raise ValueError('Bad state value.') 

    async def is_done(self): 
     return (await type(self)._event.wait()) 

my_task = LongTask() 


def done(): 
    my_task.state = 'DONE' 



async def wait_until_done(): 
    await my_task.is_done() 
    print("Finally, the task is done") 


async def main(): 
    loop.call_later(delay=5, callback=done) 
    await wait_until_done() 


loop = asyncio.get_event_loop() 
try: 
    loop.run_until_complete(main()) 
finally: 
    loop.run_until_complete(loop.shutdown_asyncgens()) 
    loop.close() 

Answer 2

はそれがObserverデザインパターンの例はありませんか？私はObserverパターンがこの問題を解決できると思います。この設計パターンでは、変更を検出するために無限ループを繰り返す必要はなく、変更をコミットしたかどうかをタスク自体が通知します。単純な実装は次のようになります。

class ObservableTask: 
    def __init__(self): 
     self.subscribers = set() 
     self.state = 'PENDING' 

    def subscribe(self, who): 
     self.subscribers.add(who) 

    def unsubscribe(self, who): 
     self.subscribers.discard(who) 

    def dispatch(self, message): 
     for subscriber in self.subscribers: 
      subscriber.update(message) 

    def random_change(self): 
     for count in range(1, 10): 
      if count % 5 == 0: 
       print('Inside task:\tDivisible by 5') 
       self.state = 'DONE' 
       self.dispatch('state: DONE') 


class Observer: 
    def __init__(self): 
     pass 

    def update(self, message): 
     print('Task is changed!\t' + message) 


# Test: 
task = ObservableTask() 
observer = Observer() 

task.subscribe(observer) 
task.random_change() 

出力：

Inside task: Divisible by 5 
Task is changed! state: DONE