v3/primitives/queue.py Bugfix - asyntest.py tests Queue more thoroughly.

Author: peterhinch

v3/primitives/queue.py

@@ -27,14 +27,14 @@ def __init__(self, maxsize=0):
         self._evget = asyncio.Event()  # Triggered by get, tested by put
 
     def _get(self):
-        self._evget.set()
+        self._evget.set()  # Schedule all tasks waiting on get
+        self._evget.clear()
         return self._queue.pop(0)
 
     async def get(self):  #  Usage: item = await queue.get()
         while self.empty():  # May be multiple tasks waiting on get()
             # Queue is empty, suspend task until a put occurs
             # 1st of N tasks gets, the rest loop again
-            self._evput.clear()
             await self._evput.wait()
         return self._get()
 
@@ -45,13 +45,13 @@ def get_nowait(self):  # Remove and return an item from the queue.
         return self._get()
 
     def _put(self, val):
-        self._evput.set()
+        self._evput.set()  # Schedule tasks waiting on put
+        self._evput.clear()
         self._queue.append(val)
 
     async def put(self, val):  # Usage: await queue.put(item)
         while self.full():
             # Queue full
-            self._evget.clear()
             await self._evget.wait()
             # Task(s) waiting to get from queue, schedule first Task
         self._put(val)

v3/primitives/tests/asyntest.py

@@ -374,39 +374,97 @@ def condition_test():
 # ************ Queue test ************
 
 from primitives.queue import Queue
-q = Queue()
 
 async def slow_process():
     await asyncio.sleep(2)
     return 42
 
-async def bar():
+async def bar(q):
     print('Waiting for slow process.')
     result = await slow_process()
     print('Putting result onto queue')
     await q.put(result)  # Put result on q
 
-async def foo():
+async def foo(q):
     print("Running foo()")
     result = await q.get()
     print('Result was {}'.format(result))
 
-async def queue_go(delay):
-    asyncio.create_task(foo())
-    asyncio.create_task(bar())
-    await asyncio.sleep(delay)
+async def q_put(n, q):
+    for x in range(8):
+        obj = (n, x)
+        await q.put(obj)
+        await asyncio.sleep(0)
+
+async def q_get(n, q):
+    for x in range(8):
+        await q.get()
+        await asyncio.sleep(0)
+
+async def putter(q):
+    # put some item, then sleep
+    for _ in range(20):
+        await q.put(1)
+        await asyncio.sleep_ms(50)
+
+
+async def getter(q):
+   # checks for new items, and relies on the "blocking" of the get method
+    for _ in range(20):
+        await q.get()
+
+async def queue_go():
+    q = Queue(10)
+    asyncio.create_task(foo(q))
+    asyncio.create_task(bar(q))
+    await asyncio.sleep(3)
+    for n in range(4):
+        asyncio.create_task(q_put(n, q))
+    await asyncio.sleep(1)
+    assert q.qsize() == 10
+    await q.get()
+    await asyncio.sleep(0.1)
+    assert q.qsize() == 10
+    while not q.empty():
+        await q.get()
+        await asyncio.sleep(0.1)
+    assert q.empty()
+    print('Competing put tasks test complete')
+
+    for n in range(4):
+        asyncio.create_task(q_get(n, q))
+    await asyncio.sleep(1)
+    x = 0
+    while not q.full():
+        await q.put(x)
+        await asyncio.sleep(0.3)
+        x += 1
+    assert q.qsize() == 10
+    print('Competing get tasks test complete')
+    await asyncio.gather(
+        putter(q),
+        getter(q)
+        )
+    print('Queue tests complete')
     print("I've seen starships burn off the shoulder of Orion...")
     print("Time to die...")
 
 def queue_test():
-    printexp('''Running (runtime = 3s):
+    printexp('''Running (runtime = 20s):
 Running foo()
 Waiting for slow process.
 Putting result onto queue
+Result was 42
+Competing put tasks test complete
+Competing get tasks test complete
+Queue tests complete
+
+
 I've seen starships burn off the shoulder of Orion...
 Time to die...
-''', 3)
-    asyncio.run(queue_go(3))
+
+''', 20)
+    asyncio.run(queue_go())
 
 def test(n):
     try: