It has been a while since taking the class Algorithms and Data Structures, so the following module was written as an exercise for implementing Python's collections.deque
class in pure Python without looking at its code in C. The Deque
class appears to pass all tests within the test.test_deque
module, but a lot of patches were made to the code before that became possible. As a result, the code is a mess, and there are many specific errors and style problems throughout the program. Fortunately, my interests are not in specific details.
Are there any glaring problems in the overall architecture of the Deque
class or any of its supporting classes? Supporting examples of the problems are welcome as are suggestions on improving the overall design of the code.
import copy
import functools
import gc
import itertools
import math
import operator
import test.support
import test.test_deque
import threading
import unittest
__all__ = ['Deque']
def main():
test.test_deque.deque = Deque
test.test_deque.TestBasic.test_sizeof = test.support.anticipate_failure(
True
)(test.test_deque.TestBasic.test_sizeof)
test.test_deque.TestSubclass.assertRaises = lambda *args, **kwargs: (
gc.collect(),
unittest.TestCase.assertRaises(*args, **kwargs)
)[1]
test.test_deque.test_main(True)
class AtomicMeta(type):
def __new__(mcs, name, bases, class_dict):
for key, value in class_dict.items():
if callable(value):
class_dict[key] = mcs.wrap(value)
# if '__slots__' not in class_dict:
# class_dict['__slots__'] = ('_AtomicMeta__mutex',)
# else:
# class_dict['__slots__'] += ('_AtomicMeta__mutex',)
class_dict['__new__'] = mcs.atomic_new
return super().__new__(mcs, name, bases, class_dict)
@staticmethod
def wrap(func):
@functools.wraps(func)
def atomic(self, *args, **kwargs):
try:
mutex = self.__mutex
except AttributeError:
mutex = self.__mutex = threading.RLock()
with mutex:
return func(self, *args, **kwargs)
return atomic
@staticmethod
def atomic_new(cls, *args, **kwargs):
instance = object.__new__(cls)
instance.__mutex = threading.RLock()
return instance
class Deque(metaclass=AtomicMeta):
# __slots__ = (
# '__max_len',
# '_left_root',
# '_right_root',
# '__len',
# '_guard',
# '__repr',
# '_version'
# )
def __init__(self, iterable=None, maxlen=None):
if maxlen is not None:
if not isinstance(maxlen, int):
raise TypeError('max_len must be None or int')
if maxlen < 0:
raise ValueError('max_len must be >= zero')
self.__max_len = maxlen
else:
self.__max_len = math.inf
self._left_root = self._right_root = None
self.__len = 0
self._guard = IterationGuard(False)
self.__repr = False
self._version = 0
if iterable is not None:
self.extend(iterable)
def append(self, x):
with self._guard.mute:
if self._right_root is None:
self._left_root = self._right_root = Node(x, None, None, self)
else:
self._right_root.append(x)
self.__len += 1
if self.__len > self.__max_len:
self.popleft()
self._version += 1
def append_left(self, x):
with self._guard.mute:
if self._left_root is None:
self._left_root = self._right_root = Node(x, None, None, self)
else:
self._left_root.append_left(x)
self.__len += 1
if self.__len > self.__max_len:
self.pop()
self._version += 1
appendleft = append_left
def clear(self):
with self._guard.mute:
self._left_root = self._right_root = None
self.__len = 0
self._version += 1
def copy(self):
with self._guard.iter:
version = self._version
result = type(self)(self, self.max_len)
if self._version != version:
raise RuntimeError()
return result
def count(self, x):
with self._guard.iter:
version = self._version
node, count = self._left_root, 0
while node is not None:
count += node.value == x
node = node.right
if self._version != version:
raise RuntimeError()
return count
def extend(self, iterable):
for item in (tuple(self) if iterable is self else iterable):
self.append(item)
def extend_left(self, iterable):
for item in (tuple(self) if iterable is self else iterable):
self.append_left(item)
extendleft = extend_left
def index(self, x, start=None, stop=None):
start = 0 if start is None else self.__get_index(start, False)
stop = len(self) if stop is None else self.__get_index(stop, False)
with self._guard.iter:
version = self._version
node = self._left_root
for index in itertools.count():
if self._version != version:
raise RuntimeError()
if node is None:
raise ValueError('item could not be found')
if node.value == x and start <= index < stop:
break
node = node.right
if self._version != version:
raise RuntimeError()
return index
def insert(self, i, x):
if len(self) == self.max_len:
raise IndexError('cannot grow anymore')
index = self.__get_index(i, False)
if index >= len(self):
self.append(x)
else:
if index < 0:
index = 0
node = self.__get_node(index)
with self._guard.mute:
if node.left is None:
self._left_root = node.left = Node(x, None, node, self)
else:
node.left.right = node.left = Node(
x, node.left, node, self
)
self.__len += 1
self._version += 1
def pop(self):
if self._right_root is None:
raise IndexError('pop from an empty deque')
with self._guard.mute:
value = self._right_root.value
self._right_root = self._right_root.left
if self._right_root is None:
self._left_root = None
else:
self._right_root.right = None
self.__len -= 1
self._version += 1
return value
def popleft(self):
if self._left_root is None:
raise IndexError('pop from an empty deque')
with self._guard.mute:
value = self._left_root.value
self._left_root = self._left_root.right
if self._left_root is None:
self._right_root = None
else:
self._left_root.left = None
self.__len -= 1
self._version += 1
return value
def remove(self, value):
node = self._left_root
version = self._version
while node is not None:
if node.value == value:
if self._version != version:
raise IndexError()
break
else:
node = node.right
else:
raise ValueError('value could not be found')
self.__del_node(node)
def reverse(self):
if self._left_root is not None:
with self._guard.mute:
self._left_root, self._right_root = \
self._right_root, self._left_root
node = self._left_root
while node is not None:
node.left, node.right = node.right, node.left
node = node.right
self._version += 1
def rotate(self, n=1):
size = len(self)
if size > 0:
with self._guard.mute:
for _ in range(n % size):
self.append_left(self.pop())
self._version += 1
# Support for the following methods are left as an exercise for the reader.
@property
def max_len(self):
return None if self.__max_len == math.inf else self.__max_len
maxlen = max_len
def __iter__(self):
return DequeIter(self)
def __getstate__(self):
return (
self.__max_len,
self._left_root,
self._right_root,
self.__len,
self._guard,
self.__repr,
self._version
)
def __setstate__(self, state):
self.__max_len, self._left_root, self._right_root, self.__len, \
self._guard, self.__repr, self._version = state
def __len__(self):
return self.__len
def __reversed__(self):
return ReverseIter(self)
def __copy__(self):
return self.copy()
def __deepcopy__(self, memo):
return type(self)(map(lambda item: copy.deepcopy(item, memo), self))
def __contains__(self, item):
with self._guard.iter:
version = self._version
node = self._left_root
while node is not None:
if node.value == item:
return True
node = node.right
if self._version != version:
raise RuntimeError()
return False
def __getitem__(self, key):
index = self.__get_index(key)
return self.__get_node(index).value
def __setitem__(self, key, value):
index = self.__get_index(key)
node = self.__get_node(index)
node.value = value
def __delitem__(self, key):
index = self.__get_index(key)
node = self.__get_node(index)
with self._guard.mute:
self.__del_node(node)
self._version += 1
def __get_index(self, key, check=True):
if key < 0:
key = len(self) + key
if check and (key < 0 or key >= len(self)):
raise IndexError('index is out of range')
return key
def __get_node(self, index):
with self._guard.iter:
version = self._version
halfway = len(self) >> 1
if index < halfway:
node = self._left_root
for _ in range(index):
node = node.right
return node
node = self._right_root
for _ in range(len(self) - index - 1):
node = node.left
if self._version != version:
raise RuntimeError()
return node
def __del_node(self, node):
with self._guard.mute:
if node.left is None:
self._left_root = node.right
else:
node.left.right = node.right
if node.right is None:
self._right_root = node.left
else:
node.right.left = node.left
self.__len -= 1
self._version += 1
# The following methods are for extra credit.
def __add__(self, other):
if not isinstance(other, type(self)):
raise TypeError(f'other must be of type {type(self).__name__}')
total = self.copy()
total.extend(other)
return total
def __mul__(self, other):
if other < 1:
return type(self)()
product = self.copy()
for _ in range(other - 1):
product.extend(self)
return product
__rmul__ = __mul__
def __imul__(self, other):
if other < 1:
self.clear()
if other > 1:
state = tuple(self)
for _ in range(other - 1):
self.extend(state)
return self
# This should be helpful while trying to debug the code.
def __repr__(self):
if self.__repr:
return '[...]'
self.__repr = True
ml = "" if self.maxlen is None else ", maxlen=" + str(self.maxlen)
representation = f'{type(self).__name__.lower()}({list(self)}{ml})'
self.__repr = False
return representation
# Testing requires a few more methods.
def __lt__(self, other):
for a, b in zip(self, other):
if a < b:
return True
if a > b:
return False
return True
def __le__(self, other):
for a, b in zip(self, other):
if a < b:
return True
if a > b:
return False
return True
def __eq__(self, other):
return type(self) == type(other) and \
len(self) == len(other) and \
all(itertools.starmap(operator.eq, zip(self, other)))
def __ne__(self, other):
return type(self) != type(other) or \
len(self) != len(other) or \
any(itertools.starmap(operator.ne, zip(self, other)))
def __gt__(self, other):
for a, b in zip(self, other):
if a < b:
return False
if a > b:
return True
return False
def __ge__(self, other):
for a, b in zip(self, other):
if a < b:
return False
if a > b:
return True
return False
def __iadd__(self, other):
self.extend(other)
return self
deque = Deque
class Node:
__slots__ = 'value', 'left', 'right', 'parent'
def __init__(self, value, left, right, parent):
self.value = value
self.left = left
self.right = right
self.parent = parent
def __repr__(self):
return f'{type(self).__name__}(' \
f'{self.value!r}, ' \
f'{None if self.left is None else self.left.value!r}, ' \
f'{None if self.right is None else self.right.value!r}, parent)'
def append(self, x):
new_node = type(self)(x, self, None, self.parent)
self.right = new_node
self.parent._right_root = new_node
def append_left(self, x):
new_node = type(self)(x, None, self, self.parent)
self.left = new_node
self.parent._left_root = new_node
def __getstate__(self):
return self.value, self.left, self.right, self.parent
def __setstate__(self, state):
self.value, self.left, self.right, self.parent = state
class DequeIter:
__slots__ = 'parent', 'node', 'guard', 'version'
def __init__(self, parent):
self.parent = parent
self.node = parent._left_root
self.guard = parent._guard.iter
self.version = parent._version
if self.node is not None:
self.guard.__enter__()
def __iter__(self):
return self
def __next__(self):
if self.parent._version != self.version:
raise RuntimeError()
if self.node is None:
raise StopIteration()
node = self.node
self.node = node.right
if self.node is None:
self.guard.__exit__(None, None, None)
return node.value
def __getstate__(self):
return self.parent, self.node, self.guard, self.version
def __setstate__(self, state):
self.parent, self.node, self.guard, self.version = state
class ReverseIter(DequeIter):
__slots__ = ()
def __init__(self, parent):
parent = parent.copy()
parent.reverse()
super().__init__(parent)
class IterationGuard:
__slots__ = '__iter', '__mute'
def __init__(self, debug):
self.__iter = IGAttr('iter', debug)
self.__mute = IGAttr('mute', debug)
self.__iter.other = self.__mute
self.__mute.other = self.__iter
@property
def iter(self):
return self.__iter
@property
def mute(self):
return self.__mute
def __getstate__(self):
return self.__iter, self.__mute
def __setstate__(self, state):
self.__iter, self.__mute = state
class IGAttr:
__slots__ = 'name', 'debug', 'other', 'level'
def __init__(self, name, debug):
self.name = name
self.debug = debug
self.other = None
self.level = 0
def __enter__(self):
if self.other.level > 0 and self.other.name != 'iter':
raise RuntimeError(f'{self.name} is not allowed right now')
self.level += 1
if self.debug:
print(f'{self.name} = {self.level}')
def __exit__(self, exc_type, exc_value, traceback):
self.level -= 1
if self.debug:
print(f'{self.name} = {self.level}')
def __getstate__(self):
return self.name, self.debug, self.other, self.level
def __setstate__(self, state):
self.name, self.debug, self.other, self.level = state
class TestHelp:
@staticmethod
def assertEqual(first, second, msg=None):
assert first == second, msg
@staticmethod
def assertRaises(exception, callable=None, *args, **kwargs):
if callable is None:
return ExceptionChecker(exception)
try:
callable(*args, **kwargs)
except exception:
pass
except:
assert False
else:
assert False
class ExceptionChecker:
def __init__(self, exception):
self.exc = exception
def __enter__(self):
pass
def __exit__(self, exc_type, exc_val, exc_tb):
return isinstance(exc_val, self.exc)
if __name__ == '__main__':
main()