benchmark_dask.py

# pip install dask[complete]

from dask.distributed import Client, Worker, Scheduler, wait

def inc(x):
    return x + 1

async def benchmark():
    async with Scheduler() as scheduler:
        async with Worker(scheduler.address):
            async with Client(scheduler.address, asynchronous=True) as client:

                futures = client.map(inc, range(1000))
                for _ in range(10):
                    futures = client.map(inc, futures)

                await wait(futures)

import asyncio
asyncio.get_event_loop().run_until_complete(benchmark())