[Python-checkins] python/nondist/sandbox/collections pairing_heap.py, NONE, 1.1 pairing_heap_test.py, NONE, 1.1

Fri Apr 2 02:46:05 EST 2004

Update of /cvsroot/python/python/nondist/sandbox/collections
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv21444
Added Files:
	pairing_heap.py pairing_heap_test.py 
Log Message:
Add Dan Stutzbach's first pass at paired heaps
(after modifying his test suite to use unittests).
The pure python version is being used to work out API issues, develop
thorough tests, and to tease out implementation issues.
Ultimately, this will be added as collections.heap
--- NEW FILE: pairing_heap.py ---
#!/usr/bin/python
"""
Note: this implementation is not currently thread-safe.
The code in this file implements a pairing heap. By default it
operates as a min-heap, but it can also work as a max-heap via a flag
passed to the heap's __init__ routing.
The important papers on pairing heaps are these:
[1] Michael L. Fredman, Robert Sedjewick, Daniel D. Sleator, and
Robert E. Tarjan. The Pairing Heap: A New Form of Self-Adjusting
Heap. Algorithmica 1:111-129, 1986.
[2] John T. Stasko and Jeffrey Scott Vitter. Pairing heaps:
experiments and analysis. Communications of the ACM, Volume 30, Issue
3, Pages 234-249, 1987.
[3] Michael L. Fredman. On the efficiency of pairing heaps and
related data structures. Journal of the ACM, Volume 46, Issue 4,
Pages 473-501, 1999.
There are actually several varieties of pairing heaps. Five are
described in [1], and two more are described in [2]. This
implementation is the basic pairing heaps described in [1], also known
as the twopass variant. There isn't much published evidence that any
of the variants are significantly better than the others.
The twopass pairing heap works as follows.
A pairing heap is a tree that maintains the heap property: each node
has a lower value than all its children. The pairing heap tree need
not be binary nor balanced. The good amortized efficiency of the
pairing heap results from clever algorithms for .adjust_key() and
.delete().
The basic operation in the pairing heap is the comparison-link. Two
nodes are compared, and the largest is linked as a child of the
smaller. To represent the tree, each node maintains a pointer to the
leftmost child and to its right sibling. Each node also maintains a
pointer to its parent. When a comparison-link is performed, the new
child becomes the leftmost child.
[1] suggests a memory optimization where only the right-most sibling
stores a pointer to the parent, and each node maintains a single bit
indicating if it is that rightmost sibling. This theoretically saves
the size of a pointer in each node, but only if we can store a bit for
free. However, storing a bit takes just as much space as storing a
pointer, due to padding. So, we bite the bullet and store the
pointer. Also, [3] shows that the amortized cost of .adjust_key()
is at least Omega(log log n) if the parent pointer is *not* stored.
The amortized cost may be better using the parent pointer.
For .adjust_key(), the topmost element is removed and set aside as
the return value. A left-to-right pass is made over its children,
comparison-linking them pairwise. For example, if there are 6
children labeled A, B, C, D, E, and F, A and B will be
comparison-linked, as will C and D, and E and F. Next, the rightmost
node is repeatedly comparison-linked with its sibling until there is
only one node left. This node is the new root.
For .delete(), first the node is cut from its parent. This results in
a tree rooted at the node being deleted. The .adjust_key()
procedure is invoked on the node, then the resulting tree is
comparison-linked with the original tree.
See [1] for more details. It also includes figures which are helpful
for understanding.
"""
class WrongHeap (Exception):
 """The user attempted to .delete() or .adjust() from the wrong heap.
 This can also occur if the user attempts to delete or adjust a
 node that has already been removed from the heap.
 """
 def __str__(self):
 return 'WrongHeap'
class Underflow (Exception):
 "The user attempted to .peek() or extract() from an empty heap."
 def __str__(self):
 return 'Underflow'
class IncompatibleHeaps (Exception):
 "The user attempted to meld heaps that have different compare operations."
 def __str__(self):
 return 'IncompatibleHeaps'
class InternalError (Exception):
 "This should never get raised in practice."
 def __str__(self):
 return 'InternalError'
class WrongAdjustKeyDirection (Exception):
 "The user attempted to adjust a key in the wrong direction."
 def __str__(self):
 return 'WrongAdjustKeyDirection'
class heap_node:
 """This class is used internally to implement the heap.
 Users will have be able to see references to this data structure,
 as this is neccessary so they can pass the reference to the
 .adjust_key() and .delete() methods. However, the data structure
 should be considered opaque. The only public methods are .value()
 and .values().
 Users should not instantiate heap_node on their own.
 In the C implementation, all of this will be enforced.
 """
 __slots__ = ('_item', '_child', '_sibling', '_parents')
 def __init__(self, item):
 self._item = item
 self._clean()
 def value(self):
 "Return the value associated with this node."
 return self._item
 def _clean(self):
 self._child = None
 self._sibling = None
 self._parent = None
 def _add_child(self, node):
 node._parent = self
 node._sibling = self._child
 self._child = node
 def _cut(self):
 "Remove this node and its children from this node's parent."
 n = self._parent
 if id(n._child) == id(self):
 n._child = self._sibling
 else:
 n = n._child
 try:
 while id(n._sibling) != id(self):
 n = n._sibling
 n._sibling = self._sibling
 except AttributeError:
 raise InternalError
 self._parent = None
 self._sibling = None
 def _extract(self, heap):
 "Link this node's children as a prelude to this node's removal"
 # Make the left-to-right pairing pass
 children = []
 n = self._child
 while n and n._sibling:
 next = n._sibling._sibling
 pair = n._sibling
 n._parent = None
 n._sibling = None
 pair._parent = None
 pair._sibling = None
 children.append(heap._link(n, pair))
 n = next
 if n:
 children.append(n)
 if not children:
 return None
 # Repeated comparison-link to the rightmost node
 root = children[-1]
 for i in range(len(children)-2,-1,-1):
 root = heap._link(root, children[i])
 root._parent = None
 # Return the new root
 return root
 def values(self):
 "Return an unsorted sequence of the values of this part of the tree."
 rv = [self._item,]
 if self._child:
 rv += self._child.values()
 if self._sibling:
 rv += self._sibling.values()
 return rv
class pairing_heap:
 "Implements a min-heap using the pairing-heap algorithm."
 def __init__(self, values=None, cmpfunc=None, key=None, reverse=False):
 self._cmpfunc = cmpfunc
 self._key = key
 self._reverse = reverse
 self._root = None
 self._len = 0
 if values is not None:
 for i in values:
 self.insert(i)
 def empty(self):
 "True if there is nothing in the heap."
 return self._root is None
 def peek(self):
 "Return, but do not remove, the top of the heap."
 if self._root:
 return self._root._item
 else:
 raise Underflow
 def __len__(self):
 "Return the number of items in the heap."
 return self._len
 def _cmp(self, item1, item2):
 "Compare two items using the heap's comparison functions."
 if self._key:
 item1 = getattr(item1, self._key)
 item2 = getattr(item2, self._key)
 if self._cmpfunc:
 c = self._cmpfunc(item1, item2)
 else:
 c = cmp(item1, item2)
 if self._reverse:
 c = -c
 return c
 def _link(self, node1, node2):
 "Perform the comparison-link operation."
 if not node1: return node2
 if not node2: return node1
 c = self._cmp(node1._item, node2._item)
 if c > 0:
 node2._add_child (node1)
 return node2
 else:
 node1._add_child (node2)
 return node1
 def insert(self, item):
 "Insert an item into the heap."
 node = heap_node(item)
 self._root = self._link(node, self._root)
 self._len += 1
 return node
 def meld(self, other):
 """Merge another heap into this one.
 other will be an empty heap after completion
 """
 if not isinstance(other, pairing_heap):
 raise TypeError
 # Check that these heaps are compatible
 if self._cmpfunc != other._cmpfunc or self._reverse != other._reverse \
 or self._key != other._key:
 raise IncompatibleHeaps
 self._root = self._link(other._root, self._root)
 self._len += other._len
 # Destroy other
 other._len = 0
 other._root = None
 def adjust_key(self, node, item):
 """Adjust the value of node to item.
 This must be decrease for a min-heap or an increase for a max-heap.
 If node is not part of this heap, both heaps may become gibberish.
 """
 self._check_heap_node (node)
 if self._cmp(node._item, item) < 0:
 raise WrongAdjustKeyDirection
 node._item = item
 if id(self._root) == id(node):
 return
 # Cut this node out of the tree
 node._cut()
 # Link it with the root
 self._root = self._link(node, self._root)
 def delete(self, node):
 """Delete a node from the middle of the heap.
 If node is not part of this heap, both heaps may become gibberish.
 """
 self._check_heap_node (node)
 if id(self._root) == id(node):
 self.extract()
 return
 node._cut()
 self._root = self._link (node._extract(self), self._root)
 self._len -= 1
 node._clean()
 return node._item
 def extract(self, n=0):
 "Extract the top value of the heap."
 if n > 0:
 return [self.extract() for i in xrange(n)]
 if not self._root:
 raise Underflow
 old_root = self._root
 self._root = self._root._extract(self)
 self._len -= 1
 old_root._clean()
 return old_root._item
 def extract_all(self):
 "Empty the heap into a sorted list of all the values"
 return [self.extract() for i in xrange(self._len)]
 def values(self):
 "Return an unsorted list of all the values in the heap"
 return self._root.values()
 def _check_heap_node(self, node):
 """Raise an error if node will cause problems.
 This function is a simple check to catch certain user errors.
 It will raise TypeError if node isn't a heap_node object at
 all. It *may* raise WrongHeap if node isn't part of this heap.
 It will return if node is part of this heap.
 However, no guarantees are made about the return value if node
 is a heap_node, but part of a different heap.
 """
 if not isinstance(node, heap_node):
 raise TypeError
 if self._root == node:
 return
 if not node._parent:
 raise WrongHeap
 return
--- NEW FILE: pairing_heap_test.py ---
#!/usr/bin/python
from pairing_heap import *
import random
import unittest
datasize = 256
class Holder:
 def __init__(self, item):
 self.xyzzy = item
def check_invariant(heap, node=None):
 if node == None:
 node = heap._root
 if node == None:
 return True
 n2 = node._child
 while n2:
 if heap._cmp(node._item, n2._item) >= 0:
 return False
 check_invariant(heap, n2)
 n2 = n2._sibling
 return True
# Uncomment to speed things up when doing code-coverage analysis
##def check_invariant(heap, node=None):
## return True
class TestPairingHeap(unittest.TestCase):
 def setUp(self):
 self.data = [random.random() for i in xrange(datasize)]
 self.data_sorted = sorted(self.data)
 def test_random(self):
 # Push random numbers and pop them off, verifying all's OK.
 heap = pairing_heap()
 self.assert_(check_invariant(heap))
 for item in self.data:
 heap.insert(item)
 self.assert_(check_invariant(heap))
 results = []
 while not heap.empty():
 item = heap.extract()
 self.assert_(check_invariant(heap))
 results.append(item)
 self.assertEqual(self.data_sorted, results)
 def test_naive_nbest(self):
 heap = pairing_heap()
 for item in self.data:
 heap.insert(item)
 if len(heap) > 10:
 heap.extract()
 result = heap.extract_all()
 self.assertEqual(result, self.data_sorted[-10:])
 def test_underflow(self):
 heap = pairing_heap()
 self.assertRaises(Underflow, heap.extract)
 self.assertRaises(Underflow, heap.peek)
 def test_return_unsorted(self):
 heap = pairing_heap()
 for item in self.data:
 heap.insert(item)
 self.assertEqual(set(heap.values()), set(self.data))
 def test_implicit_creation_and_peek(self):
 heap = pairing_heap(self.data)
 self.assert_(check_invariant(heap))
 self.assertEqual(heap.peek(), self.data_sorted[0])
 self.assert_(check_invariant(heap))
 result = heap.extract_all()
 self.assertEqual(result, self.data_sorted)
 def test_extract_with_arg(self):
 heap = pairing_heap(self.data)
 low10 = heap.extract(10)
 self.assert_(check_invariant(heap))
 self.assertEqual(low10, self.data_sorted[:10])
 def test_deleting_from_the_middle(self):
 heap = pairing_heap()
 nodes = [heap.insert(item) for item in self.data]
 data2 = self.data[:]
 count = 0
 for i in xrange(datasize-1,-1,-5):
 heap.delete(nodes[i])
 del nodes[i]
 self.assert_(check_invariant(heap))
 count += 1
 self.assertEqual(datasize-count, len(heap))
 del data2[i]
 result = heap.extract_all()
 data2_sorted = data2[:]
 data2_sorted.sort()
 self.assertEqual(data2_sorted, result)
 def test_deleting_everything_in_random_order(self):
 heap = pairing_heap()
 nodes = [heap.insert(item) for item in self.data]
 for i in xrange(datasize):
 t = random.randrange(len(heap))
 heap.delete(nodes[t])
 del nodes[t]
 self.assert_(check_invariant(heap))
 self.assert_(heap.empty())
 def test_adjust_key(self):
 heap = pairing_heap()
 nodes = [heap.insert(item) for item in self.data]
 data2 = []
 for i in xrange(datasize):
 data2.append(min(self.data[i], random.random()))
 idx = range(datasize)
 random.shuffle(idx)
 for i in idx:
 heap.adjust_key(nodes[i], data2[i])
 self.assert_(check_invariant(heap))
 self.assertEqual(sorted(data2), heap.extract_all())
 def test_build_tree_by_melding(self):
 heap = pairing_heap()
 i = 0
 while i < datasize:
 j = max(random.randrange(min(datasize - i, 5)), 1)
 heap2 = pairing_heap(self.data[i:i+j])
 i += j
 heap.meld(heap2)
 self.assert_(check_invariant(heap))
 self.assertEqual(self.data_sorted, heap.extract_all())
 def test_cmpfunc(self):
 reverse = lambda x, y: -cmp(x,y)
 heap = pairing_heap(self.data, cmpfunc=reverse)
 self.assertEqual(sorted(self.data, reverse=True), heap.extract_all())
 def test_reverse(self):
 heap = pairing_heap(self.data, reverse=True)
 self.assertEqual(sorted(self.data, reverse=True), heap.extract_all())
 def test_cmpfunc_with_reverse(self):
 reverse = lambda x, y: -cmp(x,y)
 heap = pairing_heap(self.data, cmpfunc=reverse, reverse=True)
 self.assertEqual(self.data_sorted, heap.extract_all())
 def test_key(self):
 holders = [Holder(item) for item in self.data]
 heap = pairing_heap(holders, key='xyzzy')
 results = heap.extract_all()
 self.assertEqual(self.data_sorted,[item.xyzzy for item in results])
 def test_extract_values_from_nodes(self):
 heap = pairing_heap()
 nodes = [heap.insert(datum) for datum in self.data]
 for i in xrange(len(nodes)):
 self.assertEqual(nodes[i].value(), self.data[i])
 self.assert_(check_invariant(heap))
 def test_meld_with_incompatible_heap(self):
 heap1 = pairing_heap()
 reverse = lambda x, y: -cmp(x,y)
 self.assertRaises(IncompatibleHeaps, heap1.meld, pairing_heap(cmpfunc = reverse))
 self.assertRaises(IncompatibleHeaps, heap1.meld, pairing_heap(reverse = True))
 self.assertRaises(IncompatibleHeaps, heap1.meld, pairing_heap(key = 'xyzzy'))
 def test_adj_key_wrong_direction(self):
 heap = pairing_heap ()
 nodes = [heap.insert(datum) for datum in range(5)]
 self.assertRaises(WrongAdjustKeyDirection, heap.adjust_key, nodes[4], 100)
 def test_adjust_key_on_existing_root(self):
 heap = pairing_heap ()
 data = self.data
 nodes = [heap.insert(datum) for datum in data]
 low = 0
 for i in xrange(1, len(data)):
 if data[i] == data[low]:
 low_list.append (i)
 if data[i] < data[low]:
 low = i
 low_list = [i]
 for i in low_list:
 heap.adjust_key(nodes[i], data[i] - random.random())
 def test_bogus_node_parameter(self):
 heap = pairing_heap ()
 nodes = [heap.insert(datum) for datum in self.data]
 self.assertRaises(TypeError, heap.adjust_key, "muhahahah!", 5)
 self.assert_(check_invariant(heap))
 self.assertRaises(TypeError, heap.delete, "muhahahah!")
 def test_semibogus_node_parameter(self):
 holders = [Holder(item) for item in self.data]
 heap = pairing_heap(key='xyzzy')
 nodes = [heap.insert(datum) for datum in holders]
 for i in xrange(len(holders)):
 holders[i].node = nodes[i]
 top = heap.extract ()
 self.assertRaises(WrongHeap, heap.delete, top.node)
 self.assert_(check_invariant(heap))
if __name__ == '__main__':
 suite = unittest.TestSuite()
 suite.addTest(unittest.makeSuite(TestPairingHeap))
 unittest.TextTestRunner(verbosity=2).run(suite)