【Python】asyncio（非同期I/O）のイベントループをこねくり回す

#目的
Python公式ドキュメントを参考に[asyncioライブラリ](https://docs.python.org/3.7/library/asyncio-eventloop.html)のイベントループを理解する。

#環境
python 3.7.4

#注意
`import asyncio`を忘れずに。本稿のサンプルコードでは省略しています。

#イベントループメソッド
##イベントループの取得
#####`asyncio.get_event_loop()`

カレントイベントループを取得。カレントスレッドにカレントイベントループがなければ自動的にイベントループを作り，それをカレントイベントループに設定する。一度設定し，再度作っても同じカレントイベントループになる。

```python
loop = asyncio.get_event_loop()
loop_2 = asyncio.get_event_loop()
print(loop is loop_2) # True
```

#####`asyncio.new_event_loop()`
新しいイベントループを作る。ただしカレントイベントループではない。

```python
new_loop = asyncio.new_event_loop()
```

#####`asyncio.set_event_loop(loop)`
引数*loop*をカレントイベントループに設定する。

```python
new_loop = asyncio.new_event_loop()
loop = asyncio.set_event_loop(new_loop)
```

#####`asyncio.get_running_loop()`
カレントスレッドで実行中のイベントループを返す。もしイベントループがなければ例外（RuntimeError）を発生させる。この関数はコルーチンもしくはコールバックからでのみ使用可能。

```python
async def main():
    print(asyncio.get_running_loop()) # <_UnixSelectorEventLoop running=True closed=False debug=False>
    print('Hello') # Hello
    await asyncio.sleep(1)
    print('World') # World

asyncio.run(main())
```

##ループの実行と停止

#####`loop.run_until_complete(future)`
引数*future*（Futureのインスタンス）が完了するまで実行する。引数がコルーチンの場合は，`asyncio.Task`として実行するまで予約される。Futureの戻り値か例外を返す。

```python
async def main():
    print('Hello') # Hello
    await asyncio.sleep(1)
    print('World') # World

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()
```

#####`loop.run_forever()`
`loop.stop()`が呼ばれるまでループし続ける。

```python
def main(loop):
    print('Hello World')
    loop.stop()

loop = asyncio.get_event_loop()
loop.call_soon(main, loop)

try:
    loop.run_forever()
finally:
    loop.close()
```

#####`loop.stop()`
イベントループを停止する。

#####`loop.is_running()`
イベントループが実行中なら`True`を返す。

#####`loop.is_closed()`
イベントループが閉じられたら`True`を返す。

#####`loop.close()`
イベントループを閉じる。

#####（コルーチン関数）`loop.shutdown_asyncgens（）`

非同期ジェネレーターオブジェクトを閉じる。

```python
async def my_async_generator(iterable):
    for num in iterable:
        await asyncio.sleep(1)
        yield num

async def main():
    generator = my_async_generator([1, 2, 3, 4, 5])
    async for item in generator:
        print(item)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.call_soon(loop.shutdown_asyncgens)
    loop.close()
```

##コールバックのスケジューリング

#####`loop.call_soon(callback, *args, context=None)`
引数*callback*をイベントキューに入れる。引数**args*は*callback*の引数。 *callback*は入れた順に呼び出される（FIFO）。スレッドセーフではないのでマルチスレッドでの使用する際は注意が必要。

```python
loop = asyncio.get_event_loop()
loop.call_soon(print, 'Hello') # Hello
loop.call_soon(print, 'World') # World
loop.call_soon(loop.stop)
loop.run_forever()
loop.close()
```

#####`loop.call_soon_threadsafe(callback, *args, context=None)`
マルチスレッドでスレッドセーフに*callback*を呼び出す時に使用。

```python
import threading
import asyncio

def say_hello():
    print('Hello')

def say_world():
    print('World')

loop_thread1 = asyncio.new_event_loop()
loop_thread2 = asyncio.new_event_loop()

def run(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

loop_thread1.call_soon_threadsafe(say_hello)
loop_thread2.call_soon_threadsafe(say_world)

loop_thread1.call_soon_threadsafe(loop_thread1.stop)
loop_thread2.call_soon_threadsafe(loop_thread2.stop)

thread1 = threading.Thread(target=run, args=(loop_thread1,))
thread2 = threading.Thread(target=run, args=(loop_thread2,))
thread1.start()
thread2.start()

# Hello
# World
```

##遅延コールバックのスケジューリング
#####`loop.call_later(delay, callback, *args, context=None)`
引数*delay*に遅らせたい時間（秒）を設定する。そうすることで*delay*秒後に*callback*が呼ばれる。この関数を用いる事で自前の`asyncio.sleep()`を作ることができる。

```python
def set_result_unless_cancelled(future, result):
    if future.cancelled():
        return result
    future.set_result(result)

# 自前のasyncio.sleep()
async def my_asyncio_sleep(delay, result=None):
    if delay <= 0:
        return result

    loop = asyncio.get_event_loop()
    future = loop.create_future()
    h = loop.call_later(delay,
                    set_result_unless_cancelled,
                    future,
                    result)
    try:
        return await future
    finally:
        h.cancel()
```

#####`loop.call_at（when、callback、* args、context = None)`
`loop.time()`からの経過時間後に*callback*を返す。`call_later()`は内部で`call_at()`を呼び出しているので実質的に`call_later()`と同じ。

```python
async def my_asyncio_sleep(delay, result=None):
    if delay <= 0:
        return result

    loop = asyncio.get_event_loop()
    future = loop.create_future()
    # h = loop.call_later(delay, set_result_unless_cancelled, future, result)
    h = loop.call_at(loop.time() + delay, set_result_unless_cancelled, future, result)
    try:
        return await future
    finally:
        h.cancel()
```

#####`loop.time()`
モノトニック時刻を返す。

##FutureとTaskオブジェクトの作製

#####`loop.create_future()`
Futureオブジェクトを作製する。

```python
loop = asyncio.get_event_loop()
loop.create_future()
```

#####`loop.create_task(coro)`
引数*coro*はコルーチンを指す。コルーチンを実行予約のリストに格納する。Taskオブジェクトを返す。

```python
import asyncio
import time

# firstからend-1までの和を求めて
# asyncio.sleep()で1秒待機
async def async_do_sum(first, end):
    result = sum(range(first, end))
    print(result)
    await asyncio.sleep(1)

async def async_main():
    t1 = time.time()
    
    # Taskオブジェクトの作製
    task1 = asyncio.create_task(async_do_sum(1, 10 ** 2 + 1))
    task2 = asyncio.create_task(async_do_sum(1, 10 ** 4 + 1))
    task3 = asyncio.create_task(async_do_sum(1, 10 ** 6 + 1))

    await asyncio.wait({task1, task2, task3})

    t2 = time.time()

    print('計算に{:.2f}秒かかりました'.format((t2 - t1)))

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(async_main())

# 5050
# 50005000
# 500000500000
# 計算に1.04秒かかりました
```
#####`loop.set_task_factory(factory)`
タスクファクトリーの作製。引数*factory*の引数は`(loop, coro)`でなければいけない。詳しくは[こちら](https://blog.sqreen.com/asyncio/)。

```python
import asyncio

def set_value(value):
    task = asyncio.current_task()
    setattr(task, 'value', value)

def get_value():
    task = asyncio.current_task()
    return getattr(task, 'value', None)

async def do_sum():
    value = get_value()
    sum_value = sum(range(1, value + 1))
    print(sum_value)
    await asyncio.sleep(1)

async def do_fibonacci():
    value = get_value()
    print([i for i in fibonacci(value)])
    await asyncio.sleep(1)

def fibonacci(n):
    a, b = 0, 1

    for _ in range(n):
        yield a
        a, b = b, a + b

def task_factory(loop, coro):
    child_task = asyncio.Task(coro, loop=loop)
    parent_task = asyncio.current_task(loop=loop)
    current_task = getattr(parent_task, 'value', None)
    setattr(child_task, 'value', current_task)

    return child_task

async def main(value):
    set_value(value)

    await asyncio.gather(do_sum(),
                         do_fibonacci(),
                         )

loop = asyncio.get_event_loop()
loop.set_task_factory(task_factory)
loop.run_until_complete(main(value=10))
loop.close()

# 55
# [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
```

#####`loop.get_task_factory()`
現在のタスクファクトリーを返す。

---
ここで一旦アップします。時間を見つけて更新します。

#参考文献
[Python公式リファレンス（イベントループ）](https://docs.python.org/3.7/library/asyncio-eventloop.html)
[Context information storage for asyncio](https://blog.sqreen.com/asyncio/)