bpo-43119: Add a preemption point to asyncio.Queue.put when the queue is unbounded

Lib/asyncio/queues.py

@@ -5,6 +5,7 @@
 
 from . import locks
 from . import mixins
+from . import tasks
 
 
 class QueueEmpty(Exception):
@@ -133,7 +134,11 @@ async def put(self, item):
                     # the call.  Wake up the next in line.
                     self._wakeup_next(self._putters)
                 raise
-        return self.put_nowait(item)
+        self.put_nowait(item)
+        if self._maxsize <= 0:
+            # If the queue is unbounded then it will never be full. Add a
+            # preemption point to give getters a chance.
+            await tasks.sleep(0)
 
     def put_nowait(self, item):
         """Put an item into the queue without blocking.

Lib/test/test_asyncio/test_queues.py

@@ -496,6 +496,18 @@ def test_put_with_waiting_getters(self):
         self.loop.run_until_complete(q.put('a'))
         self.assertEqual(self.loop.run_until_complete(t), 'a')
 
+    def test_infinite_put_with_waiting_getters(self):
+        q = asyncio.Queue()
+        async def put_forever():
+            while True:
+                await q.put('a')
+        put_task = self.loop.create_task(put_forever())
+        test_utils.run_briefly(self.loop)
+        self.assertEqual(self.loop.run_until_complete(q.get()), 'a')
+        put_task.cancel()
+        with self.assertRaises(asyncio.CancelledError):
+            self.loop.run_until_complete(put_task)
+
     def test_why_are_putters_waiting(self):
         # From issue #265.
         asyncio.set_event_loop(self.loop)

Misc/NEWS.d/next/Library/2021-02-03-12-33-46.bpo-43119.rjKCZw.rst

@@ -0,0 +1,2 @@
+Calling the ``put`` method of an unbounded ``asyncio.Queue`` will now yield
+to other coroutines.