Add benchmarks (#706)

Use codspeed.io as the infrastructure

.github/workflows/ci.yml

@@ -79,9 +79,43 @@ jobs:
       with:
         key: unit-${{ matrix.python-version }}
 
+  benchmark:
+    name: Benchmark
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v4
+    - name: Setup Python 3.13
+      uses: actions/setup-python@v5
+      with:
+        python-version: 3.13
+    - name: Get pip cache dir
+      id: pip-cache
+      run: |
+        echo "dir=$(pip cache dir)" >> $GITHUB_OUTPUT    # - name: Cache
+      shell: bash
+    - name: Cache PyPI
+      uses: actions/cache@v4
+      with:
+        key: pip-ci-ubuntu-3.13-${{ hashFiles('requirements-dev.txt') }}
+        path: ${{ steps.pip-cache.outputs.dir }}
+        restore-keys: |
+            pip-ci-ubuntu-3.13-
+    - name: Install dependencies
+      uses: py-actions/[email protected]
+      with:
+        path: requirements-dev.txt
+    - name: Run benchmarks
+      uses: CodSpeedHQ/action@v3
+      with:
+        token: ${{ secrets.CODSPEED_TOKEN }}
+        run: python -Im pytest --no-cov -vvvvv --codspeed
+
+
   check:  # The branch protection check
     if: always()
-    needs: [lint, unit]
+    needs: [lint, unit, benchmark]
     runs-on: ubuntu-latest
     steps:
     - name: Decide whether the needed jobs succeeded or failed

requirements-dev.txt

@@ -1,4 +1,5 @@
 -e .
+backports.asyncio.runner==1.1.0; python_version < "3.11"
 black==24.10.0
 bandit==1.8.0
 coverage==7.6.8
@@ -9,6 +10,7 @@ pyroma==4.2
 pytest-cov==6.0.0
 pytest==8.3.4
 pytest-asyncio==0.24.0
+pytest-codspeed==3.0.0
 isort==5.13.2
 tox==4.23.2
 wheel==0.45.1

setup.cfg

@@ -59,3 +59,7 @@ junit_family=xunit2
 asyncio_mode=strict
 asyncio_default_fixture_loop_scope=function
 filterwarnings=error
+    # pytest-asyncio 0.24.0 does't close the event loop for some reason
+    # it is the source of unclosed loop and socket warnings
+    ignore:unclosed event loop:ResourceWarning
+    ignore:unclosed <socket.socket.+>:ResourceWarning

tests/test_benchmarks.py

@@ -0,0 +1,152 @@
+import asyncio
+import sys
+import janus
+
+if sys.version_info >= (3, 11):
+    from asyncio import Runner
+else:
+    from backports.asyncio.runner import Runner
+
+
+def test_bench_sync_put_async_get(benchmark):
+    q: janus.Queue
+
+    async def init():
+        nonlocal q
+        q = janus.Queue()
+
+    def threaded():
+        for i in range(5):
+            q.sync_q.put(i)
+
+    async def go():
+        for i in range(100):
+            f = asyncio.get_running_loop().run_in_executor(None, threaded)
+            for i in range(5):
+                val = await q.async_q.get()
+                assert val == i
+
+            await f
+            assert q.async_q.empty()
+
+    async def finish():
+        q.close()
+        await q.wait_closed()
+
+    with Runner(debug=True) as runner:
+        runner.run(init())
+
+        @benchmark
+        def _run():
+            runner.run(go())
+
+        runner.run(finish())
+
+
+def test_bench_sync_put_async_join(benchmark):
+    q: janus.Queue
+
+    async def init():
+        nonlocal q
+        q = janus.Queue()
+
+    async def go():
+        for i in range(100):
+            for i in range(5):
+                q.sync_q.put(i)
+
+            async def do_work():
+                await asyncio.sleep(0.01)
+                while not q.async_q.empty():
+                    await q.async_q.get()
+                    q.async_q.task_done()
+
+            task = asyncio.create_task(do_work())
+
+            await q.async_q.join()
+            await task
+
+    async def finish():
+        q.close()
+        await q.wait_closed()
+
+    with Runner(debug=True) as runner:
+        runner.run(init())
+
+        @benchmark
+        def _run():
+            runner.run(go())
+
+        runner.run(finish())
+
+
+def test_bench_async_put_sync_get(benchmark):
+    q: janus.Queue
+
+    async def init():
+        nonlocal q
+        q = janus.Queue()
+
+    def threaded():
+        for i in range(5):
+            val = q.sync_q.get()
+            assert val == i
+
+    async def go():
+        for i in range(100):
+            f = asyncio.get_running_loop().run_in_executor(None, threaded)
+            for i in range(5):
+                await q.async_q.put(i)
+
+            await f
+            assert q.async_q.empty()
+
+    async def finish():
+        q.close()
+        await q.wait_closed()
+
+    with Runner(debug=True) as runner:
+        runner.run(init())
+
+        @benchmark
+        def _run():
+            runner.run(go())
+
+        runner.run(finish())
+
+
+def test_sync_join_async_done(benchmark):
+    q: janus.Queue
+
+    async def init():
+        nonlocal q
+        q = janus.Queue()
+
+    def threaded():
+        for i in range(5):
+            q.sync_q.put(i)
+        q.sync_q.join()
+
+    async def go():
+        for i in range(100):
+            f = asyncio.get_running_loop().run_in_executor(None, threaded)
+            for i in range(5):
+                val = await q.async_q.get()
+                assert val == i
+                q.async_q.task_done()
+
+            await f
+            assert q.async_q.empty()
+
+    async def finish():
+        q.close()
+        await q.wait_closed()
+
+    with Runner(debug=True) as runner:
+        runner.run(init())
+
+        @benchmark
+        def _run():
+            runner.run(go())
+
+        runner.run(finish())

tests/test_mixed.py

@@ -77,16 +77,16 @@ async def test_sync_put_async_join(self):
             q.sync_q.put(i)
 
         async def do_work():
-            await asyncio.sleep(1)
-            while True:
+            await asyncio.sleep(0.1)
+            while not q.async_q.empty():
                 await q.async_q.get()
                 q.async_q.task_done()
 
         task = loop.create_task(do_work())
 
         async def wait_for_empty_queue():
             await q.async_q.join()
-            task.cancel()
+            await task
 
         await wait_for_empty_queue()