Add a Dask benchmark (gh-246)

The Dask scheduler is largely pure-python-code-bound. There is a discussion about the details here. For this reason, it makes a good real-world workload of a distributed system benchmark (even when the benchmark is running locally on a single core).

Thanks to @TomAugspurger for the suggestion, and @mrocklin for providing the meat of the benchmark itself.

Author: mdboom

pyperformance/data-files/benchmarks/MANIFEST

@@ -14,6 +14,7 @@ generators	<local>
 chameleon	<local>
 chaos	<local>
 crypto_pyaes	<local>
+dask	<local>
 deepcopy	<local>
 deltablue	<local>
 django_template	<local>

pyperformance/data-files/benchmarks/bm_dask/pyproject.toml

@@ -0,0 +1,9 @@
+[project]
+name = "pyperformance_bm_dask"
+requires-python = ">=3.8"
+dependencies = ["pyperf"]
+urls = {repository = "https://github.com/python/pyperformance"}
+dynamic = ["version"]
+
+[tool.pyperformance]
+name = "dask"

pyperformance/data-files/benchmarks/bm_dask/requirements.txt

@@ -0,0 +1 @@
+dask[distributed]==2022.2.0

pyperformance/data-files/benchmarks/bm_dask/run_benchmark.py

@@ -0,0 +1,31 @@
+"""
+Benchmark the Dask scheduler running a large number of simple jobs.
+
+Author: Matt Rocklin, Michael Droettboom
+"""
+
+from dask.distributed import Client, Worker, Scheduler, wait
+
+import pyperf
+
+
+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(100))
+                for _ in range(10):
+                    futures = client.map(inc, futures)
+
+                await wait(futures)
+
+
+if __name__ == "__main__":
+    runner = pyperf.Runner()
+    runner.metadata['description'] = "Benchmark dask"
+    runner.bench_async_func('dask', benchmark)