00use_thread_pool_executor.py

import logging
import asyncio
import time
from functools import partial
from mysleep import mysleep

logger = logging.getLogger(__name__)


async def run():
    loop = asyncio.get_event_loop()
    st = time.time()
    logger.info("**start**")
    result = await asyncio.gather(
        *[
            loop.run_in_executor(None, partial(mysleep, "x", 1)),
            loop.run_in_executor(None, partial(mysleep, "y", 1)),
            loop.run_in_executor(None, partial(mysleep, "z", 2)),
        ]
    )
    logger.info("**end** %r %r", result, time.time() - st)


logging.basicConfig(level=logging.INFO, format="%(asctime)s" + logging.BASIC_FORMAT)
asyncio.run(run(), debug=True)

01use_process_pool_executor.py

import logging
import asyncio
import time
from concurrent.futures import ProcessPoolExecutor
from functools import partial
from mysleep import mysleep

logger = logging.getLogger(__name__)


async def run():
    loop = asyncio.get_event_loop()
    st = time.time()
    logger.info("**start**")
    executor = ProcessPoolExecutor(max_workers=3)
    result = await asyncio.gather(
        *[
            loop.run_in_executor(executor, partial(mysleep, "x", 1)),
            loop.run_in_executor(executor, partial(mysleep, "y", 1)),
            loop.run_in_executor(executor, partial(mysleep, "z", 2)),
        ]
    )
    logger.info("**end** %r %r", result, time.time() - st)


logging.basicConfig(level=logging.INFO, format="%(asctime)s" + logging.BASIC_FORMAT)
asyncio.run(run(), debug=True)

README.md

build

$ pip install -r requirements.txt
$ make build

check

$ make build
$ make check
PYTHONPATH=. python -c 'import mysleep; print(mysleep.mysleep("OK", 1))'
('    ... start mysleep', 'OK', 1)
('    ... end mysleep', 'OK')
OK

thread pool executorを使った方。4s掛かっている(1+1+2)

$ PYTHONPATH=. python 00use_thread_pool_executor.py
2019-03-20 20:54:59,837INFO:__main__:**start**
('    ... start mysleep', 'x', 1)
('    ... end mysleep', 'x')
('    ... start mysleep', 'y', 1)
('    ... end mysleep', 'y')
('    ... start mysleep', 'z', 2)
('    ... end mysleep', 'z')
2019-03-20 20:55:03,846WARNING:asyncio:Executing <Task pending coro=<run() running at 00use_thread_pool_executor.py:18> wait_for=<_GatheringFuture pending cb=[<TaskWakeupMethWrapper object at 0x7f4d1a05a828>()] created at /usr/lib/python3.7/asyncio/tasks.py:615> cb=[_run_until_complete_cb() at /usr/lib/python3.7/asyncio/base_events.py:158] created at /usr/lib/python3.7/asyncio/base_events.py:563> took 4.009 seconds
2019-03-20 20:55:03,850INFO:__main__:**end** ['x', 'y', 'z'] 4.012997150421143

process pool executorを使っている方。2sで済む(max(1,1,2))

PYTHONPATH=. python 01use_process_pool_executor.py
2019-03-20 20:55:03,959INFO:__main__:**start**
('    ... start mysleep', 'x', 1)
('    ... start mysleep', 'y', 1)
('    ... start mysleep', 'z', 2)
('    ... end mysleep', 'x')
('    ... end mysleep', 'y')
2019-03-20 20:55:04,967INFO:asyncio:poll took 1001.933 ms: 1 events
('    ... end mysleep', 'z')
2019-03-20 20:55:05,969INFO:__main__:**end** ['x', 'y', 'z'] 2.0095720291137695

gilを開放すると

mysleep.pyxを変えてGILを開放するとThreadPoolExecutorでも2s(max(1,1,2))で終わる様になる。

例えば以下の様な変更を加える。

diff --git a/daily/20190320/example_run_in_executor/mysleep.pyx b/daily/20190320/example_run_in_executor/mysleep.pyx
index c27e4da..bbf6764 100644
--- a/daily/20190320/example_run_in_executor/mysleep.pyx
+++ b/daily/20190320/example_run_in_executor/mysleep.pyx
@@ -2,6 +2,7 @@ from posix.unistd cimport sleep
 
 def mysleep(x, unsigned int n):
     print("    ... start mysleep", x, n)
-    sleep(n)
+    with nogil:
+        sleep(n)
     print("    ... end mysleep", x)
     return x

2sで終わる

$ PYTHONPATH=. python 00use_thread_pool_executor.py
2019-03-21 15:12:40,999INFO:__main__:**start**
('    ... start mysleep', 'x', 1)
('    ... start mysleep', 'y', 1)
('    ... start mysleep', 'z', 2)
('    ... end mysleep', 'x')
('    ... end mysleep', 'y')
2019-03-21 15:12:42,004INFO:asyncio:poll took 1000.853 ms: 1 events
('    ... end mysleep', 'z')
2019-03-21 15:12:43,005INFO:__main__:**end** ['x', 'y', 'z'] 2.00624680519104

Makefile

build: clean
	python setup.py build_ext --inplace
check:
	PYTHONPATH=. python -c 'import mysleep; print(mysleep.mysleep("OK", 1))'

run:
	PYTHONPATH=. python 00use_thread_pool_executor.py
	PYTHONPATH=. python 01use_process_pool_executor.py

clean:
	rm -f *.so *.c

mysleep.pyx

from posix.unistd cimport sleep

def mysleep(x, unsigned int n):
    print("    ... start mysleep", x, n)
    with nogil:
        sleep(n)
    print("    ... end mysleep", x)
    return x

requirements.txt

cython

setup.py

from distutils.core import setup
from distutils.extension import Extension
from Cython.Distutils import build_ext

setup(
    cmdclass={"build_ext": build_ext},
    ext_modules=[Extension("mysleep", sources=["mysleep.pyx"], libraries=["m"])],
)