Python asyncio queue use get/put simultaneously

user_13725098 0
Viewed 2079

I'd like to run a Python script that can simulate real-world time-series data and process the data in real-time. That is to say, from a given dataset, (which is ~8hr long measurement), I want to get one-second-long data every second, and process as soon as each second of data is read. In the real world, the data will be gathered every second using some detectors. For this task, I've decided to use Python asyncio module. Below is basically what I came up with.

import scipy
import numpy as np
import asyncio
import time
import queue

q = queue.Queue()

async def get_data (data):

    while True:
        await asyncio.sleep(1)
        q.put(data[idx,:])
        idx += 1
        #Each row of data is read every second. 

async def run_algorithm ():

    while True:

        if q.empty() == True:
            await asyncio.sleep(1)

        data_read = q.get(block = False)

        #I do something here with the read data


async def main (data):

    feed_data = asyncio.create_task(get_data(data))
    process_data = asyncio.create_task(run_algorithm ())
    await asyncio.gather(feed_data, process_data)

While it seems to work fine, the problem is that #I do something here with the read data part. That is where I use my algorithm to process the data in real-time. When I use an algorithm that is fast, it works pretty well. It reads the data every second as it should, and the run_algorithm function waits for a second when there is no data in the queue.

However, when I have a slow algorithm, it does not read the data every second. If the algorithm takes 0.5 seconds to run, then the next data is read in 1.5 seconds, instead of 1 second.

It is like the get_data function slows down as run_algorithm part slows down. Is there a way to get that get_data read the data every second no matter how long the run_algorithm part takes?

async-await python-asyncio asynchronous python-3.x python
2 Answers

As indicated in the comments, you need to use an asyncio queue, in which case you don't need the sleeps in run_algorithm:

q = asyncio.Queue()

async def get_data (data):
    while True:
        await asyncio.sleep(1)
        await q.put(data[idx,:])
        idx += 1
        #Each row of data is read every second. 

async def run_algorithm ():
    while True:
        data_read = await q.get()
        #I do something here with the read data

However, when I have a slow algorithm, it does not read the data every second

It sounds like your algorithm contains CPU-bound or otherwise blocking code. Since asyncio is single-threaded, this causes the event loop to stall. To fix it, you should run blocking code in a separate thread. Asyncio contains a utility for this purpose, run_in_executor. For example, emulating the blocking/CPU-bound call with time.sleep(1) (intentionally not using asyncio.sleep here to emulate blocking), here is how you would call it from asyncio:

        #time.sleep(1)   # this is forbidden
        # instead, invoke blocking code like this:
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(None, time.sleep, 1)
user_16008980

maybe something like:

from time import monotonic

async def get_data (data):
    start = monotonic() 
    for i in range(data.shape[0]):
        delay = start + i - monotonic()
        if delay > 0:
            await asyncio.sleep(delay)
        q.put(data[i,:])

i.e. track when you started, then just sleep for the remaining delay. note that this will tend to cause a many items to be added if the consumer is too slow for one of them

user_13583080
Related