`UniversalQueue`'s deadlock on cancellation

[x42005e1f]

UniversalQueue can cause a deadlock in any put() method (sync / curio / asyncio) as a result of cancelling a get() call from Curio. This is because the cancellation handling code contains an attachment of a callback containing a _put() call to a future object. All callbacks are handled by the one that puts the future object in the cancelled or finished state, that is, calls future.cancel() / future.set_result() / future.set_exception(). And this is exactly the kind of call that occurs inside the _put() method while the lock is held, resulting in a second attempt to acquire the lock in a nested call to _put(), which is a deadlock.

Below is the simplified code to reproduce the issue.

import curio


async def main():
    q = curio.UniversalQueue()

    async with curio.TaskGroup() as g:
        task = await g.spawn(q.get)

        await task.cancel()

        await q.put(42)  # hangs!

    print("passed")


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

And here is a stack trace after Control-C:

Traceback (most recent call last):
  ...
  File "/home/user/workspace/test.py", line 12, in main
    await q.put(42)
  File "/home/user/.local/lib/python3.12/site-packages/curio/queue.py", line 262, in put
    fut = self._put(item)
          ^^^^^^^^^^^^^^^
  File "/home/user/.local/lib/python3.12/site-packages/curio/queue.py", line 247, in _put
    getter.set_result(self._get_item())  # queue.popleft())
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/concurrent/futures/_base.py", line 550, in set_result
    self._invoke_callbacks()
  File "/usr/lib/python3.12/concurrent/futures/_base.py", line 340, in _invoke_callbacks
    callback(self)
  File "/home/user/.local/lib/python3.12/site-packages/curio/queue.py", line 201, in <lambda>
    lambda f: self._put(f.result(), True) if not f.cancelled() else None
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/user/.local/lib/python3.12/site-packages/curio/queue.py", line 226, in _put
    with self._mutex:
KeyboardInterrupt

The quickest solution is to replace threading.Lock with threading.RLock.