summaryrefslogtreecommitdiffstats
path: root/Lib
diff options
context:
space:
mode:
Diffstat (limited to 'Lib')
-rw-r--r--Lib/Queue.py5
-rw-r--r--Lib/mutex.py6
-rwxr-xr-xLib/pydoc.py9
-rw-r--r--Lib/shlex.py16
-rw-r--r--Lib/test/test_bisect.py17
-rw-r--r--Lib/test/test_deque.py337
-rw-r--r--Lib/threading.py5
7 files changed, 360 insertions, 35 deletions
diff --git a/Lib/Queue.py b/Lib/Queue.py
index 980aee6..44c9ca3 100644
--- a/Lib/Queue.py
+++ b/Lib/Queue.py
@@ -1,6 +1,7 @@
"""A multi-producer, multi-consumer queue."""
from time import time as _time, sleep as _sleep
+from collections import deque
__all__ = ['Empty', 'Full', 'Queue']
@@ -184,7 +185,7 @@ class Queue:
# Initialize the queue representation
def _init(self, maxsize):
self.maxsize = maxsize
- self.queue = []
+ self.queue = deque()
def _qsize(self):
return len(self.queue)
@@ -203,4 +204,4 @@ class Queue:
# Get an item from the queue
def _get(self):
- return self.queue.pop(0)
+ return self.queue.popleft()
diff --git a/Lib/mutex.py b/Lib/mutex.py
index e15710a..5d35bdf 100644
--- a/Lib/mutex.py
+++ b/Lib/mutex.py
@@ -12,11 +12,13 @@ Of course, no multi-threading is implied -- hence the funny interface
for lock, where a function is called once the lock is aquired.
"""
+from collections import deque
+
class mutex:
def __init__(self):
"""Create a new mutex -- initially unlocked."""
self.locked = 0
- self.queue = []
+ self.queue = deque()
def test(self):
"""Test the locked bit of the mutex."""
@@ -44,7 +46,7 @@ class mutex:
"""Unlock a mutex. If the queue is not empty, call the next
function with its argument."""
if self.queue:
- function, argument = self.queue.pop(0)
+ function, argument = self.queue.popleft()
function(argument)
else:
self.locked = 0
diff --git a/Lib/pydoc.py b/Lib/pydoc.py
index e53aa16..e6b53c1 100755
--- a/Lib/pydoc.py
+++ b/Lib/pydoc.py
@@ -55,6 +55,7 @@ Mynd you, møøse bites Kan be pretty nasti..."""
import sys, imp, os, re, types, inspect, __builtin__
from repr import Repr
from string import expandtabs, find, join, lower, split, strip, rfind, rstrip
+from collections import deque
# --------------------------------------------------------- common routines
@@ -685,7 +686,7 @@ class HTMLDoc(Doc):
hr = HorizontalRule()
# List the mro, if non-trivial.
- mro = list(inspect.getmro(object))
+ mro = deque(inspect.getmro(object))
if len(mro) > 2:
hr.maybe()
push('<dl><dt>Method resolution order:</dt>\n')
@@ -763,7 +764,7 @@ class HTMLDoc(Doc):
while attrs:
if mro:
- thisclass = mro.pop(0)
+ thisclass = mro.popleft()
else:
thisclass = attrs[0][2]
attrs, inherited = _split_list(attrs, lambda t: t[2] is thisclass)
@@ -1083,7 +1084,7 @@ class TextDoc(Doc):
push = contents.append
# List the mro, if non-trivial.
- mro = list(inspect.getmro(object))
+ mro = deque(inspect.getmro(object))
if len(mro) > 2:
push("Method resolution order:")
for base in mro:
@@ -1152,7 +1153,7 @@ class TextDoc(Doc):
inspect.classify_class_attrs(object))
while attrs:
if mro:
- thisclass = mro.pop(0)
+ thisclass = mro.popleft()
else:
thisclass = attrs[0][2]
attrs, inherited = _split_list(attrs, lambda t: t[2] is thisclass)
diff --git a/Lib/shlex.py b/Lib/shlex.py
index b302699..ccf9038 100644
--- a/Lib/shlex.py
+++ b/Lib/shlex.py
@@ -9,6 +9,7 @@
import os.path
import sys
+from collections import deque
try:
from cStringIO import StringIO
@@ -45,11 +46,11 @@ class shlex:
self.escape = '\\'
self.escapedquotes = '"'
self.state = ' '
- self.pushback = []
+ self.pushback = deque()
self.lineno = 1
self.debug = 0
self.token = ''
- self.filestack = []
+ self.filestack = deque()
self.source = None
if self.debug:
print 'shlex: reading from %s, line %d' \
@@ -59,13 +60,13 @@ class shlex:
"Push a token onto the stack popped by the get_token method"
if self.debug >= 1:
print "shlex: pushing token " + `tok`
- self.pushback.insert(0, tok)
+ self.pushback.appendleft(tok)
def push_source(self, newstream, newfile=None):
"Push an input source onto the lexer's input source stack."
if isinstance(newstream, basestring):
newstream = StringIO(newstream)
- self.filestack.insert(0, (self.infile, self.instream, self.lineno))
+ self.filestack.appendleft((self.infile, self.instream, self.lineno))
self.infile = newfile
self.instream = newstream
self.lineno = 1
@@ -78,8 +79,7 @@ class shlex:
def pop_source(self):
"Pop the input source stack."
self.instream.close()
- (self.infile, self.instream, self.lineno) = self.filestack[0]
- self.filestack = self.filestack[1:]
+ (self.infile, self.instream, self.lineno) = self.filestack.popleft()
if self.debug:
print 'shlex: popping to %s, line %d' \
% (self.instream, self.lineno)
@@ -88,7 +88,7 @@ class shlex:
def get_token(self):
"Get a token from the input stream (or from stack if it's nonempty)"
if self.pushback:
- tok = self.pushback.pop(0)
+ tok = self.pushback.popleft()
if self.debug >= 1:
print "shlex: popping token " + `tok`
return tok
@@ -226,7 +226,7 @@ class shlex:
or self.whitespace_split:
self.token = self.token + nextchar
else:
- self.pushback.insert(0, nextchar)
+ self.pushback.appendleft(nextchar)
if self.debug >= 2:
print "shlex: I see punctuation in word state"
self.state = ' '
diff --git a/Lib/test/test_bisect.py b/Lib/test/test_bisect.py
index 809d8af..6bb2112 100644
--- a/Lib/test/test_bisect.py
+++ b/Lib/test/test_bisect.py
@@ -170,23 +170,6 @@ This example uses bisect() to look up a letter grade for an exam total
>>> map(grade, [33, 99, 77, 44, 12, 88])
['E', 'A', 'B', 'D', 'F', 'A']
-The bisect module can be used with the Queue module to implement
-a priority queue (example courtesy of Fredrik Lundh):
-
->>> import Queue, bisect
->>> class PriorityQueue(Queue.Queue):
-... def _put(self, item):
-... bisect.insort(self.queue, item)
-...
->>> queue = PriorityQueue(0)
->>> queue.put((2, "second"))
->>> queue.put((1, "first"))
->>> queue.put((3, "third"))
->>> queue.get()
-(1, 'first')
->>> queue.get()
-(2, 'second')
-
"""
#------------------------------------------------------------------------------
diff --git a/Lib/test/test_deque.py b/Lib/test/test_deque.py
new file mode 100644
index 0000000..6221c91
--- /dev/null
+++ b/Lib/test/test_deque.py
@@ -0,0 +1,337 @@
+from collections import deque
+import unittest
+from test import test_support
+import copy
+import cPickle as pickle
+from cStringIO import StringIO
+
+BIG = 100000
+
+class TestBasic(unittest.TestCase):
+
+ def test_basics(self):
+ d = deque(xrange(100))
+ d.__init__(xrange(100, 200))
+ for i in xrange(200, 400):
+ d.append(i)
+ for i in reversed(xrange(-200, 0)):
+ d.appendleft(i)
+ self.assertEqual(list(d), range(-200, 400))
+ self.assertEqual(len(d), 600)
+
+ left = [d.popleft() for i in xrange(250)]
+ self.assertEqual(left, range(-200, 50))
+ self.assertEqual(list(d), range(50, 400))
+
+ right = [d.pop() for i in xrange(250)]
+ right.reverse()
+ self.assertEqual(right, range(150, 400))
+ self.assertEqual(list(d), range(50, 150))
+
+ def test_len(self):
+ d = deque('ab')
+ self.assertEqual(len(d), 2)
+ d.popleft()
+ self.assertEqual(len(d), 1)
+ d.pop()
+ self.assertEqual(len(d), 0)
+ self.assertRaises(LookupError, d.pop)
+ self.assertEqual(len(d), 0)
+ d.append('c')
+ self.assertEqual(len(d), 1)
+ d.appendleft('d')
+ self.assertEqual(len(d), 2)
+ d.clear()
+ self.assertEqual(len(d), 0)
+
+ def test_underflow(self):
+ d = deque()
+ self.assertRaises(LookupError, d.pop)
+ self.assertRaises(LookupError, d.popleft)
+
+ def test_clear(self):
+ d = deque(xrange(100))
+ self.assertEqual(len(d), 100)
+ d.clear()
+ self.assertEqual(len(d), 0)
+ self.assertEqual(list(d), [])
+
+ def test_repr(self):
+ d = deque(xrange(200))
+ e = eval(repr(d))
+ self.assertEqual(list(d), list(e))
+ d.append(d)
+ self.assert_('...' in repr(d))
+
+ def test_print(self):
+ d = deque(xrange(200))
+ d.append(d)
+ f = StringIO()
+ print >> f, d,
+ self.assertEqual(f.getvalue(), repr(d))
+ f.close()
+
+ def test_hash(self):
+ self.assertRaises(TypeError, hash, deque('abc'))
+
+ def test_long_steadystate_queue_popleft(self):
+ for size in (0, 1, 2, 100, 1000):
+ d = deque(xrange(size))
+ append, pop = d.append, d.popleft
+ for i in xrange(size, BIG):
+ append(i)
+ x = pop()
+ if x != i - size:
+ self.assertEqual(x, i-size)
+ self.assertEqual(list(d), range(BIG-size, BIG))
+
+ def test_long_steadystate_queue_popright(self):
+ for size in (0, 1, 2, 100, 1000):
+ d = deque(reversed(xrange(size)))
+ append, pop = d.appendleft, d.pop
+ for i in xrange(size, BIG):
+ append(i)
+ x = pop()
+ if x != i - size:
+ self.assertEqual(x, i-size)
+ self.assertEqual(list(reversed(list(d))), range(BIG-size, BIG))
+
+ def test_big_queue_popleft(self):
+ pass
+ d = deque()
+ append, pop = d.append, d.popleft
+ for i in xrange(BIG):
+ append(i)
+ for i in xrange(BIG):
+ x = pop()
+ if x != i:
+ self.assertEqual(x, i)
+
+ def test_big_queue_popright(self):
+ d = deque()
+ append, pop = d.appendleft, d.pop
+ for i in xrange(BIG):
+ append(i)
+ for i in xrange(BIG):
+ x = pop()
+ if x != i:
+ self.assertEqual(x, i)
+
+ def test_big_stack_right(self):
+ d = deque()
+ append, pop = d.append, d.pop
+ for i in xrange(BIG):
+ append(i)
+ for i in reversed(xrange(BIG)):
+ x = pop()
+ if x != i:
+ self.assertEqual(x, i)
+ self.assertEqual(len(d), 0)
+
+ def test_big_stack_left(self):
+ d = deque()
+ append, pop = d.appendleft, d.popleft
+ for i in xrange(BIG):
+ append(i)
+ for i in reversed(xrange(BIG)):
+ x = pop()
+ if x != i:
+ self.assertEqual(x, i)
+ self.assertEqual(len(d), 0)
+
+ def test_roundtrip_iter_init(self):
+ d = deque(xrange(200))
+ e = deque(d)
+ self.assertNotEqual(id(d), id(e))
+ self.assertEqual(list(d), list(e))
+
+ def test_pickle(self):
+ d = deque(xrange(200))
+ s = pickle.dumps(d)
+ e = pickle.loads(s)
+ self.assertNotEqual(id(d), id(e))
+ self.assertEqual(list(d), list(e))
+
+ def test_deepcopy(self):
+ mut = [10]
+ d = deque([mut])
+ e = copy.deepcopy(d)
+ self.assertEqual(list(d), list(e))
+ mut[0] = 11
+ self.assertNotEqual(id(d), id(e))
+ self.assertNotEqual(list(d), list(e))
+
+ def test_copy(self):
+ mut = [10]
+ d = deque([mut])
+ e = copy.copy(d)
+ self.assertEqual(list(d), list(e))
+ mut[0] = 11
+ self.assertNotEqual(id(d), id(e))
+ self.assertEqual(list(d), list(e))
+
+def R(seqn):
+ 'Regular generator'
+ for i in seqn:
+ yield i
+
+class G:
+ 'Sequence using __getitem__'
+ def __init__(self, seqn):
+ self.seqn = seqn
+ def __getitem__(self, i):
+ return self.seqn[i]
+
+class I:
+ 'Sequence using iterator protocol'
+ def __init__(self, seqn):
+ self.seqn = seqn
+ self.i = 0
+ def __iter__(self):
+ return self
+ def next(self):
+ if self.i >= len(self.seqn): raise StopIteration
+ v = self.seqn[self.i]
+ self.i += 1
+ return v
+
+class Ig:
+ 'Sequence using iterator protocol defined with a generator'
+ def __init__(self, seqn):
+ self.seqn = seqn
+ self.i = 0
+ def __iter__(self):
+ for val in self.seqn:
+ yield val
+
+class X:
+ 'Missing __getitem__ and __iter__'
+ def __init__(self, seqn):
+ self.seqn = seqn
+ self.i = 0
+ def next(self):
+ if self.i >= len(self.seqn): raise StopIteration
+ v = self.seqn[self.i]
+ self.i += 1
+ return v
+
+class N:
+ 'Iterator missing next()'
+ def __init__(self, seqn):
+ self.seqn = seqn
+ self.i = 0
+ def __iter__(self):
+ return self
+
+class E:
+ 'Test propagation of exceptions'
+ def __init__(self, seqn):
+ self.seqn = seqn
+ self.i = 0
+ def __iter__(self):
+ return self
+ def next(self):
+ 3/0
+
+class S:
+ 'Test immediate stop'
+ def __init__(self, seqn):
+ pass
+ def __iter__(self):
+ return self
+ def next(self):
+ raise StopIteration
+
+from itertools import chain, imap
+def L(seqn):
+ 'Test multiple tiers of iterators'
+ return chain(imap(lambda x:x, R(Ig(G(seqn)))))
+
+
+class TestVariousIteratorArgs(unittest.TestCase):
+
+ def test_constructor(self):
+ for s in ("123", "", range(1000), ('do', 1.2), xrange(2000,2200,5)):
+ for g in (G, I, Ig, S, L, R):
+ self.assertEqual(list(deque(g(s))), list(g(s)))
+ self.assertRaises(TypeError, deque, X(s))
+ self.assertRaises(TypeError, deque, N(s))
+ self.assertRaises(ZeroDivisionError, deque, E(s))
+
+ def test_iter_with_altered_data(self):
+ d = deque('abcdefg')
+ it = iter(d)
+ d.pop()
+ self.assertRaises(RuntimeError, it.next)
+
+class Deque(deque):
+ pass
+
+class TestSubclass(unittest.TestCase):
+
+ def test_basics(self):
+ d = Deque(xrange(100))
+ d.__init__(xrange(100, 200))
+ for i in xrange(200, 400):
+ d.append(i)
+ for i in reversed(xrange(-200, 0)):
+ d.appendleft(i)
+ self.assertEqual(list(d), range(-200, 400))
+ self.assertEqual(len(d), 600)
+
+ left = [d.popleft() for i in xrange(250)]
+ self.assertEqual(left, range(-200, 50))
+ self.assertEqual(list(d), range(50, 400))
+
+ right = [d.pop() for i in xrange(250)]
+ right.reverse()
+ self.assertEqual(right, range(150, 400))
+ self.assertEqual(list(d), range(50, 150))
+
+ d.clear()
+ self.assertEqual(len(d), 0)
+
+ def test_copy_pickle(self):
+
+ d = Deque('abc')
+
+ e = d.__copy__()
+ self.assertEqual(type(d), type(e))
+ self.assertEqual(list(d), list(e))
+
+ e = Deque(d)
+ self.assertEqual(type(d), type(e))
+ self.assertEqual(list(d), list(e))
+
+ s = pickle.dumps(d)
+ e = pickle.loads(s)
+ self.assertNotEqual(id(d), id(e))
+ self.assertEqual(type(d), type(e))
+ self.assertEqual(list(d), list(e))
+
+
+#==============================================================================
+
+def test_main(verbose=None):
+ import sys
+ from test import test_sets
+ test_classes = (
+ TestBasic,
+ TestVariousIteratorArgs,
+ TestSubclass,
+ )
+
+ test_support.run_unittest(*test_classes)
+
+ # verify reference counting
+ if verbose and hasattr(sys, "gettotalrefcount"):
+ import gc
+ counts = [None] * 5
+ for i in xrange(len(counts)):
+ test_support.run_unittest(*test_classes)
+ gc.collect()
+ counts[i] = sys.gettotalrefcount()
+ print counts
+
+if __name__ == "__main__":
+ test_main(verbose=True)
diff --git a/Lib/threading.py b/Lib/threading.py
index c5d5af3..6461adc 100644
--- a/Lib/threading.py
+++ b/Lib/threading.py
@@ -10,6 +10,7 @@ except ImportError:
from time import time as _time, sleep as _sleep
from traceback import format_exc as _format_exc
+from collections import deque
# Rename some stuff so "from threading import *" is safe
__all__ = ['activeCount', 'Condition', 'currentThread', 'enumerate', 'Event',
@@ -639,7 +640,7 @@ def _test():
self.rc = Condition(self.mon)
self.wc = Condition(self.mon)
self.limit = limit
- self.queue = []
+ self.queue = deque()
def put(self, item):
self.mon.acquire()
@@ -657,7 +658,7 @@ def _test():
while not self.queue:
self._note("get(): queue empty")
self.rc.wait()
- item = self.queue.pop(0)
+ item = self.queue.popleft()
self._note("get(): got %s, %d left", item, len(self.queue))
self.wc.notify()
self.mon.release()
generated by cgit v0.12 at 2024-11-16 01:54:07 (GMT)