asyncio_multiloop_demo.py

# vim:fileencoding=utf-8
import asyncio
import uvloop
from concurrent.futures import ProcessPoolExecutor
from concurrent.futures import ThreadPoolExecutor
import logging
log = logging.getLogger(__name__)
log.setLevel(logging.DEBUG)




def process_bound_job(t, n):
    print('proc {}: >>'.format(n))
    policy = asyncio.get_event_loop_policy()
    print('proc {}: {} {}'.format(n, id(policy), policy))
    loop = policy.new_event_loop()
    # id() doesn't necessarily give different values when using ProcessExecutor
    # when loops are different objects residing in different processes
    print('proc {}: {} {}'.format(n, id(loop), loop))
    policy.set_event_loop(loop)
    loop.run_until_complete(top_one(t, n))
    print('proc {}: done'.format(n))
    return n

async def top_level(loop):
    t1 = loop.run_in_executor(None, process_bound_job, 5, 1)
    t2 = loop.run_in_executor(None, process_bound_job, 10, 2)
    #await asyncio.wait([t1, t2])
    t3 = loop.run_in_executor(None, process_bound_job, 15, 3)
    await asyncio.wait([t1, t2, t3])

async def top_one(t, n):
    print('{}: top_one >>'.format(n))
    loop = asyncio.get_event_loop()
    print('{}: {} {}'.format(n, id(loop), loop))
    await asyncio.sleep(t)
    print('{}: top_one done'.format(n))

def run():

    # optional: uvloop as a drop-in replacement
    policy = uvloop.EventLoopPolicy()
    print('proc {}: {} {}'.format(0, id(policy), policy))
    asyncio.set_event_loop_policy(policy=policy)

    loop = asyncio.get_event_loop()

    # total: n + 1 processes/threads
    executor = ProcessPoolExecutor(3)
    #executor = ThreadPoolExecutor(3)
    loop.set_default_executor(executor)

    loop.create_task(top_one(3, 0))
    loop.run_until_complete(top_level(loop))
    executor.shutdown()


if __name__ == '__main__':
    run()