asyncio concurrent multiprocess study

mp.py

# -*- coding: utf-8 -*-
import concurrent.futures
import asyncio
import time
import sys
import signal
import functools
import logging
logging.basicConfig(format='%(asctime)s [%(process)d] %(name)s %(levelname)s: %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)


worker_will_shutting_down = False


def run(i):
    global worker_will_shutting_down
    if worker_will_shutting_down:
        logger.info('not execute %d', i)
        return
    try:
        logger.info('run: %d', i)
        time.sleep(i)
        logger.info('%s -> %s', i, i**2)
        return i**2
    except KeyboardInterrupt:
        logger.warning('KeyboardInterrupt Exception!')
        worker_will_shutting_down = True
        return


def main():
    loop = asyncio.get_event_loop()
    executor = concurrent.futures.ProcessPoolExecutor()
    loop.set_default_executor(executor)
    fs = []

    for i in range(20):
        future = loop.run_in_executor(None, run, i)
        fs.append(future)
        time.sleep(0.5)

    try:
        done, not_done = loop.run_until_complete(asyncio.wait(fs))
        results = [f.result() for f in done]
        print(results)
    except KeyboardInterrupt:
        logger.info('Interrupted')
        # import pdb;pdb.set_trace()
        # logger.info('1. cancelled? %s', [f.cancelled() for f in fs])
        # logger.info('2. cancell %s', [f.cancel() for f in fs])
        # logger.info('3. cancelled? %s', [f.cancelled() for f in fs])
        for f in fs:
            f.cancel()
        #     # logger.info('Future cancel: %r, %s', f, f.cancel())
        # for task in asyncio.Task.all_tasks():
        #     # TODO: ここでcancelしてるのに次のタスクも実行されちゃう
        #     task.cancel()
        #     logger.info('Task cancel: %r, %s', task, task.cancel())
        logger.info('waiting shutdown...')
        executor.shutdown(True)
        logger.info('shutdown')
    finally:
        logger.info('loop closing...')
        loop.run_until_complete(loop.shutdown_asyncgens())
        loop.close()

if __name__ == '__main__':
    main()