bench_queue.py

#!/usr/bin/env python3
import time
import functools
from threading import Thread
from itertools import count
import asyncio
import queue

import janus


class Timer():
    def __init__(self):
        self._start = None

    def start(self):
        self._start = time.time()

    def stop(self):
        return time.time() - self._start


def consumer_thread_sync(taskq, result):
    while True:
        task = taskq.get()
        if task is None:
            return
        result[0] = task


async def consumer_thread_async(taskq, result):
    while True:
        task = await taskq.get()
        if task is None:
            return
        result[0] = task


async def producer_thread_async(taskq):
    for x in range(10000 + 1):
        await taskq.put(x)
    await taskq.put(None)


def producer_thread_sync(taskq):
    for x in range(10000 + 1):
        taskq.put(x)
    taskq.put(None)


def bench_async_to_sync():
    taskq = janus.Queue(maxsize=100)

    loop = asyncio.get_event_loop()

    result = [None]
    consumer = Thread(target=consumer_thread_sync, args=[taskq.sync_q, result])
    consumer.start()

    loop.run_until_complete(producer_thread_async(taskq.async_q))
    consumer.join()

    assert result[0] == 10000

def bench_sync_to_async():
    taskq = janus.Queue(maxsize=100)

    loop = asyncio.get_event_loop()

    result = [None]
    producer = loop.run_in_executor(None, producer_thread_sync, taskq.sync_q)
    loop.run_until_complete(consumer_thread_async(taskq.async_q, result))
    loop.run_until_complete(producer)

    assert result[0] == 10000

def bench_async():
    taskq = asyncio.Queue(maxsize=100)

    loop = asyncio.get_event_loop()

    result = [None]
    consumer = loop.create_task(consumer_thread_async(taskq, result))
    loop.create_task(producer_thread_async(taskq))
    loop.run_until_complete(consumer)

    assert result[0] == 10000


def bench_sync():
    taskq = queue.Queue(maxsize=100)

    loop = asyncio.get_event_loop()

    result = [None]
    consumer = Thread(target=consumer_thread_sync, args=[taskq, result])
    consumer.start()

    producer = Thread(target=producer_thread_sync, args=[taskq])
    producer.start()

    consumer.join()

    assert result[0] == 10000


def main(**kwargs):
    timer = Timer()

    timer.start()
    bench_sync_to_async()
    elapsed_a2s = timer.stop()
    
    timer.start()
    bench_sync_to_async()
    elapsed_s2a = timer.stop()
    
    timer.start()
    bench_async()
    elapsed_a2a = timer.stop()

    timer.start()
    bench_sync()
    elapsed_s2s = timer.stop()

    print('------------------------------------')
    print('Async -> sync: %.2f sec.' % elapsed_a2s)
    print('Sync -> async: %.2f sec.' % elapsed_s2a)
    print('Async: %.2f sec.' % elapsed_a2a)
    print('Sync: %.2f sec.' % elapsed_s2s)



if __name__ == '__main__':
    main()

Add case of sync as producer and async as consumer