Linked Lists, Stacks, and Queues Questions

The idea in plain words

A deque (double-ended queue, pronounced "deck") lets you add and remove at both the front and the back. On a circular array (a fixed array whose end wraps around to its start) you keep an index head for the front element and the count size. Element number i, counting from the front, lives at slot (head + i) % capacity.

• Add at the back: write at (head + size) % capacity.
• Add at the front: move head one step left, wrapping from slot 0 to the last slot, then write there.
• Remove at the front: read head, move head one step right.
• Remove at the back: read (head + size - 1) % capacity, no index moves.

None of these shifts any element. The only expensive event is resizing.

Growing and shrinking

• Grow: when an add finds size == capacity, allocate an array twice as large and copy the elements in logical order so that the front lands at slot 0 and head becomes 0. Logical order means front to back as the deque sees it, which is not always the order of the slots in memory: after add_last(1), add_last(2), add_first(0) and add_first(-1) on a capacity-4 array the front element sits in slot 2, and the slots read [1, 2, -1, 0] while the deque reads -1, 0, 1, 2. To unroll the ring is to copy it out in logical order into a fresh array that starts at slot 0. Copying the raw array without unrolling would break the order, because the stored data may wrap around.
• Shrink: when a removal leaves size <= capacity / 4, copy into an array half as large (never below a minimum of 4).
• Why double: growth happens when the size is 4, then 8, then 16, and so on, and each growth copies exactly that many elements. Those copies add up to 4 + 8 + ... + s, a geometric sum (each term is double the last, so the sum is less than twice its largest term). The largest term s is at most half the final capacity, and the final capacity is below 2n, so for n adds the total copying is below 2n elements and the cost per add is amortized O(1). Amortized means that an individual call (the one that triggers a resize) can cost O(n), but the average over any long sequence of calls is constant.
• Why shrink at a quarter, not a half: after a shrink at size = C / 4 (C is the old capacity), the new array has capacity C / 2 and is half full. It shrinks again only when the size falls to C / 8, so C / 8 further removals are needed, and it grows again only when an add finds it full at size C / 2, so C / 4 further adds fill it and the next add grows it. The shrink demonstration in the output shows both counts for C = 64: 8 removals, and 17 adds (16 to fill the array, then the one that grows it). If you shrank at one half instead, the new array would be exactly full and the next add would force a grow, then the next removal a shrink, copying on every call. The measured comparison is in the output below.

The code

import random
from collections import deque


class CircularDeque:
    """Deque on a circular array. head is the index of the front element;
    element i (0 = front) lives at slot (head + i) % capacity."""
    MIN_CAPACITY = 4

    def __init__(self, shrink_at=0.25):
        self._slots = [None] * self.MIN_CAPACITY
        self._head = 0
        self._size = 0
        self._shrink_at = shrink_at
        self.copies = 0                    # elements moved by resizes, for the cost check

    def _resize(self, new_capacity):
        new_slots = [None] * new_capacity
        for i in range(self._size):        # unroll the ring so the front lands at slot 0
            new_slots[i] = self._slots[(self._head + i) % len(self._slots)]
        self.copies += self._size
        self._slots, self._head = new_slots, 0

    def add_last(self, value):
        if self._size == len(self._slots):
            self._resize(2 * len(self._slots))
        self._slots[(self._head + self._size) % len(self._slots)] = value
        self._size += 1

    def add_first(self, value):
        if self._size == len(self._slots):
            self._resize(2 * len(self._slots))
        self._head = (self._head - 1) % len(self._slots)   # step left, wrapping to the end
        self._slots[self._head] = value
        self._size += 1

    def _shrink_if_sparse(self):
        cap = len(self._slots)
        if cap > self.MIN_CAPACITY and self._size <= cap * self._shrink_at:
            self._resize(max(self.MIN_CAPACITY, cap // 2))

    def remove_first(self):
        if self._size == 0:
            raise IndexError("remove from empty deque")
        value = self._slots[self._head]
        self._slots[self._head] = None                      # do not keep a stale reference
        self._head = (self._head + 1) % len(self._slots)
        self._size -= 1
        self._shrink_if_sparse()
        return value

    def remove_last(self):
        if self._size == 0:
            raise IndexError("remove from empty deque")
        idx = (self._head + self._size - 1) % len(self._slots)
        value = self._slots[idx]
        self._slots[idx] = None
        self._size -= 1
        self._shrink_if_sparse()
        return value

    def peek_first(self):
        if self._size == 0:
            raise IndexError("peek at empty deque")
        return self._slots[self._head]

    def peek_last(self):
        if self._size == 0:
            raise IndexError("peek at empty deque")
        return self._slots[(self._head + self._size - 1) % len(self._slots)]

    def __len__(self):
        return self._size

    def logical(self):
        """The elements in logical order: front first, whatever slots they sit in."""
        return [self._slots[(self._head + i) % len(self._slots)] for i in range(self._size)]

    def state(self):
        return (f"cap={len(self._slots)} head={self._head} size={self._size} "
                f"slots={self._slots} front-to-back={self.logical()}")


d = CircularDeque()
steps = [
    ("add_last(1)", lambda: d.add_last(1)),
    ("add_last(2)", lambda: d.add_last(2)),
    ("add_first(0)", lambda: d.add_first(0)),
    ("add_first(-1)", lambda: d.add_first(-1)),
    ("add_last(3)", lambda: d.add_last(3)),
    ("remove_first()", lambda: d.remove_first()),
    ("remove_last()", lambda: d.remove_last()),
]
for label, step in steps:
    result = step()
    print(f"{label:15} returned {result}")
    print(f"    {d.state()}")

random.seed(48)
for shrink_at in (0.25, 0.5):
    mine, ref = CircularDeque(shrink_at), deque()
    for _ in range(5000):
        r = random.random()
        if ref and r < 0.4:
            assert mine.remove_first() == ref.popleft()
        elif ref and r < 0.8:
            assert mine.remove_last() == ref.pop()
        elif r < 0.9:
            mine.add_first(len(ref)); ref.appendleft(len(ref))
        else:
            mine.add_last(len(ref)); ref.append(len(ref))
        assert len(mine) == len(ref)
        assert len(mine._slots) >= CircularDeque.MIN_CAPACITY   # shrinking stops at the minimum
        if ref:
            assert mine.peek_first() == ref[0] and mine.peek_last() == ref[-1]
print("5000 random operations agree with collections.deque, for both shrink thresholds, and the capacity never fell below 4")

grow = CircularDeque()
for i in range(1000):
    grow.add_last(i)
print(f"1000 add_last calls: {grow.copies} elements copied by growth (under 2 per call)")

def just_shrunk():
    t = CircularDeque()
    for i in range(33):
        t.add_last(i)                 # 33 items force capacity 64
    while len(t._slots) == 64:
        t.remove_last()               # the removal that reaches size 16 shrinks to 32
    return t


t = just_shrunk()
print(f"quarter rule: capacity 64 shrank to {len(t._slots)} when the size reached {len(t)}, so the new array is half full")
removals = 0
while len(t._slots) == 32:
    t.remove_last()
    removals += 1
print(f"  removals until it shrinks again (to 16): {removals} (size {len(t)} <= 8)")
t = just_shrunk()
adds = 0
while len(t._slots) == 32:
    t.add_last(0)
    adds += 1
print(f"  adds until it grows again (to 64): {adds} (the first {adds - 1} fill the array, the last one finds it full)")

for shrink_at in (0.25, 0.5):
    t = CircularDeque(shrink_at)
    for i in range(33):
        t.add_last(i)                 # 33 items force capacity 64
    t.copies = 0
    for _ in range(1000):             # remove one, add one, 1000 times
        t.remove_last()
        t.add_last(0)
    print(f"shrink when size <= {shrink_at:.2f} of capacity: "
          f"{t.copies} elements copied by 2000 calls around 33 items, capacity {len(t._slots)}")

Worked trace

The first fourteen lines come from the instrumented operations: each operation's returned value, then the whole state after it, including the elements read front to back. The program printed:

add_last(1)     returned None
    cap=4 head=0 size=1 slots=[1, None, None, None] front-to-back=[1]
add_last(2)     returned None
    cap=4 head=0 size=2 slots=[1, 2, None, None] front-to-back=[1, 2]
add_first(0)    returned None
    cap=4 head=3 size=3 slots=[1, 2, None, 0] front-to-back=[0, 1, 2]
add_first(-1)   returned None
    cap=4 head=2 size=4 slots=[1, 2, -1, 0] front-to-back=[-1, 0, 1, 2]
add_last(3)     returned None
    cap=8 head=0 size=5 slots=[-1, 0, 1, 2, 3, None, None, None] front-to-back=[-1, 0, 1, 2, 3]
remove_first()  returned -1
    cap=8 head=1 size=4 slots=[None, 0, 1, 2, 3, None, None, None] front-to-back=[0, 1, 2, 3]
remove_last()   returned 3
    cap=8 head=1 size=3 slots=[None, 0, 1, 2, None, None, None, None] front-to-back=[0, 1, 2]
5000 random operations agree with collections.deque, for both shrink thresholds, and the capacity never fell below 4
1000 add_last calls: 1020 elements copied by growth (under 2 per call)
quarter rule: capacity 64 shrank to 32 when the size reached 16, so the new array is half full
  removals until it shrinks again (to 16): 8 (size 8 <= 8)
  adds until it grows again (to 64): 17 (the first 16 fill the array, the last one finds it full)
shrink when size <= 0.25 of capacity: 0 elements copied by 2000 calls around 33 items, capacity 64
shrink when size <= 0.50 of capacity: 64000 elements copied by 2000 calls around 33 items, capacity 64

Reading the trace: add_first(0) moves head from 0 to (0 - 1) % 4 = 3 and writes slot 3, so the slots read [1, 2, None, 0] while front to back it is [0, 1, 2]. add_first(-1) moves head to 2, and the array [1, 2, -1, 0] is now full; its logical order from the front is -1, 0, 1, 2 because slots 2 and 3 come before slots 0 and 1. add_last(3) finds the array full, so it doubles the capacity to 8, unrolls the contents to [-1, 0, 1, 2] starting at slot 0, and then writes 3 at slot 4. remove_first returns -1 and moves head to 1; remove_last returns 3. Size 3 in capacity 8 is above the shrink limit of 2, so no shrink happens.

The later lines are: the random comparison with collections.deque for both shrink thresholds, which also asserts that the capacity never drops below the minimum of 4; 1020 elements copied by growth over 1000 add_last calls (resizes at sizes 4, 8, 16, ..., 512 copy 4 + 8 + ... + 512 = 1020, which is under 2 per call); the shrink-at-a-quarter demonstration (capacity 64 shrinks to 32 when the size reaches 16, then 8 more removals shrink it again and the 17th add grows it, where the 16 adds before it fill the array); and the two shrink thresholds hovering around 33 items in a capacity-64 array: with the quarter rule nothing is copied, with the half rule 64000 elements are copied by 2000 calls (each remove-then-add pair costs a 32-element shrink and a 32-element grow).

Complexity

All operations are amortized O(1) time. Space is O(n): the capacity stays at most about four times the size (or the minimum of 4), because the shrink rule fires when the size falls to a quarter. Resizing a single time is O(n). Nothing is recursive.

Edge cases and common mistakes

• Empty deque: remove and peek raise an error.
• Resizing after the write instead of before, which overwrites the front element when the array is full (head + size wraps onto head).
• Copying the array as-is on resize instead of in logical order.
• Left step with a negative index. Python's % gives 3 for -1 % 4. In C and C++, -1 % 4 is -1 for signed integers, so write (head + capacity - 1) % capacity and use unsigned types.
• Using the same threshold for growing and shrinking (the thrash above).
• Not clearing a removed slot, which keeps a reference to the removed object alive in a garbage-collected language.
• Computing capacity // 2 when capacity is already at the minimum.

Follow-ups

Index access deque[i] is (head + i) % capacity, still O(1). A fixed-capacity version with overwrite is a ring buffer. To avoid O(n) pauses, grow incrementally while serving requests. In C++ the same design uses std::vector<T> or a raw array with head and size, and must construct and destroy elements in place rather than copy raw bytes.

Running the code

# save the listing above as circular_deque.py (Python 3.12, standard library only)
python circular_deque.py

The idea

A plain FIFO queue (first in, first out) is a linked list where you add at the tail and remove at the head, both in O(1). The hard part is pulling one specific task out of the middle and moving it to the front in O(1). Two facts make that possible:

• In a doubly linked list each node knows its prev and its next, so a node you already hold can be unlinked without searching for its neighbour.
• A hash map from task name to node hands you that node in O(1) on average, so you never walk the list to find it.

The list keeps the order, the map keeps the addresses. This is the same pairing that an LRU cache (least recently used cache) uses.

Why the obvious structures are not enough

• A Python list or collections.deque: finding a task is a scan, and removing it from the middle shifts or walks the elements, so promote is O(n). list.pop(0) is also O(n).
• A binary heap (priority queue, such as Python's heapq): the entry for a task cannot be found without a scan, re-ranking it once found takes a sift of O(log n), and dequeue is O(log n) as well, so promote is not O(1). Keeping equal priorities in arrival order also needs an extra counter.

Preventing starvation

If "promote" always means "go to the very front", a steady stream of promotions causes starvation (a task is ready to run but never gets served, because others keep jumping ahead of it) for everyone who was not promoted: each new promoted task lands ahead of the old ones, so the old ones are never reached. The run below shows exactly that with a cap switched off.

The fix shipped here is a burst cap. The queue holds two doubly linked lists, a promoted lane and a normal lane. Dequeue serves the promoted lane first, but after max_burst promoted tasks in a row it must serve one task from the normal lane (if there is one), then the count restarts. Promoting a task unlinks it from wherever it is and pushes it onto the front of the promoted lane, which is O(1).

The guarantee is a hard bound: a normal task with p normal tasks ahead of it is served within (p + 1) * (max_burst + 1) dequeue calls. The reasoning: the task is served after p + 1 normal services (the p tasks ahead of it, then itself), because new arrivals join behind it and so never add to that count. Before each of those normal services at most max_burst promoted tasks can be served, because the cap forces a normal one out once the promoted count reaches max_burst. So at most (p + 1) * max_burst promoted services plus p + 1 normal services happen first, which is (p + 1) * (max_burst + 1) dequeue calls. The starvation_demo lines in the output check the bound for p from 0 to 3. The cost of the cap is that a promoted task is not always literally first: when the cap fires, one normal task goes ahead of it (the second trace in the output shows this). If the product rule is that a promoted task must always be first, no queue can also promise that unpromoted work finishes, so the real lever is to limit who may promote and how often (a per-user quota: a cap on how many promotions each user may make in a period) instead. Aging (raising a waiting task's rank over time) also works but needs a rank per task and a heap, so it gives up O(1). A typical interview asks for the list-plus-map design and some fairness rule; the LRU comparison, aging and quotas are context around that.

Approach

1. Each Task node stores its name, prev, next and which lane it is in. Each Lane is a doubly linked list with two sentinel nodes (permanent dummy head and tail nodes), so inserting and unlinking never special-case an empty list.
2. enqueue(name): create the node, record it in by_name, push it on the back of the normal lane.
3. promote(name): look up the node, unlink it from its current lane, push it on the front of the promoted lane. Promoting an already promoted task just moves it to the front again.
4. dequeue(): choose the promoted lane unless the cap says the normal lane is due, pop the front, delete the name from the map. An empty queue raises IndexError. In words, the code serves the promoted lane when it has a task and at least one of these holds: the normal lane is empty (nothing else to serve), there is no cap, or fewer than max_burst promoted tasks have been served in a row. In every other case it serves the normal lane and resets the burst count to 0.

Code

import random


class Task:
    __slots__ = ("name", "prev", "next", "lane")

    def __init__(self, name):
        self.name = name
        self.prev = self.next = None
        self.lane = None          # the Lane this task currently sits in


class Lane:
    """Doubly linked list with two sentinel nodes, so no operation special-cases an empty list."""

    def __init__(self):
        self.head, self.tail = Task(None), Task(None)
        self.head.next, self.tail.prev = self.tail, self.head
        self.size = 0

    def push_back(self, task):
        last = self.tail.prev
        last.next = task
        task.prev, task.next = last, self.tail
        self.tail.prev = task
        task.lane = self
        self.size += 1

    def push_front(self, task):
        first = self.head.next
        self.head.next = task
        task.prev, task.next = self.head, first
        first.prev = task
        task.lane = self
        self.size += 1

    def unlink(self, task):
        task.prev.next, task.next.prev = task.next, task.prev
        task.prev = task.next = task.lane = None
        self.size -= 1

    def pop_front(self):
        task = self.head.next
        self.unlink(task)
        return task

    def names(self):
        out, node = [], self.head.next
        while node is not self.tail:
            out.append(node.name)
            node = node.next
        return out


class TaskQueue:
    """FIFO queue with O(1) promote. After max_burst promoted tasks in a row, one normal task is served."""

    def __init__(self, max_burst=None):
        self.normal, self.promoted = Lane(), Lane()
        self.by_name = {}         # name -> Task, so promote(name) finds the node in O(1)
        self.max_burst = max_burst
        self.burst = 0            # promoted tasks served since the last normal one

    def enqueue(self, name):
        task = Task(name)
        self.by_name[name] = task
        self.normal.push_back(task)

    def promote(self, name):
        task = self.by_name[name]
        task.lane.unlink(task)    # works for a task in either lane
        self.promoted.push_front(task)

    def dequeue(self):
        if len(self) == 0:
            raise IndexError("dequeue from an empty queue")
        use_promoted = self.promoted.size > 0 and (
            self.normal.size == 0 or self.max_burst is None or self.burst < self.max_burst)
        if use_promoted:
            self.burst += 1
            task = self.promoted.pop_front()
        else:
            self.burst = 0
            task = self.normal.pop_front()
        del self.by_name[task.name]
        return task.name

    def __len__(self):
        return self.normal.size + self.promoted.size

    def show(self):
        return f"promoted={self.promoted.names()} normal={self.normal.names()}"


def trace_demo(names, promote, max_burst):
    """Enqueue `names`, promote some of them, then dequeue everything, printing the lanes after each call."""
    q = TaskQueue(max_burst=max_burst)
    steps = [("enqueue", n) for n in names] + [("promote", n) for n in promote] + [("dequeue", None)] * len(names)
    print(f"max_burst={max_burst}")
    for op, arg in steps:
        result = q.dequeue() if op == "dequeue" else getattr(q, op)(arg)
        label = f"{op} {arg}" if arg else f"{op} -> {result}"
        print(f"{label:<14} {q.show()}  burst={q.burst}")


def links(lane):
    """One line per lane: every node with its prev and next, walking from head to tail."""
    def name(node):
        return "head" if node is lane.head else "tail" if node is lane.tail else node.name
    parts, node = [], lane.head
    while True:
        fields = [f"{side}={name(getattr(node, side))}" for side in ("prev", "next") if getattr(node, side)]
        parts.append(f"{name(node)}[{' '.join(fields)}]")
        if node is lane.tail:
            return " ".join(parts)
        node = node.next


def pointer_demo():
    """promote('b') on the queue a, b, c, split into its two steps."""
    q = TaskQueue()
    for name in "abc":
        q.enqueue(name)
    b = q.by_name["b"]
    print("start     normal  :", links(q.normal))
    print("          promoted:", links(q.promoted))
    b.lane.unlink(b)
    print("unlink b  normal  :", links(q.normal))
    print(f"          b itself: prev={b.prev} next={b.next} lane={b.lane}")
    q.promoted.push_front(b)
    print("push_front promoted:", links(q.promoted))


def starvation_demo(max_burst, ahead=0, rounds=50):
    """Rounds until the task with `ahead` normal tasks in front of it is served, or None."""
    q = TaskQueue(max_burst)
    for i in range(ahead + 1):
        q.enqueue(f"old{i}")
    for step in range(1, rounds + 1):     # every round: a new task arrives and is promoted at once
        q.enqueue(f"x{step}")
        q.promote(f"x{step}")
        if q.dequeue() == f"old{ahead}":
            return step
    return None


def reference_check(trials=300, seed=7):
    rng = random.Random(seed)
    for _ in range(trials):
        burst_cap = rng.choice([None, 1, 2, 3])
        q, normal, promoted, burst, counter = TaskQueue(burst_cap), [], [], 0, 0
        for _ in range(60):
            op = rng.choice(["enqueue", "enqueue", "promote", "dequeue"])
            if op == "enqueue":
                counter += 1
                q.enqueue(counter); normal.append(counter)
            elif op == "promote" and (normal or promoted):
                pick = rng.choice(normal + promoted)
                q.promote(pick)
                (normal if pick in normal else promoted).remove(pick)
                promoted.insert(0, pick)
            elif op == "dequeue" and (normal or promoted):
                if promoted and (not normal or burst_cap is None or burst < burst_cap):
                    burst += 1; expected = promoted.pop(0)
                else:
                    burst = 0; expected = normal.pop(0)
                assert q.dequeue() == expected
            assert q.normal.names() == normal and q.promoted.names() == promoted
    print("model check passed:", trials, "random sequences")


def edge_demo():
    """Operations the queue must refuse: an empty dequeue, and a promote of a name that is not queued."""
    q = TaskQueue()
    for label, call in (("dequeue on empty", q.dequeue), ("promote unknown name", lambda: q.promote("zzz"))):
        try:
            call()
        except (IndexError, KeyError) as error:
            print(f"{label}: {type(error).__name__}")
    q.enqueue("a")
    q.dequeue()
    try:
        q.promote("a")
    except KeyError:
        print("promote after it was served: KeyError")


trace_demo("abcd", "cd", max_burst=2)
trace_demo("abcde", "cde", max_burst=2)
pointer_demo()
print("old task served at round, no cap:", starvation_demo(None))
print("old task served at round, max_burst=3:", starvation_demo(3))
for ahead in range(4):
    print(f"max_burst=3, {ahead} normal tasks ahead: served at round {starvation_demo(3, ahead)}, bound {(ahead + 1) * (3 + 1)}")
reference_check()
edge_demo()

Run on Python 3.12 (python:3.12 image), it prints:

max_burst=2
enqueue a      promoted=[] normal=['a']  burst=0
enqueue b      promoted=[] normal=['a', 'b']  burst=0
enqueue c      promoted=[] normal=['a', 'b', 'c']  burst=0
enqueue d      promoted=[] normal=['a', 'b', 'c', 'd']  burst=0
promote c      promoted=['c'] normal=['a', 'b', 'd']  burst=0
promote d      promoted=['d', 'c'] normal=['a', 'b']  burst=0
dequeue -> d   promoted=['c'] normal=['a', 'b']  burst=1
dequeue -> c   promoted=[] normal=['a', 'b']  burst=2
dequeue -> a   promoted=[] normal=['b']  burst=0
dequeue -> b   promoted=[] normal=[]  burst=0
max_burst=2
enqueue a      promoted=[] normal=['a']  burst=0
enqueue b      promoted=[] normal=['a', 'b']  burst=0
enqueue c      promoted=[] normal=['a', 'b', 'c']  burst=0
enqueue d      promoted=[] normal=['a', 'b', 'c', 'd']  burst=0
enqueue e      promoted=[] normal=['a', 'b', 'c', 'd', 'e']  burst=0
promote c      promoted=['c'] normal=['a', 'b', 'd', 'e']  burst=0
promote d      promoted=['d', 'c'] normal=['a', 'b', 'e']  burst=0
promote e      promoted=['e', 'd', 'c'] normal=['a', 'b']  burst=0
dequeue -> e   promoted=['d', 'c'] normal=['a', 'b']  burst=1
dequeue -> d   promoted=['c'] normal=['a', 'b']  burst=2
dequeue -> a   promoted=['c'] normal=['b']  burst=0
dequeue -> c   promoted=[] normal=['b']  burst=1
dequeue -> b   promoted=[] normal=[]  burst=0
start     normal  : head[next=a] a[prev=head next=b] b[prev=a next=c] c[prev=b next=tail] tail[prev=c]
          promoted: head[next=tail] tail[prev=head]
unlink b  normal  : head[next=a] a[prev=head next=c] c[prev=a next=tail] tail[prev=c]
          b itself: prev=None next=None lane=None
push_front promoted: head[next=b] b[prev=head next=tail] tail[prev=b]
old task served at round, no cap: None
old task served at round, max_burst=3: 4
max_burst=3, 0 normal tasks ahead: served at round 4, bound 4
max_burst=3, 1 normal tasks ahead: served at round 8, bound 8
max_burst=3, 2 normal tasks ahead: served at round 12, bound 12
max_burst=3, 3 normal tasks ahead: served at round 16, bound 16
model check passed: 300 random sequences
dequeue on empty: IndexError
promote unknown name: KeyError
promote after it was served: KeyError

Worked trace

The first block shows the lanes after every call, with max_burst=2. Promoting c and then d leaves the promoted lane as ['d', 'c'] (the latest promotion is first) and the normal lane as ['a', 'b']. The two dequeues return d and c and the burst counter reaches 2. The cap is now reached, but the promoted lane is empty, so the next two dequeues serve a and b in their original FIFO order and the counter resets to 0. In this block the cap never has anything to hold back.

The second block is the one where it does. Three promotions leave the promoted lane as ['e', 'd', 'c']. The dequeues return e and d and the burst count reaches 2, and now c is still waiting in the promoted lane. The next dequeue returns a from the normal lane (the burst count drops to 0) even though c is waiting; only then does c come out, followed by b. That one normal task ahead of a promoted one is the price of the guarantee.

The edge_demo lines at the end of the output show the refusals described under the edge cases. The pointer_demo lines show what promote('b') does to the links, on the queue a, b, c. Each lane prints as name[prev=... next=...] from head to tail. At the start the normal lane is head -> a -> b -> c -> tail. Unlinking b is two writes, a.next = c and c.prev = a, and then b has no links (prev=None next=None lane=None). Pushing b on the front of the promoted lane makes it head -> b -> tail: head.next becomes b, tail.prev becomes b, and b points back to head and on to tail. The unlink line is one statement, task.prev.next, task.next.prev = task.next, task.prev. Python evaluates the whole right-hand side first, reading both task.next and task.prev, and only then assigns to the left-hand targets from left to right, so neither read sees a half-updated list. As separate statements: before = task.prev, after = task.next, before.next = after, after.prev = before.

The starvation lines use a queue holding one old task. Every round enqueues a new task, promotes it immediately and dequeues once. With no cap the old task is never served in 50 rounds (None). With max_burst=3 it is served in round 4, which matches the bound: p = 0, so (0 + 1) * (3 + 1) = 4. The last four lines repeat this with p = 0 to 3 normal tasks in front of the target: it is served at round 4, 8, 12 and 16, exactly the bound (p + 1) * 4 in each case.

Complexity

enqueue, promote and dequeue are each O(1): a fixed number of pointer assignments plus one dictionary operation (average O(1)). Space is O(n) for n queued tasks (one node and one map entry each).

Edge cases and common mistakes

• Dequeue from an empty queue raises IndexError. The check comes first so that no lane operation ever touches a sentinel node (the dequeue on empty line in the output).
• Task names must be unique, because the map is keyed by name. Use a task id in production.
• Promoting a name that is not queued raises KeyError, and so does promoting one that has already been served (the last two lines of the output); a service would return false instead.
• Forgetting to delete the map entry on dequeue leaks memory (the map keeps every served task), and promote of a task that already ran finds its stale node, whose lane is None, and dies with AttributeError instead of the KeyError the last output line expects, so the demo crashes there.
• Using a singly linked list makes promote O(n), because you must find the predecessor first.
• With several worker threads, guard the structure with one lock. Each critical section is O(1), so the lock stays short.

Running the code

docker run --rm -v "$PWD":/w -w /w python:3.12 python task_queue.py

Save the program above as task_queue.py. It needs only the standard library.

Direct answer

Use two pointers that both start at the head. slow moves one node per loop round and fast moves two. When fast can no longer move two steps, slow is standing on the middle. That is one trip over the list (a pass means one walk from the head to the end; each repetition of the loop is a round), O(n) time and O(1) extra space. For an even-length list I return the SECOND of the two middle nodes (index n // 2, counting from 0; // is whole-number division that drops any remainder, so 5 // 2 is 2 and 6 // 2 is 3), which is what the loop while fast and fast.next produces; the first middle (index (n - 1) // 2, which is 2 for 5 nodes and 2 for 6 nodes) needs a slightly earlier stop condition.

The idea in plain words

A singly linked list gives you no way to jump to position n // 2 because you do not know n and there is no index. The usual first answer is two passes over the list: count the nodes, then walk n // 2 steps. The one-pass trick is a speed difference. If one pointer moves twice as fast as the other, then at the moment the fast one reaches the end, the slow one has covered half the distance, so it is at the middle.

Why the pointers land where they do

Concretely, with the 5 nodes 1 2 3 4 5 (indexes 0 to 4): after 2 rounds slow is at index 2 and fast at index 4, the last node, so the loop stops and slow is on the 3. The general version follows. After k rounds, slow has moved k nodes (it is at index k) and fast has moved 2k nodes (index 2k).

• Odd length n = 2k + 1: after k rounds fast is at index 2k, the last node, and fast.next is None. The loop stops with slow at index k, the exact middle.
• Even length n = 2k: after k - 1 rounds fast is at index 2k - 2 and the last node (index 2k - 1) is still after it, so fast.next exists and the loop runs once more. After k rounds fast is None (it stepped past the end) and slow is at index k = n / 2, the second middle.

So the loop runs exactly n // 2 times and returns index n // 2.

Which middle to return for an even length, and why

I commit to the second middle (index n // 2) as the default, for two reasons: it is exactly what the simplest loop condition returns, and it equals nodes[len(nodes) // 2], the answer you would get by indexing into an array, so it is easy to explain and test. The first middle is the better choice when you need to CUT the list in half, as in merge sort of a list or a palindrome check: there slow must stop at the last node of the first half so you can set slow.next = None. The last lines of the program show it on 1 -> 2 -> 3 -> 4: cutting after the first middle gives two halves of 2 and 2, cutting after the second middle gives 3 and 1, and a palindrome check or a merge sort wants the balanced split. The code ships both. The first-middle version stops one round earlier by testing fast.next and fast.next.next instead.

Code

import random


class ListNode:
    def __init__(self, val, next=None):
        self.val = val
        self.next = next


def build(values):
    head = None
    for v in reversed(values):
        head = ListNode(v, head)
    return head


def to_list(head):
    out = []
    while head:
        out.append(head.val)
        head = head.next
    return out


def middle_second(head):
    """Second middle on even length (index n // 2)."""
    slow = fast = head
    while fast and fast.next:
        slow = slow.next
        fast = fast.next.next
    return slow


def middle_first(head):
    """First middle on even length (index (n - 1) // 2)."""
    slow = fast = head
    while fast.next and fast.next.next:
        slow = slow.next
        fast = fast.next.next
    return slow


def show_middle(values):
    """Print slow and fast positions after each loop round, then the answer."""
    head = build(values)
    nodes = to_list(head)
    print("list:", " -> ".join(map(str, nodes)), "-> None")
    slow = fast = head
    step = 0
    while fast and fast.next:
        slow = slow.next
        fast = fast.next.next
        step += 1
        print(f"  round {step}: slow at {slow.val}, fast at {fast.val if fast else None}")
    print(f"  returns {slow.val}   (first middle would be {middle_first(head).val})")


show_middle([1, 2, 3, 4, 5])
show_middle([1, 2, 3, 4, 5, 6])
show_middle([7, 8])
show_middle([9])

# Why the first middle is the one to use for cutting a list in half
def cut_after(head, node):
    """Detach everything after node; return the second half."""
    second, node.next = node.next, None
    return second


for name, find_middle in (("first middle ", middle_first), ("second middle", middle_second)):
    head = build([1, 2, 3, 4])
    second = cut_after(head, find_middle(head))
    print(f"cut after the {name}: {to_list(head)} and {to_list(second)}")

# Check against indexing, for every length from 1 to 40
rng = random.Random(3)
for n in range(1, 41):
    values = [rng.randint(0, 99) for _ in range(n)]
    head = build(values)
    node = middle_second(head)
    # walk from the head to find the position of the returned node
    pos, cur = 0, head
    while cur is not node:
        cur, pos = cur.next, pos + 1
    assert pos == n // 2
    first = middle_first(head)
    pos, cur = 0, head
    while cur is not first:
        cur, pos = cur.next, pos + 1
    assert pos == (n - 1) // 2
print("lengths 1 to 40: second middle is index n // 2, first middle is index (n - 1) // 2")

Output:

list: 1 -> 2 -> 3 -> 4 -> 5 -> None
  round 1: slow at 2, fast at 3
  round 2: slow at 3, fast at 5
  returns 3   (first middle would be 3)
list: 1 -> 2 -> 3 -> 4 -> 5 -> 6 -> None
  round 1: slow at 2, fast at 3
  round 2: slow at 3, fast at 5
  round 3: slow at 4, fast at None
  returns 4   (first middle would be 3)
list: 7 -> 8 -> None
  round 1: slow at 8, fast at None
  returns 8   (first middle would be 7)
list: 9 -> None
  returns 9   (first middle would be 9)
cut after the first middle : [1, 2] and [3, 4]
cut after the second middle: [1, 2, 3] and [4]
lengths 1 to 40: second middle is index n // 2, first middle is index (n - 1) // 2

Worked trace

Both pointers start on the head. For 1 -> 2 -> 3 -> 4 -> 5 -> None the printout shows round 1 with slow at 2 and fast at 3, round 2 with slow at 3 and fast at 5. fast.next is now None, so the loop stops and returns 3. For 1 -> 2 -> 3 -> 4 -> 5 -> 6 -> None there is a third round: slow moves to 4 and fast steps off the end to None, so the function returns 4 (the second middle) while the first-middle function returns 3. The two-node list 7 -> 8 takes one round and returns 8. The one-node list returns its only node after zero rounds. The two lines before the last show the cut. The final line is an assertion run over every length from 1 to 40, confirming the index formulas.

Complexity

• Time O(n): fast visits about every node once; the loop runs n // 2 rounds.
• Space O(1): two pointers, no recursion and no array.

Edge cases

• Empty list: middle_second(None) returns None (the loop test fast and fast.next is false at once). middle_first(None) would raise AttributeError because it reads fast.next; guard with if head is None: return None when empty input is possible.
• One node: both functions return it.
• Two nodes: second middle is the tail node, first middle is the head.
• Very long lists: the code is iterative, so Python's recursion limit (1000 frames by default) is not involved.

Common mistakes

• Writing while fast.next and fast.next.next and expecting the second middle: that returns the first middle for even lengths.
• Writing while fast.next: with fast = fast.next.next: on an even-length list fast becomes None, and the next test reads None.next, which raises AttributeError. The shipped conditions check fast before reading fast.next.
• Counting first, then walking: correct, but two passes over the list and it does not meet a "one pass" requirement.

Running the code

python middle_node.py

Python 3.12, standard library only.

The idea in plain words

For each element you want the first value to its right that is strictly greater (not equal), or -1 if none exists. The key observation: scan left to right and keep the indices that are still waiting for a greater value. When a new value arrives, it is the answer for every waiting element that is smaller than it. Those are always the most recent waiting elements, so a stack (last-in, first-out: the newest entry is the first to come off) holds the waiting indices, and resolving them means popping from the top until the top is not smaller than the new value.

The stack has a property worth stating in words: reading from the bottom to the top, each waiting value is equal to or smaller than the one below it (this is what "non-increasing" means). The reason is a short argument. Suppose a waiting value sat directly above a smaller one. When the larger value arrived, the smaller one was already waiting and the larger one is greater than it, so the smaller one would have been resolved and popped at that moment. It cannot still be waiting. You can see the property in the trace below: at i=4 the waiting entries are (1, 7), (3, 5), (4, 4), whose values 7, 5, 4 only go down from bottom to top, and when 6 arrives at i=5 it pops 4 and then 5 from the top and stops at 7.

A stack that keeps its contents in sorted order by popping anything that would break the order is called a monotonic stack (monotonic means always moving in one direction, here never increasing).

The brute force and why it is not enough

For each element, scan right until a larger value appears. On a descending array like [5, 4, 3, 2, 1] no element finds a greater value, so every scan runs to the end and the total is O(n^2). The program keeps this version as the reference to test against.

Approach, step by step

1. result = [-1] * n and an empty stack of indices.
2. For each index i with value v: while the stack is not empty and the value at the top index is smaller than v, pop that index j and set result[j] = v.
3. Push i. It now waits for something greater.
4. Indices still on the stack when the scan ends keep their -1.

Equal values do not resolve each other: the comparison is <, so for [4, 4, 4] nothing is popped and the answer is all -1.

The code

import random


def next_greater(nums, trace=False, stats=None):
    result = [-1] * len(nums)
    waiting = []                    # indices still waiting for a greater value; their values never increase
    for i, value in enumerate(nums):
        while waiting:
            if stats is not None:
                stats["comparisons"] += 1
            if nums[waiting[-1]] >= value:
                break               # the newest waiting value is not smaller, so nothing below it is either
            j = waiting.pop()
            result[j] = value
            if stats is not None:
                stats["pops"] += 1
            if trace:
                print(f"    i={i} value={value}: resolves index {j} (value {nums[j]})")
        waiting.append(i)
        if trace:
            print(f"  i={i} value={value}: waiting={[(k, nums[k]) for k in waiting]} result={result}")
    return result


def brute_force(nums):
    out = []
    for i, x in enumerate(nums):
        out.append(next((y for y in nums[i + 1:] if y > x), -1))
    return out


def demo():
    nums = [2, 7, 3, 5, 4, 6, 8]
    print("input:", nums)
    stats = {"comparisons": 0, "pops": 0}
    result = next_greater(nums, trace=True, stats=stats)
    print("answer:", result, "| pushes:", len(nums), "| pops:", stats["pops"], "| comparisons:", stats["comparisons"])
    print("duplicates [4, 4, 4]:", next_greater([4, 4, 4]))
    print("descending [5, 4, 3]:", next_greater([5, 4, 3]))
    print("empty []:", next_greater([]))


def random_check():
    rng = random.Random(9)
    for _ in range(2000):
        nums = [rng.randint(-5, 5) for _ in range(rng.randint(0, 12))]
        assert next_greater(nums) == brute_force(nums), nums
    print("2000 random arrays (with duplicates and negatives) match the brute force")
    for n in (1000, 2000, 4000):
        nums = [rng.randint(0, 10**6) for _ in range(n)]
        stats = {"comparisons": 0, "pops": 0}
        next_greater(nums, stats=stats)
        assert stats["pops"] <= n and stats["comparisons"] <= 2 * n
        print(f"n={n}: pops={stats['pops']} (at most n), comparisons={stats['comparisons']} (at most 2n={2 * n})")


if __name__ == "__main__":
    demo()
    random_check()

Run, it prints:

input: [2, 7, 3, 5, 4, 6, 8]
  i=0 value=2: waiting=[(0, 2)] result=[-1, -1, -1, -1, -1, -1, -1]
    i=1 value=7: resolves index 0 (value 2)
  i=1 value=7: waiting=[(1, 7)] result=[7, -1, -1, -1, -1, -1, -1]
  i=2 value=3: waiting=[(1, 7), (2, 3)] result=[7, -1, -1, -1, -1, -1, -1]
    i=3 value=5: resolves index 2 (value 3)
  i=3 value=5: waiting=[(1, 7), (3, 5)] result=[7, -1, 5, -1, -1, -1, -1]
  i=4 value=4: waiting=[(1, 7), (3, 5), (4, 4)] result=[7, -1, 5, -1, -1, -1, -1]
    i=5 value=6: resolves index 4 (value 4)
    i=5 value=6: resolves index 3 (value 5)
  i=5 value=6: waiting=[(1, 7), (5, 6)] result=[7, -1, 5, 6, 6, -1, -1]
    i=6 value=8: resolves index 5 (value 6)
    i=6 value=8: resolves index 1 (value 7)
  i=6 value=8: waiting=[(6, 8)] result=[7, 8, 5, 6, 6, 8, -1]
answer: [7, 8, 5, 6, 6, 8, -1] | pushes: 7 | pops: 6 | comparisons: 10
duplicates [4, 4, 4]: [-1, -1, -1]
descending [5, 4, 3]: [-1, -1, -1]
empty []: []
2000 random arrays (with duplicates and negatives) match the brute force
n=1000: pops=993 (at most n), comparisons=1985 (at most 2n=2000)
n=2000: pops=1995 (at most n), comparisons=3986 (at most 2n=4000)
n=4000: pops=3988 (at most n), comparisons=7978 (at most 2n=8000)

Worked trace

On [2, 7, 3, 5, 4, 6, 8]:

• 2 waits. 7 arrives and is greater than 2, so index 0 gets 7. Stack: [7].
• 3 is smaller than 7, so it joins the stack: [7, 3]. 5 is greater than 3, so index 2 gets 5; then 7 is not smaller than 5, so the popping stops.
• 4 joins: [7, 5, 4]. 6 pops 4 (index 4 gets 6) and pops 5 (index 3 gets 6), and stops at 7.
• 8 pops 6 (index 5 gets 8) and 7 (index 1 gets 8) and empties the stack. 8 itself is pushed and never resolved, so its answer stays -1.
• Final answer: [7, 8, 5, 6, 6, 8, -1].

Why the total work is linear

The inner while loop looks like it could make the algorithm quadratic, but count the pops across the whole run instead of per element. Every index is pushed exactly once, so at most n pops can ever happen. The comparisons are the pops plus at most one failing comparison per outer iteration (the one that ends the loop), so at most n + n = 2n. In the run above, 7 elements gave 6 pops and 10 comparisons. For random arrays of 1000, 2000 and 4000 values, the counters printed pops below n and comparisons below 2n, and the program asserts both bounds. One individual iteration can pop many elements (the 8 popped two), but those pops were paid for when the elements were pushed. This is an amortized argument: the cost is averaged over the whole sequence, not bounded per step.

Complexity

Time O(n). Space O(n): the result array, plus a stack that can hold all n indices at once (a strictly descending array pushes everything and pops nothing).

Edge cases and common mistakes

• Empty array: the loops never run and the result is [].
• Duplicates: use <, not <=, if the question means strictly greater. With <=, [4, 4, 4] would wrongly give [4, 4, -1].
• Storing values instead of indices on the stack. You need the index to know where to write the answer.
• Negative numbers need no special handling; the random test includes them.
• The last element is always -1 because nothing lies to its right.

Running the code

python next_greater.py

Python 3.12, standard library only.

The recommendation

For k in the tens of thousands with the lists already held as linked lists in memory, use pairwise divide and conquer: repeatedly merge neighbouring lists two at a time, so the number of lists halves each round and after about log2(k) rounds one list remains. Choose the heap (a k-entry min-heap holding the current front node of each list) when the inputs are streams you cannot hold at once, or when you need the merged output incrementally. Never use sequential merging (fold the lists into one accumulated result, one list at a time) at this k. What would flip the choice is listed after the measurements.

Terms: a comparison is one test of which of two keys is smaller; N is the total node count across all lists; min-heap is a structure whose smallest entry is always readable in O(1) and which replaces its top in O(log of its size); heapify means arranging k entries into a valid heap in O(k) time; constant factors are the per-step costs that big-O hides (a heap step does a few comparisons and moves, a plain merge step does one), which is why two methods with similar growth can still differ by a noticeable percentage.

The three methods

• Sequential: merge list 1 with list 2, then the result with list 3, and so on. The accumulated list grows to N nodes, and every later merge walks it again. For k lists of N/k nodes each, merge number i compares at most the accumulated (i-1)N/k nodes plus the new N/k, about iN/k comparisons. Adding i = 1 to k gives (1 + 2 + ... + k) * N/k = k(k+1)/2 * N/k = N(k+1)/2, which is about N*k/2 for large k. The ratio table below prints measured counts against N(k+1)/2.
• Pairwise: round 1 merges lists (1,2), (3,4), ...; round 2 merges the results in pairs; and so on. Each round touches at most N nodes and there are ceil(log2 k) rounds, so at most N*ceil(log2 k) comparisons. Merging two lists relinks nodes with pointers, so no new nodes are created.
• Heap: put the first node of each non-empty list in a min-heap. Repeatedly take the smallest, append it to the output, and put that list's next node in its place. Each node enters the heap once and each replacement costs O(log k) comparisons. Replacement (heapq.heapreplace) removes the smallest entry and inserts the new one in a single sift, which is cheaper than a pop followed by a push. In the code, Entry.__lt__ adds 1 to the shared counter each time the heap compares two entries, so the heap's comparisons are counted the same way as the merges' comparisons.

Measured comparison counts

The program counts every key comparison for all three methods on the same seeded data. It shows a small trace first (each heap line is printed after the take and refill, and a heap entry is (value, list index)), then the counts for N = 20000 at four values of k, a ratio table computed from them, then a skewed case where one list holds half of the nodes.

import heapq
import math
import random


class Node:
    __slots__ = ("val", "next")

    def __init__(self, val, nxt=None):
        self.val, self.next = val, nxt


comparisons = 0                       # every key comparison any method makes is counted here


def merge_two(a, b):
    global comparisons
    dummy = tail = Node(0)
    while a and b:
        comparisons += 1
        if b.val < a.val:
            tail.next, b = b, b.next
        else:
            tail.next, a = a, a.next
        tail = tail.next
    tail.next = a or b                # the leftover run is attached without comparing
    return dummy.next


def merge_sequential(heads):
    result = None
    for head in heads:
        result = merge_two(result, head)
    return result


def merge_pairwise(heads, trace=False):
    heads = list(heads)
    round_number = 0
    while len(heads) > 1:             # one round: merge neighbours, halving the number of lists
        merged = [merge_two(heads[i], heads[i + 1]) for i in range(0, len(heads) - 1, 2)]
        if len(heads) % 2:
            merged.append(heads[-1])
        heads = merged
        round_number += 1
        if trace:
            print(f"  round {round_number}: {[values_of(h) for h in heads]}")
    return heads[0] if heads else None


class Entry:
    __slots__ = ("val", "idx", "node")

    def __init__(self, node, idx):
        self.val, self.idx, self.node = node.val, idx, node

    def __lt__(self, other):          # ties broken by list index, so nodes are never compared
        global comparisons
        comparisons += 1
        return (self.val, self.idx) < (other.val, other.idx)


def merge_heap(heads, trace=False):
    heap = [Entry(h, i) for i, h in enumerate(heads) if h]
    heapq.heapify(heap)
    dummy = tail = Node(0)
    while heap:
        top = heap[0]
        tail.next = top.node
        tail = tail.next
        nxt = top.node.next
        if nxt:
            heapq.heapreplace(heap, Entry(nxt, top.idx))      # pop and push in one sift
        else:
            heapq.heappop(heap)
        if trace:
            refill = f"list {top.idx} refills with {nxt.val}" if nxt else f"list {top.idx} is exhausted"
            print(f"  take {top.val} from list {top.idx}; {refill}; heap now holds {sorted((e.val, e.idx) for e in heap)}")
    tail.next = None
    return dummy.next


def make_lists(sizes, rng):
    heads = []
    for size in sizes:
        values = sorted(rng.randrange(10**9) for _ in range(size))
        head = None
        for v in reversed(values):
            head = Node(v, head)
        heads.append(head)
    return heads


def values_of(head):
    out = []
    while head:
        out.append(head.val)
        head = head.next
    return out


def run(method, sizes, seed):
    global comparisons
    heads = make_lists(sizes, random.Random(seed))        # same data for every method
    comparisons = 0
    merged = method(heads)
    return comparisons, values_of(merged)


small = [[1, 4, 7], [2, 5], [0, 6], [3]]


def small_lists():
    heads = []
    for values in small:
        head = None
        for v in reversed(values):
            head = Node(v, head)
        heads.append(head)
    return heads


print("Lists:", small)
print("Pairwise (divide and conquer):")
print("  result:", values_of(merge_pairwise(small_lists(), trace=True)))
print("Heap:")
print("  result:", values_of(merge_heap(small_lists(), trace=True)))
print()

N = 20000
print(f"N = {N} nodes in total, k lists of N/k nodes each; counts are key comparisons")
print(f"{'k':>6} {'sequential':>12} {'pairwise':>10} {'heap':>10} {'N*ceil(log2 k)':>15} {'N*k/2':>12}")
table = []
for k in (125, 250, 500, 1000):
    sizes = [N // k] * k
    seq, out_a = run(merge_sequential, sizes, k)
    pair, out_b = run(merge_pairwise, sizes, k)
    hp, out_c = run(merge_heap, sizes, k)
    assert out_a == out_b == out_c == sorted(out_a) and len(out_a) == N
    print(f"{k:>6} {seq:>12} {pair:>10} {hp:>10} {N * math.ceil(math.log2(k)):>15} {N * k // 2:>12}")
    table.append((k, seq, pair, hp))
print("\nRatios computed from the table above:")
print(f"{'k':>6} {'seq/(N*(k+1)/2)':>16} {'heap vs pairwise':>17} {'pairwise vs heap':>17}")
for k, seq, pair, hp in table:
    print(f"{k:>6} {seq / (N * (k + 1) / 2):>15.1%} {hp / pair - 1:>+16.1%} {pair / hp - 1:>+16.1%}")
for (k1, _, p1, _), (k2, _, p2, _) in zip(table, table[1:]):
    print(f"pairwise count grows by {p2 - p1} from k={k1} to k={k2}")

print("\nSkewed sizes: k = 1000 lists, one holds half of all nodes")
sizes = [N // 2] + [(N // 2) // 999] * 999
sizes[0] += N - sum(sizes)
for name, method in (("sequential", merge_sequential), ("pairwise", merge_pairwise), ("heap", merge_heap)):
    count, out = run(method, sizes, 7)
    assert out == sorted(out) and len(out) == N
    print(f"  {name:10} {count:>9}")

print("\nEdge cases:")
print("  no lists:", values_of(merge_heap([])), values_of(merge_pairwise([])), values_of(merge_sequential([])))
print("  empty lists mixed in:", values_of(merge_pairwise([None, make_lists([2], random.Random(1))[0], None])))

Output:

Lists: [[1, 4, 7], [2, 5], [0, 6], [3]]
Pairwise (divide and conquer):
  round 1: [[1, 2, 4, 5, 7], [0, 3, 6]]
  round 2: [[0, 1, 2, 3, 4, 5, 6, 7]]
  result: [0, 1, 2, 3, 4, 5, 6, 7]
Heap:
  take 0 from list 2; list 2 refills with 6; heap now holds [(1, 0), (2, 1), (3, 3), (6, 2)]
  take 1 from list 0; list 0 refills with 4; heap now holds [(2, 1), (3, 3), (4, 0), (6, 2)]
  take 2 from list 1; list 1 refills with 5; heap now holds [(3, 3), (4, 0), (5, 1), (6, 2)]
  take 3 from list 3; list 3 is exhausted; heap now holds [(4, 0), (5, 1), (6, 2)]
  take 4 from list 0; list 0 refills with 7; heap now holds [(5, 1), (6, 2), (7, 0)]
  take 5 from list 1; list 1 is exhausted; heap now holds [(6, 2), (7, 0)]
  take 6 from list 2; list 2 is exhausted; heap now holds [(7, 0)]
  take 7 from list 0; list 0 is exhausted; heap now holds []
  result: [0, 1, 2, 3, 4, 5, 6, 7]

N = 20000 nodes in total, k lists of N/k nodes each; counts are key comparisons
     k   sequential   pairwise       heap  N*ceil(log2 k)        N*k/2
   125      1253242     139427     158245          140000      1250000
   250      2476883     159187     178142          160000      2500000
   500      4891329     178748     197928          180000      5000000
  1000      9540577     197735     217120          200000     10000000

Ratios computed from the table above:
     k  seq/(N*(k+1)/2)  heap vs pairwise  pairwise vs heap
   125           99.5%           +13.5%           -11.9%
   250           98.7%           +11.9%           -10.6%
   500           97.6%           +10.7%            -9.7%
  1000           95.3%            +9.8%            -8.9%
pairwise count grows by 19760 from k=125 to k=250
pairwise count grows by 19561 from k=250 to k=500
pairwise count grows by 18987 from k=500 to k=1000

Skewed sizes: k = 1000 lists, one holds half of all nodes
  sequential  13649446
  pairwise      196482
  heap          280733

Edge cases:
  no lists: [] [] []
  empty lists mixed in: [144272509, 611178002]

Worked trace

For the four lists [1,4,7], [2,5], [0,6], [3]: pairwise round 1 merges the first two lists and the last two, giving [1,2,4,5,7] and [0,3,6]. Round 2 merges those into [0,1,2,3,4,5,6,7]. The heap takes 0 from list 2 (the lists are numbered from 0), replaces it with that list's next node 6, takes 1 from list 0, replaces it with 4, and so on until the heap is empty. Each trace line shows the heap after that step, so the first line already contains the refill (6, 2). Both end at the same sequence.

What the numbers say

• Sequential grows with k. Its count goes 1,253,242; 2,476,883; 4,891,329; 9,540,577 as k doubles from 125 to 1000, about double each time. It is 100.3% down to 95.4% of N*k/2 and 99.5% down to 95.3% of the exact sum N(k+1)/2 in the ratio table.
• Pairwise and heap grow with log k. Pairwise rises by about N (18,987 to 19,760 against N = 20000, as the printed differences show) per doubling of k, one extra round, and stays under N*ceil(log2 k). The heap's count tracks it, 9.8% to 13.5% higher at every k measured. Read the other way round, pairwise makes 8.9% to 11.9% fewer comparisons than the heap, not 10% to 14%: a percentage taken against the smaller number is bigger than the same gap taken against the larger one.
• Skew: with k = 1000 and one list holding half of all nodes, sequential needs 13,649,446 comparisons, pairwise 196,482 and the heap 280,733. This is one seeded random data set, but pairwise is already within 2% of its N*ceil(log2 k) ceiling in every row above (196,482 against 200,000 here), so no input can make it cost much more than shown.

ESTIMATE at the stated scale (extrapolated, not part of the output above)

Take k = 20,000 and N = 10 million. Sequential: about Nk/2 = 1e11 comparisons from the formula the table validates (the table shows 95% to 100% of it, so the true count is a little lower). Pairwise: at most N15 = 1.5e8, since 2^14 = 16,384 < 20,000 <= 2^15. The heap does not keep the 10% to 14% margin of the table. That margin narrows as k grows (13.5% at k = 125 down to 9.8% at k = 1000), and a separate run of the same functions at N = 2,500,000 and k = 20,000 (equal lists, seed 20000) gave 36,540,003 comparisons for pairwise (97.4% of N*ceil(log2 k)) and 38,350,043 for the heap, 5.0% more. Scaling that run by four is an ESTIMATE: pairwise about 1.46e8, heap about 1.53e8, sequential about 700 times pairwise. Sequential is out at this scale either way.

Choosing, with the constraints

1. Lists already in memory as nodes: pairwise. It makes fewer comparisons than the heap in every run above (8.9% to 11.9% fewer in the equal-size table, 196,482 against 280,733 in the skewed case). It allocates no node or object per element (each merge creates one dummy head, so k - 1 in all), and each merge scans two lists in sequence. Extra space is the array of list heads, at most k entries, halving each round.
2. Lists on disk or arriving over a network, or output needed before the merge finishes: heap. It needs only the current front of each list, k entries in total, and its first output is available after the initial heapify (building the heap in O(k)). Pairwise must finish whole rounds, so it needs access to all lists at once.
3. You only need the smallest m values of the merge: heap, O(k + m log k) work, because it can stop early.
4. k below about ten: any method is fine; sequential is the simplest to read.

Wall-clock speed is not measured above. In CPython a heap of plain (value, list index, node) tuples compares in C code (heapq), which can offset its extra comparisons, whereas the Entry class used here runs Python code on every comparison so that it can count them. In compiled languages the comparison counts predict the ranking more closely. Measure in your own language before committing.

Complexity of the shipped code

• Sequential: about N*k/2 comparisons for equal sizes (worst case O(kN)); O(1) extra space.
• Pairwise: at most N*ceil(log2 k) comparisons, O(1) extra space beyond the head array of at most k entries. merge_two is a loop, not recursion, so list length does not hit Python's recursion limit (about 1000 nested calls by default).
• Heap: O(N log k) comparisons, O(k) extra space for the heap; this code allocates one small Entry object per node.

Edge cases and common mistakes

• No lists, empty lists mixed in: handled (the last two output lines). The heap skips None heads; pairwise passes None through merge_two.
• Ties in the heap: pushing (value, node) pairs fails with a TypeError on equal values if nodes define no ordering. The Entry class breaks ties by list index, so nodes are never compared.
• Sequential merge "because it is simplest": the kN cost appears only at scale.
• Recursive two-way merge: its depth is the number of nodes, so a Python version fails above about a thousand nodes.
• Forgetting the leftover run: when one list runs out, attach the rest of the other in one pointer assignment (tail.next = a or b) with no further comparisons.

Follow-ups

heapq.merge in the standard library does the heap method over sorted iterables. For lists that do not fit in memory, merge in batches of as many runs as the memory allows, then merge the batch outputs. That is the external merge sort pattern: sort data too big for memory by sorting pieces that fit, writing each sorted piece to disk, and merging the pieces.

Running the code

Save as merge_k_lists.py, then run: python merge_k_lists.py
Needs Python 3.12 and the standard library only. It uses no external packages.