Parallel quicksort in Python
Quicksort that partitions in parallel and sorts the two halves concurrently.
- File
parallel_quicksort.py, 69 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/parallel-quicksort/. The authors are listed in prov/X4M2NF.jsonl.
"""Parallel quicksort with a three-way partition by parallel pack.
The code runs the parallel steps one after the other, which is a legal
schedule. It also reports the recursion depth, and the work and span of
the computation with unit cost per key per pass and a scan of span
about lg n.
"""
import random
from dataclasses import dataclass
from typing import Protocol, TypeVar
class Ordered(Protocol):
def __lt__(self, other: "Ordered", /) -> bool: ...
T = TypeVar("T", bound=Ordered)
@dataclass
class Cost:
work: int = 0
span: int = 0
depth: int = 0
def exclusive_scan(flags: list[int]) -> tuple[list[int], int]:
"""Exclusive prefix sums of flags, and their total."""
out, total = [], 0
for f in flags:
out.append(total)
total += f
return out, total
def pack(a: list[T], flags: list[int]) -> list[T]:
"""Keep the items with flag 1, in their input order."""
pos, total = exclusive_scan(flags) # @line L10
out: list[T] = [a[0]] * total if total else []
for i, f in enumerate(flags): # @line L11
if f:
out[pos[i]] = a[i]
return out # @line L12
def parallel_quicksort(a: list[T], rng: random.Random | None = None) -> tuple[list[T], Cost]:
"""Return a sorted copy of a (stable), and the cost of the run."""
if rng is None:
rng = random.Random(0)
n = len(a) # @line L1
if n <= 1: # @line L2
return list(a), Cost(1, 1, 0)
x = a[rng.randrange(n)] # @line L3
lt = [1 if v < x else 0 for v in a] # @line L4
gt = [1 if x < v else 0 for v in a]
eq = [1 - p - q for p, q in zip(lt, gt)]
less, equal, greater = pack(a, lt), pack(a, eq), pack(a, gt) # @line L5
left, c1 = parallel_quicksort(less, rng) # @line L6
right, c2 = parallel_quicksort(greater, rng) # @line L7
# sync: both parts are sorted here # @line L8
out = left + equal + right # @line L9
scan_span = n.bit_length()
work = 5 * n + c1.work + c2.work
span = 3 * scan_span + max(c1.span, c2.span)
return out, Cost(work, span, 1 + max(c1.depth, c2.depth))
test_parallel_quicksort.py 66 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/parallel-quicksort/. The authors are listed in prov/X4M2NF.jsonl.
import math
import random
from parallel_quicksort import pack, parallel_quicksort
def test_example() -> None:
out, _ = parallel_quicksort([5, 2, 8, 2, 9, 1, 5, 3], random.Random(1))
assert out == [1, 2, 2, 3, 5, 5, 8, 9]
def test_pack() -> None:
assert pack([5, 2, 8, 2, 9], [0, 1, 0, 1, 1]) == [2, 2, 9]
assert pack([5, 2], [0, 0]) == []
def test_edges() -> None:
for a in ([], [1], [2, 1], [1, 1, 1, 1], [3, 2, 1], [1, 2, 3]):
out, _ = parallel_quicksort(a)
assert out == sorted(a)
def test_stable() -> None:
class Key:
def __init__(self, k: int, tag: int) -> None:
self.k, self.tag = k, tag
def __lt__(self, other: "Key") -> bool:
return self.k < other.k
rng = random.Random(5)
for _ in range(50):
a = [Key(rng.randint(0, 5), i) for i in range(40)]
out, _ = parallel_quicksort(a, rng)
assert [(x.k, x.tag) for x in out] == sorted((x.k, x.tag) for x in a)
def test_depth_is_logarithmic() -> None:
rng = random.Random(7)
for n in (100, 1000, 10000):
a = list(range(n))
rng.shuffle(a)
out, cost = parallel_quicksort(a, rng)
assert out == sorted(a)
k = math.ceil(math.log(n, 4 / 3)) + 1
assert cost.depth <= 8 * k
assert cost.work <= 5 * n * (cost.depth + 1) + 2 * n + 1
def test_sorted_and_all_equal_inputs() -> None:
rng = random.Random(9)
out, cost = parallel_quicksort([4] * 500, rng)
assert out == [4] * 500 and cost.depth == 1
out, _ = parallel_quicksort(list(range(2000)), rng)
assert out == list(range(2000))
def test_random_against_sorted() -> None:
rng = random.Random(11)
for _ in range(300):
a = [rng.randint(-20, 20) for _ in range(rng.randint(0, 80))]
out, _ = parallel_quicksort(a, rng)
assert out == sorted(a)
What the next level needs
- the tests did not run in this build