Add C version of SimpleQueue

Author: pitrou

Lib/queue.py

@@ -8,16 +8,26 @@
 from heapq import heappush, heappop
 from time import monotonic as time
 
+try:
+    import _queue
+except ImportError:
+    _queue = None
+
 __all__ = ['Empty', 'Full', 'Queue', 'PriorityQueue', 'LifoQueue']
 
-class Empty(Exception):
-    'Exception raised by Queue.get(block=0)/get_nowait().'
-    pass
+
+try:
+    Empty = _queue.Empty
+except AttributeError:
+    class Empty(Exception):
+        'Exception raised by Queue.get(block=0)/get_nowait().'
+        pass
 
 class Full(Exception):
     'Exception raised by Queue.put(block=0)/put_nowait().'
     pass
 
+
 class Queue:
     '''Create a queue object with a given maximum size.
 
@@ -246,7 +256,7 @@ def _get(self):
         return self.queue.pop()
 
 
-class SimpleQueue:
+class _PySimpleQueue:
 
     def __init__(self):
         self._queue = deque()
@@ -262,3 +272,9 @@ def get(self, block=True, timeout=None):
         if not self._count.acquire(block, timeout):
             raise Empty
         return self._queue.popleft()
+
+
+try:
+    SimpleQueue = _queue.SimpleQueue
+except AttributeError:
+    SimpleQueue = _PySimpleQueue

Lib/test/test_queue.py

@@ -8,6 +8,11 @@
 from test import support
 threading = support.import_module('threading')
 
+try:
+    import _queue
+except ImportError:
+    _queue = None
+
 QUEUE_SIZE = 5
 
 def qfull(q):
@@ -356,8 +361,7 @@ def test_failing_queue(self):
         self.failing_queue_test(q)
 
 
-class SimpleQueueTest(unittest.TestCase):
-    type2test = queue.SimpleQueue
+class BaseSimpleQueueTest:
 
     def setUp(self):
         self.q = self.type2test()
@@ -436,7 +440,7 @@ def wrapper(*args, **kwargs):
 
         return results
 
-    def test_simple(self):
+    def test_basic(self):
         q = self.q
         q.put(1)
         q.put(2)
@@ -488,12 +492,25 @@ def test_many_threads_nonblock(self):
     def test_many_threads_timeout(self):
         N = 50
         q = self.q
-        inputs = list(range(10000))
+        inputs = list(range(1000))
         results = self.run_threads(N, N, q, inputs,
                                    self.feed, self.consume_timeout)
 
         self.assertEqual(sorted(results), inputs)
 
 
+class PySimpleQueueTest(BaseSimpleQueueTest, unittest.TestCase):
+    type2test = queue._PySimpleQueue
+
+
+if _queue is not None:
+
+    class CSimpleQueueTest(BaseSimpleQueueTest, unittest.TestCase):
+        type2test = _queue.SimpleQueue
+
+        def test_is_default(self):
+            self.assertIs(self.type2test, queue.SimpleQueue)
+
+
 if __name__ == "__main__":
     unittest.main()