Reimplementation of Python's deque

Question

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()

Answer 1

Just reviewing the AtomicMeta class.

1. Review

1. There's no docstring. What is the class supposed to do?

2. There are two mechanisms for initializing the __mutex attribute: one in the atomic_new method, and another in the wrap method. Why two mechanisms? Is one of them somehow unreliable? But if it is, then why not just use the other? I don't understand this, so have I missed a requirement?

3. AtomicMeta does not support classes that have __slots__:

   >>> class Test1(metaclass=AtomicMeta):
   ... __slots__ = ['slot']
   ...
   >>> Test1()
   Traceback (most recent call last):
    File "<stdin>", line 1, in <module>
    File "cr178957.py", line 31, in atomic_new
    instance.__mutex = threading.RLock()
   AttributeError: 'Class1' object has no attribute '_AtomicMeta__mutex'
   

   To fix this, the __new__ method needs:

   if '__slots__' in class_dict:
    class_dict['__slots__'] += ('_AtomicMeta__mutex',)
   

4. AtomicMeta does not support classes that have a __new__ method:

   >>> class Test2(metaclass=AtomicMeta):
   ... def __new__(cls, value):
   ... instance = super().__new__(cls)
   ... instance.attr = value
   ... return instance
   ...
   >>> Test2('value').attr
   Traceback (most recent call last):
    File "<stdin>", line 1, in <module>
   AttributeError: 'Test2' object has no attribute 'attr'
   

   The simplest approach for supporting this would seem to be to drop the atomic_new method (relying on the wrap method to initialize the __mutex attribute, as discussed above), and ensure that the __new__ method does not get wrapped:

   if key != '__new__' and callable(value):
    class_dict[key] = mcs.wrap(value)
   

5. The __init__ method also does not need to be wrapped, since at the point where __init__ is called, only one thread has access to the object. So this test can become:

   if key not in ('__init__', '__new__') and callable(value):
    class_dict[key] = mcs.wrap(value)
   

6. Since wrap is called from only one place, it would be simpler to inline it at the point of use.

2. Revised code

import functools
import threading
class AtomicMeta(type):
 "Metaclass that serializes access to the methods of each instance."
 def __new__(mcs, name, bases, class_dict):
 for key, value in class_dict.items():
 if key not in ('__init__', '__new__') and callable(value):
 @functools.wraps(value)
 def atomic(self, *args, **kwargs):
 try:
 mutex = self._AtomicMeta__mutex
 except AttributeError:
 mutex = self._AtomicMeta__mutex = threading.RLock()
 with mutex:
 return value(self, *args, **kwargs)
 class_dict[key] = atomic
 if '__slots__' in class_dict:
 class_dict['__slots__'] += ['_AtomicMeta__mutex']
 return super().__new__(mcs, name, bases, class_dict)

Answer 2

The biggest problem I see with this whole thing, and the reason your code hasn't gotten much attention, is the lack of comments. In particular comments about why you're doing something a certain way, or what the purpose (goal) of a section of code is. This makes it cumbersome for someone looking over the code to give the kind of high-level analysis you're asking for. Putting some docstrings in (even though the interface is the same as a standard library) would also help out people like me who sometimes forget the difference between append and extend.

• What's the reason for maintaining version in the deque itself? I don't undestand the Atomic stuff, I guess that's supposed to keep it threadsafe, but shouldn't that take care of things without needing version? You really need some comments here.
• Is there any way to test if the atomicity actualy works?
• Is an unexplained RuntimeError the best exception to raise if you detect a failure of atomicity using version?
• The exceptions for __init__() refer to max_len but the named parameter is maxlen
• Using a temporary tuple when duplicating the dequeue in extend() is quick to write but inefficient. Shouldn't it be possible to keep track using pointers?
• In index(), I'm not sure if re-using start and stop that way is easy enough to understand. I don't think it hurts anything to create a separate variable, and name them in keeping with their actual content
• Since the start <= index < stop test will almost always be quicker than the equality test, shouldn't it be first?
• Does the reference implementation actually require a IndexError exception if you try to add to a full deque? That seems counterintuitive.
• Isn't the last half of insert the same as append_left?
• In the functions that come in left/right pairs, isn't there some way to combine them so the code is not duplicated?
• I'm pretty sure there's a better implementation of rotate() that only requires patching four pointers.
• Is the "exercise for the reader" comment supposed to be there?

Answer 3

I guess I'd add an alias for "the other side" everywhere a method name is decorated left or right, the way the library grew a bisect_right().
For reverse(), you could swap accessors (including, depending on implementation details, iterators) instead of references - better yet, create a class with swapped accessors and mutate the object on reverse().
What about __str__()?