I ran into this in my own project, see https://github.com/simonw/datasette/pull/1481 - then I tried using a fork of this project to run the unit tests against Python 3.10 and got the same error: https://github.com/simonw/janus/runs/3842463703?check_suite_focus=true

 ============================= test session starts ==============================
 platform linux -- Python 3.10.0, pytest-6.2.4, py-1.10.0, pluggy-0.13.1
 rootdir: /home/runner/work/janus/janus
 plugins: cov-2.12.1, asyncio-0.15.1
 collected 72 items
 
 tests/test_async.py FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF                      [ 43%]
 tests/test_mixed.py .FFFFFFFFFFFFFFFFF                                   [ 68%]
 tests/test_sync.py FFFFFFFFFFFFFFFFFFFFFFF                               [100%]
 
 =================================== FAILURES ===================================
 __________________________ TestQueueBasic.test_empty ___________________________
 
 self = <test_async.TestQueueBasic object at 0x7fb0f561e4d0>
 
     @pytest.mark.asyncio
     async def test_empty(self):
 >       _q = janus.Queue()
 
 tests/test_async.py:65: 
 _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
 janus/__init__.py:39: in __init__
     self._async_not_empty = asyncio.Condition(self._async_mutex)
 _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
 
 self = <[AttributeError("'Condition' object has no attribute 'locked'") raised in repr()] Condition object at 0x7fb0f569a620>
 lock = <asyncio.locks.Lock object at 0x7fb0f569ada0 [unlocked]>
 
     def __init__(self, lock=None, *, loop=mixins._marker):
         super().__init__(loop=loop)
         if lock is None:
             lock = Lock()
         elif lock._loop is not self._get_loop():
 >           raise ValueError("loop argument must agree with lock")
 E           ValueError: loop argument must agree with lock

What do these changes do?

Use case: If one wants to create the Queue in the main thread of the app, but use it with an async loop that is running in a thread (not the one in the main thread).

Otw. one would have to create the queue in the thread to pick up the correct loop.

Are there changes in behavior for the user?

no change in behavior if used as before. Now user can pass a custom loop.

Checklist

• [x] I think the code is well written
• [x] Add a new news fragment into the CHANGES folder

This change is 

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.

async is reserved keyword in Python 3.7

Fix issue #95 Most likely https://github.com/Rapptz/discord.py/commit/096584733e8a8025b13f46fa920e18abe19352c1

Unfortunately, changes in #246 broke the code that relies on creating an instance of janus.Queue outside of the scope of a running event loop, for no obvious reason.

Also, the behavior now contradicts with the behavior of asyncio.Queue.

>>> import asyncio
>>> asyncio.Queue()
<Queue at 0x7f76be085730 maxsize=0>

>>> import janus
>>> janus.Queue()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/user/projects/project/.venv/lib64/python3.8/site-packages/janus/__init__.py", line 29, in __init__
    self._loop = current_loop()
RuntimeError: no running event loop

Is there any good reason to not just use asyncio.get_event_loop? From performance perspective it shouldn't be significant.

if loop.run_in_executor(executor,...) is ProcessPoolExecutor instead of None, exceptions will be raised such as inability to serialize thread lock objects and callItems.

The class typing.Deque was first introduced in Python 3.5.4. The parameter for package dependencies should be updated to: python_requires='>=3.5.4'

What do these changes do?

1. Removed stub file
2. Source code fully annotated
3. Added bandit and pyroma linters

Are there changes in behavior for the user?

Related issue number

Checklist

• [ ] I think the code is well written
• [ ] Unit tests for the changes exist
• [ ] Documentation reflects the changes
• [ ] If you provide code modification, please add yourself to CONTRIBUTORS.txt
  • The format is <Name> <Surname>.
  • Please keep alphabetical order, the file is sorted by names.
• [ ] Add a new news fragment into the CHANGES folder
  • name it <issue_id>.<type> (e.g. 588.bugfix)
  • if you don't have an issue_id change it to the pr id after creating the PR
  • ensure type is one of the following:
    • .feature: Signifying a new feature.
    • .bugfix: Signifying a bug fix.
    • .doc: Signifying a documentation improvement.
    • .removal: Signifying a deprecation or removal of public API.
    • .misc: A ticket has been closed, but it is not of interest to users.
  • Make sure to use full sentences with correct case and punctuation, for example: Fix issue with non-ascii contents in doctest text files.

What do these changes do?

Add protocols to avoid the use of private classes _SyncQueueProxy and _AsyncQueueProxy when adding type hints.

Are there changes in behavior for the user?

I assume that not. But it would be good if somebody more experienced could check this.

Related issue number

See https://github.com/aio-libs/janus/issues/372

This PR is not intended to be directly merged, but to open a discussion about this approach.

Using this approach, the following seems to work for adding type hints:

import asyncio
import janus


def threaded(sync_q: janus.SyncQueue[int]):
    for i in range(100):
        sync_q.put(i)
    sync_q.join()


async def async_coro(async_q: janus.AsyncQueue[int]):
    for i in range(100):
        val = await async_q.get()
        assert val == i
        async_q.task_done()


async def main():
    queue: janus.Queue[int] = janus.Queue()
    loop = asyncio.get_running_loop()
    fut = loop.run_in_executor(None, threaded, queue.sync_q)
    await async_coro(queue.async_q)
    await fut
    queue.close()
    await queue.wait_closed()


asyncio.run(main())

Full disclaimer: I never used Protocols before. I skimmed https://www.python.org/dev/peps/pep-0544 and watched https://www.youtube.com/watch?v=kDDCKwP7QgQ Also, I only recently started using typing with Python.

While this approach seems to work, it looks rather verbose.

When I try to add type hints to the example, pylance creates reportPrivateUsage warnings.

import asyncio
import janus


def threaded(sync_q: janus._SyncQueueProxy[int]):
    for i in range(100):
        sync_q.put(i)
    sync_q.join()


async def async_coro(async_q: janus._AsyncQueueProxy[int]):
    for i in range(100):
        val = await async_q.get()
        assert val == i
        async_q.task_done()


async def main():
    queue: janus.Queue[int] = janus.Queue()
    loop = asyncio.get_running_loop()
    fut = loop.run_in_executor(None, threaded, queue.sync_q)
    await async_coro(queue.async_q)
    await fut
    queue.close()
    await queue.wait_closed()


asyncio.run(main())

Warnings:

"_SyncQueueProxy" is private and used outside of the module in which it is declared

"_AsyncQueueProxy" is private and used outside of the module in which it is declared

What is the preferred way to add type hints to janus Queues?

What do these changes do?

Tests now also run against Python 3.10, and code incorporates a workaround for a bug in Python 3.10 that affects Janus.

Related issue number

• #358

Checklist

• [x] I think the code is well written
• [x] Unit tests for the changes exist
• [x] Documentation reflects the changes - (not needed)
• [ ] Add a new news fragment into the CHANGES folder (skipped, not sure how to do this)

Hi aio-libs/janus!

This is not an automatic, 🤖-generated PR, as you can check in my GitHub profile, I work for GitHub and I am part of the GitHub Security Lab which is helping out with the migration of LGTM configurations to Code Scanning. You might have heard that we've integrated LGTM's underlying CodeQL analysis engine natively into GitHub. The result is GitHub code scanning!

With LGTM fully integrated into code scanning, we are focused on improving CodeQL within the native GitHub code scanning experience. In order to take advantage of current and future improvements to our analysis capabilities, we suggest you enable code scanning on your repository. Please take a look at our blog post for more information.

This pull request enables code scanning by adding an auto-generated codeql.yml workflow file for GitHub Actions to your repository — take a look! We tested it before opening this pull request, so all should be working :heavy_check_mark:. In fact, you might already have seen some alerts appear on this pull request!

Where needed and if possible, we’ve adjusted the configuration to the needs of your particular repository. But of course, you should feel free to tweak it further! Check this page for detailed documentation.

Questions? Check out the FAQ below!

FAQ

I am having trouble creating the janus Queue outside of asyncio. It requires me to make dummy background tasks to just create the object, since not running loop exists I am wonder if this could be lazy or allow the loop to be passed in the constructor? It makes sync to async code a tad harder.

What do these changes do?

Prior to this commit, you could only initialize a queue in an asynchronous function. Now it is possible in synchronous, but you need to pass the event loop (even empty\not running) to the constructor

Just a new way to init janus. But gives more ways to handle exceptions and code manage.

(Minimum code to understand the changes)

Before

import asyncio
import janus

def async_f(async_q: janus.AsyncQueue[int]):
    ...
def sync_f(sync_q: janus.SyncQueue[int]):
    ...
async def main() -> None:
    loop = asyncio.get_running_loop()
    queue: janus.Queue[int] = janus.Queue()

    loop.run_in_executor(None, sync_f, queue.sync_q)
    await async_f(queue.async_q)
    ...

if __name__ == "__main__":
    asyncio.run(main())

Now

import asyncio
import janus

def async_f(async_q: janus.AsyncQueue[int]):
    ...
def sync_f(sync_q: janus.SyncQueue[int]):
    ...

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    queue: janus.Queue[int] = janus.Queue(loop=loop)
    loop.create_task(async_f(queue.async_q))
    loop.run_in_executor(None, sync_f, queue.sync_q)
    try:
         loop.run_forever()
    except KeyboardInterrupt:
         pass

Are there changes in behavior for the user?

There are no behavior changes for users.

Checklist

• [X] I think the code is well written
• [ ] Unit tests for the changes exist
• [ ] Documentation reflects the changes
• [ ] Add a new news fragment into the CHANGES folder
  • name it <issue_id>.<type> (e.g. 588.bugfix)
  • if you don't have an issue_id change it to the pr id after creating the PR
  • ensure type is one of the following:
    • .feature: Signifying a new feature.
    • .bugfix: Signifying a bug fix.
    • .doc: Signifying a documentation improvement.
    • .removal: Signifying a deprecation or removal of public API.
    • .misc: A ticket has been closed, but it is not of interest to users.
  • Make sure to use full sentences with correct case and punctuation, for example: Fix issue with non-ascii contents in doctest text files.

What do these changes do?

License file is missing from distribution. This change fixes an issue.

Are there changes in behavior for the user?

Yes. Users can read license file, packagers shouldn't add hacks to package license file.

Related issue number

No issues.

Checklist

• [x] I think the code is well written
• [x] Unit tests for the changes exist
• [x] Documentation reflects the changes
• [x] Add a new news fragment into the CHANGES folder
  • name it <issue_id>.<type> (e.g. 588.bugfix)
  • if you don't have an issue_id change it to the pr id after creating the PR
  • ensure type is one of the following:
    • .feature: Signifying a new feature.
    • .bugfix: Signifying a bug fix.
    • .doc: Signifying a documentation improvement.
    • .removal: Signifying a deprecation or removal of public API.
    • .misc: A ticket has been closed, but it is not of interest to users.
  • Make sure to use full sentences with correct case and punctuation, for example: Fix issue with non-ascii contents in doctest text files.

(I've checked all checkboxes, however they are not applicable, and there is no CHANGES folder in this repository)

Hey! Recently I found your library and it's really helping me with the communication between sub thread coroutines and main thread functions. One of my use cases needs the ability to clear all the already exiting messages in a queue, or to insert the new message to the end of the queue. That ability is not possible when using a normal queue, so I wanted to know if there is any planning for a double-ended queue support any time soon. Thanks!

So I've been testing the performance a bit, varying e.g.

• 1:1, 16:1, 64:1 producer : consumer ratio
• normal/lifo/priority queue etc
• asyncio/uvloop event loop
• async -> async, async -> sync, sync -> sync
• (probably more)

I found that, janus queues are ~5x slower in sync->sync, ~9x slower in sync->async, and ~15x slower in async->async. This is pretty much consistent across all parameter sets.

This confirmed my suspicion that the performance gain of parallel computation is often less than the cost of using e.g. threading.Lock a lot (the GIL certainly doesn't help either).

Right now, I can imagine that many users have incorrect expectations of janus. To avoid this, you could add an example that shows how janus can outperform single-threaded asyncio, by employing multiple threads. Additionally, a caveat about janus' performance would be helpful.