I met the following traceback: Traceback (most recent call last): File "/usr/lib/pypy3/lib-python/3/concurrent/futures/process.py", line 102, in _python_exit thread_wakeup.wakeup() File "/usr/lib/pypy3/lib-python/3/concurrent/futures/process.py", line 90, in wakeup self._writer.send_bytes(b"") File "/usr/lib/pypy3/lib-python/3/multiprocessing/connection.py", line 183, in send_bytes self._check_closed() File "/usr/lib/pypy3/lib-python/3/multiprocessing/connection.py", line 136, in _check_closed raise OSError("handle is closed") OSError: handle is closed debug: OperationError: debug: operror-type: OSError debug: operror-value: handle is closed The code is: @pytest.mark.asyncio async def test_peace_search_store(): import multiprocessing from found import pstore from concurrent import futures db = await open() with futures.ProcessPoolExecutor(max_workers=multiprocessing.cpu_count()) as pool: store = pstore.make('test-pstore', (42,), pool) DOC0 = dict( foundationdb=1, okvs=2, database=42, ) DOC1 = dict( sqlite=1, sql=2, database=3 ) DOC2 = dict( spam=42, ) for uid, doc in enumerate((DOC0, DOC1, DOC2)): await found.transactional(db, pstore.index, store, uid, doc) expected = [(0, 1)] out = await found.transactional(db, pstore.search, store, ["foundationdb"], 10) assert out == expected expected = [(2, 42)] out = await found.transactional(db, pstore.search, store, ["spam"], 10) assert out == expected expected = [(0, 42), (1, 3)] out = await found.transactional(db, pstore.search, store, ["database"], 10) assert out == expected Also pstore.search will call the following pool_for_each_par_map: from aiostream import pipe, stream import asyncio async def pool_for_each_par_map(loop, pool, f, p, iterator): zx = stream.iterate(iterator) zx = zx | pipe.map(lambda x: loop.run_in_executor(pool, p, x)) async with zx.stream() as streamer: limit = pool._max_workers unfinished = [] while True: tasks = [] for i in range(limit): try: task = await streamer.__anext__() except StopAsyncIteration: limit = 0 else: tasks.append(task) tasks = tasks + list(unfinished) assert len(tasks) <= pool._max_workers if not tasks: break finished, unfinished = await asyncio.wait( tasks, return_when=asyncio.FIRST_COMPLETED ) for finish in finished: out = finish.result() f(out) limit = len(finished) You can reproduce with the following cli dance: git clone --branch=rework https://github.com/amirouche/asyncio-foundationdb/ cd asyncio-founndationdb make init poetry run make check I am not sure how to proceed I can try nightly builds but which ones? py3.7 oor py3.8? Feedback welcome.