Much of the academic exposition of Dijkstra's Algorithm (such as Wikipedia, or this class page from Darmouth) relies on your priority queue having the ability decrease the key of an entry: the priority queue fills up once at the beginning of the algorithm, and then exclusively shrinks, with one entry being popped at each loop.

However, the implementation of PriorityQueue in the Python's queue module has no available decrease-key method. Neither does Java's standard library. Probably as a result of this, many implementations I see online resort to adding all elements immutably to the priorityQueue as they're seen, resulting in an unnecessarily large heap structure with multiple nodes corresponding to a single vertex. This is asymptotically worse: localized decrease-key/siftdown operations with an ever-shrinking heap size should be faster than replacing those operations with heappushing and heappoping with a potentially much larger heap.

Because I haven't seen many very clear implementations of the IndexedHeap structure required for a Decrease-Key operation, I wrote one myself, with the goal of being able to write the following straight-out-of-a-textbook pseudocode implementation of Dijkstra's Algorithm:

from collections import defaultdict
from math import inf
from indexedheap import IndexedHeap

class Graph:
    """ A nonnegatively weighted, directed multigraph.

    Stored as a list of outgoing edges for each vertex.

    >>> g = Graph()
    >>> for source, dest, weight in [ \
            ('a', 'b', 7), ('a', 'b', 8), ('a', 'c', 9), ('a', 'f', 14), \
            ('b', 'c', 10), ('b', 'd', 15), ('c', 'd', 11), ('c', 'f', 2), \
            ('d', 'e', 6), ('e', 'f', 9), ('f', 'g', 100), ('g', 'b', 100), \
            ('a', 'a', 100) \
        ]: \
            g.add_edge(source, dest, weight)
    >>> g.distance_and_shortest_path('a', 'f')
    (11, ['a', 'c', 'f'])
    >>> g.distance_and_shortest_path('b', 'e')
    (21, ['b', 'd', 'e'])
    >>> g.distance_and_shortest_path('a', 'a')
    (0, ['a'])
    >>> g.distance_and_shortest_path('f', 'a')
    (inf, None)
    >>> g.distance_and_shortest_path('garbage', 'junk')
    (inf, None)

    def __init__(self):
        self.vertices = set()
        self.edges = defaultdict(list)

    def add_edge(self, source, destination, weight):
        """ Add a weighted edge from source to destination.
        if weight < 0:
            raise ValueError("Edge weights cannot be negative.")
        self.vertices |= {destination, source}
        self.edges[source].append((destination, weight))

    def distance_and_shortest_path(self, source, destination):
        """Find the lightest-weighted path from source to destination.

        We use Dijkstra's algorithm with an indexed heap.

        :return: A 2-tuple (d, [v0, v1, ..., vn]),
            where v0==source and vn==destination.
        if not {source, destination} <= self.vertices:
            return inf, None

        # For each vertex v, store the weight of the shortest path to found
        # so far to v, along with v's predecessor in that path.
        distance = {v: inf for v in self.vertices}
        distance[source] = 0
        predecessor = {}

        # A priority queue of exactly the unexplored vertices,
        h = IndexedHeap((distance[v], v) for v in self.vertices)

        # Explore until all vertices closer to source have been exhausted,
        # at which point we will have already found the shortest path (if any) to destination.
        while h.peek() != destination:
            v = h.pop()
            v_dist = distance[v]
            for neighbor, edge_weight in self.edges[v]:
                # We've found a new path to neighbor. If the distance along
                # this new path (through v) is better than previously found,
                # then "relax" the stored distance to that along the new path.
                alt_dist = v_dist + edge_weight
                if alt_dist < distance[neighbor]:
                    distance[neighbor] = alt_dist
                    predecessor[neighbor] = v
                    h.change_weight(neighbor, alt_dist)

        if distance[destination] == inf:
            # No path was found.
            return inf, None

        # Trace back the predecessors to get the path.
        path = [destination]
        while path[-1] != source:
        return distance[destination], path

if __name__ == "__main__":
    import doctest

What follows is my implementation of the IndexedHeap.

import pyheapq
from collections import defaultdict

class IndexedHeap():
    """A priority queue with the ability to modify existing priority.

    >>> h = IndexedHeap(['1A', '0B', '5C', '2M'])
    >>> h.pop()
    >>> h.peek()
    >>> h.change_weight('M', '6')
    >>> h.pushpop('W', '7')
    >>> h.poppush('R', '8')
    >>> [h.pop() for _ in range(len(h))]
    ['M', 'W', 'R']

    def __init__(self, iterable=()):
        self.heap = _IndexedWeightList(map(tuple, iterable))

    def __len__(self):
        return len(self.heap)

    def __contains__(self, item):
        return (item in self.heap)

    def push(self, item, weight):
        pyheapq.heappush(self.heap, (weight, item))

    def pop(self):
        weight, item = pyheapq.heappop(self.heap)
        return item

    def peek(self):
        weight, item = self.heap[0]
        return item

    def pushpop(self, item, weight):
        # First push, then pop.
        weight, item2 = pyheapq.heappushpop(self.heap, (weight, item))
        return item2

    def poppush(self, item, weight):
        # First pop, then push.
        weight, item2 = pyheapq.heapreplace(self.heap, (weight, item))
        return item2

    def change_weight(self, item, weight):
        i = self.heap.index(item)
        old_weight, item = self.heap[i]
        self.heap[i] = weight, item
        if weight < old_weight:
            pyheapq.siftdown(self.heap, 0, self.heap.index(item))
        elif weight > old_weight:
            pyheapq.siftup(self.heap, self.heap.index(item))

    def __bool__(self):
        return bool(self.heap)

class _IndexedWeightList(list):
    """A list of (weight, item) pairs, along with the indices of each "item".

    We maintain an auxiliary dict consisting of, for each item, the set of
    indices of that item. Each set will typically have just one index, but
    we do not enforce this because the heapq module updates multiple entries
    at the same time. You could say that this class has all of the
    functionality of priority queue, but without the prioritization.

    >>> arr = _IndexedWeightList(['1A', '0B', '5C', '2M'])
    >>> arr
    _IndexedWeightList(['1A', '0B', '5C', '2M'])
    >>> arr[2]
    >>> arr[0], arr[3] = arr[3], arr[0]
    >>> arr
    _IndexedWeightList(['2M', '0B', '5C', '1A'])
    >>> arr.append('6D')
    >>> arr
    _IndexedWeightList(['2M', '0B', '5C', '1A', '6D'])
    >>> [arr.index(x) for x in 'ABCDM']
    [3, 1, 2, 4, 0]
    >>> arr.remove('B')
    Traceback (most recent call last):
    TypeError: 'NoneType' object is not callable
    >>> pyheapq.heapify(arr)
    >>> arr.index('B')

    def __init__(self, iterable=()):
        self._index = defaultdict(set)
        for i, (weight, item) in enumerate(super().__iter__()):

    def __setitem__(self, i, pair):
        weight, item = pair
        old_weight, old_item = super().__getitem__(i)
        super().__setitem__(i, pair)

    def index(self, item, start=..., stop=...) -> int:
        only, = self._index[item]
        return only

    def __contains__(self, item):
        return bool(self._index[item])

    def append(self, pair):
        weight, item = pair
        self._index[item].add(len(self) - 1)

    def extend(self, iterable):
        for pair in iterable:

    def pop(self, i=...):
        weight, item = super().pop()
        return (weight, item)

    def __repr__(self):
        return '{}({})'.format(self.__class__.__qualname__, str(list(self)))

    insert = None
    remove = None
    __delitem__ = None
    sort = None

if __name__ == "__main__":
    import doctest


Above, I would rely heavily on Python's standard heapq module, but the C implementation of heapq didn't work with with my overriding of __setattr__ subscripting, so I copied its pure-python implementation into a pyheapq.py:

This is all directly copied from heapq.py in the python standard library.
This local copy ensures that we use this pure Python implementation so that
subscripting can be overridden to maintain an index. Underscores in protected
methods have been removed.

def heappush(heap, item):
    """Push item onto heap, maintaining the heap invariant."""
    siftdown(heap, 0, len(heap) - 1)

The rest of the above file is here.

My question: Was my overriding of __getattr__ in the _IndexedWeightList class an appropriate way of re-using all the existing heapq code?

  • \$\begingroup\$ I assume that the doctests work for you? Testing under Windows with Python 3.6.6 I find that _IndexedWeightList fails the last test. Apparently pyheapq.heapify is bypassing the overridden __setitem__. Do you have any idea why? \$\endgroup\$ May 13 '19 at 8:16
  • 1
    \$\begingroup\$ Ah. "The rest of the above file is here" but it's necessary to remove the C imports from the end of that file. \$\endgroup\$ May 13 '19 at 8:23
  • \$\begingroup\$ As a possible point of interest: github.com/coderodde/SearchHeapBenchmark/tree/master/src/main/… \$\endgroup\$
    – coderodde
    May 13 '19 at 10:09
  • \$\begingroup\$ Can you provide a link to a possible GitHub project/Gist containing all the source code needed to run your program? \$\endgroup\$
    – coderodde
    May 13 '19 at 11:34

Firstly, it's always good to see doctests.

I'm not entirely convinced by the name IndexedHeap. The API it exposes has nothing to do with indexes.

    def __bool__(self):
        return bool(self.heap)

It took a bit of thought to figure out what this is about, but once I got there I appreciate the elegant implementation.

    def pop(self, i=...):
        weight, item = super().pop()
        return (weight, item)

Probably my biggest criticism: in some use cases this will leak memory by leaving self._index full of empty sets. I'd prefer to see it delete the key when the last index is removed.

Overall, though, the index is quite neat. I must say that it's very nice the way the heap will handle repeated values with different keys but won't allow change_weight of a repeated value. Perhaps a case could be made for giving change_weight an optional current_weight argument to distinguish between repeated values, but YAGNI.

Was my overriding of __getattr__ in the _IndexedWeightList class an appropriate way of re-using all the existing heapq code?

I assume you actually mean __setitem__ rather than __getattr__.

IMO given that you not only have to copy heapq.py but edit it to make it suitable for your usage, you've lost the advantage of code reuse. It becomes painful to track upstream changes and merge in bugfixes. You might as well go the whole hog and refactor it to suit your needs. It makes more sense to me to add index as an argument to the forked heapq methods1 and essentially inline _IndexedWeightList away. That has the added advantage that it's compatible with doing the same thing to a fork of the C implementation.

1 Except heapify, because you might as well just rebuild the index after calling that.


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.