expose the unfinished tasks variable

[richardbaronpenman]

I use queue.empty() and queue._parent._unfinished_tasks == 0 to check whether a queue is complete and this pull request exposes the _unfinished_tasks variable.

My use case is 2 queues that can send work to each other until both are complete - is there a better way to achieve this? Note that I'm not trying to join() since a thread should wake up if there is more work to do.

[asvetlov]

Sorry, I dont understand your need clean but my guts feel that exposing a new property is not necessary.

Both queue.Queue and asyncio.Queue doesn't have such property.

Let's imagine you have two chained queue.Queue instances. How do you solve your need in this case? Maybe two queues is not enough for getting rid of dead locks and you need additional synchronization primitives like semaphores or conditional variables?

[richardbaronpenman]

For example if had a web crawler with a queue for URL's to download and a queue for responses to scrape, then I could use a third variable to check whether all work is complete:

async def crawler(download_q, response_q, work_q):
    while True:
        if download_q.empty():
            if response_q.empty() and work_q.empty():
                break
            else:
                asyncio.sleep(delay)
        else:
            await work_q.put(1)
            url = await download_q.get()
            response = await download(url)
            await response_q.put(response)
            await work_q.get()


async def scraper(download_q, response_q, work_q):
    while True:
        if response_q.empty():
            if download_q.empty() and work_q.empty():
                break
            else:
                asyncio.sleep(delay)
        else:
            await work_q.put(1)
            response = await response_q.get()
            for url in crawl(response):
                await download_q.put(url)
            await work_q.get()

With janus I have been avoiding this additional variable by using task_done() and checking ._parent._unfinished_tasks, which works fine, however would prefer not to rely on this private variable.

[asvetlov]

asyncio.sleep() in your code is a sign of bad design.
The code shouldn't sleep in cycle but wait for explicit event.
Learn how asyncio.Condition works, use https://github.com/python/cpython/blob/master/Lib/queue.py for inspiration.
In your case easier to create to condition variables (have URL for download and have a page content to analyze) in pair with two conventional lists instead of using queues.

Or take a look on crawler example (it's built using slightly different approach than your code).

[richardbaronpenman]

Thanks for sleep tip - have changed to using wait_for.

The 500lines crawler example only uses a single download queue, so avoids the multiple queue problem.
Mine is reading data from a database in a separate thread and then using janus to communicate between the async and threaded parts.

I noticed the queue implementation linked makes unfinished_tasks a public variable:
https://github.com/python/cpython/blob/master/Lib/queue.py#L48

[asvetlov]

Good point.
I'm ok with your proposal but please make tests for covering added properties

[richardbaronpenman]

Thanks - see latest commit

[asvetlov]

tests/test_mixed.py:59:1: W293 blank line contains whitespace

Please fix it.
The test looks correct

[asvetlov]

Thanks!

[asvetlov]

janus 0.3.0 released