Blocked matrix multiplication in Python
Runs the triple loop on square tiles sized so that three tiles fit in fast memory, which cuts the block transfers from about n^3/B to O(n^3/(B sqrt(M)) + n^2/B).
- File
blocked_matrix_multiplication.py, 113 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/blocked-matrix-multiplication/. The authors are listed in prov/4VVM7H.jsonl.
from collections import OrderedDict
from typing import Iterator, Optional
Matrix = list[list[int]]
def blocked_matrix_multiplication(a: Matrix, b: Matrix, p: Optional[int] = None,
mod: Optional[int] = None, tile: int = 32) -> Matrix:
"""Return a * b, computed tile by tile with square tiles of side `tile`.
a is n x m and b is m x p (lists of rows); pass p when m = 0. The ring is
Z, or Z/mod Z when mod is set. Edge tiles are smaller when a side is not a
multiple of `tile`. Cost: n m p multiplications and additions, as for the
triple loop; only the order of the operations changes.
"""
if tile < 1:
raise ValueError("the tile side must be at least 1")
n, m = len(a), len(b)
if p is None:
p = len(b[0]) if m > 0 else 0
if any(len(row) != m for row in a) or any(len(row) != p for row in b):
raise ValueError("the shapes do not match")
c = [[0] * p for _ in range(n)]
for i0 in range(0, n, tile): # @line L1
i1 = min(i0 + tile, n)
for j0 in range(0, p, tile): # @line L2
j1 = min(j0 + tile, p)
ct = [c[i][j0:j1] for i in range(i0, i1)] # @line L3
for k0 in range(0, m, tile): # @line L4
k1 = min(k0 + tile, m)
at = [a[i][k0:k1] for i in range(i0, i1)] # @line L5
bt = [b[k][j0:j1] for k in range(k0, k1)]
for ii, arow in enumerate(at): # @line L6
crow = ct[ii]
for kk, aik in enumerate(arow):
if aik:
brow = bt[kk]
for jj in range(j1 - j0):
crow[jj] += aik * brow[jj]
if mod is not None:
ct = [[x % mod for x in row] for row in ct]
for ii, i in enumerate(range(i0, i1)): # @line L7
c[i][j0:j1] = ct[ii]
return c
def _lines(base: int, ld: int, r0: int, r1: int, c0: int, c1: int, line: int) -> set[int]:
"""Indices of the memory blocks of `line` words touched by rows r0..r1-1, columns c0..c1-1
of a row-major matrix that starts at word `base` and has `ld` words per row."""
out: set[int] = set()
if c1 <= c0:
return out
for r in range(r0, r1):
first, last = base + r * ld + c0, base + r * ld + c1 - 1
out.update(range(first // line, last // line + 1))
return out
def explicit_transfers(n: int, m: int, p: int, tile: int, line: int) -> int:
"""Block transfers of the tiled schedule in the external memory model.
A, B and C are row-major and stored one after the other. Each tile of C is
read once and written once, and each pair of tiles of A and B is read once
for each tile of C, as in lines L3, L5 and L7. Memory needs room for three tiles.
"""
base_a, base_b, base_c = 0, n * m, n * m + m * p
total = 0
for i0 in range(0, n, tile):
i1 = min(i0 + tile, n)
for j0 in range(0, p, tile):
j1 = min(j0 + tile, p)
total += 2 * len(_lines(base_c, p, i0, i1, j0, j1, line))
for k0 in range(0, m, tile):
k1 = min(k0 + tile, m)
total += len(_lines(base_a, m, i0, i1, k0, k1, line))
total += len(_lines(base_b, p, k0, k1, j0, j1, line))
return total
def _trace(n: int, tile: Optional[int]) -> Iterator[int]:
"""Word addresses read by the n x n product in loop order i, k, j (tile=None) or tiled."""
base_b, base_c = n * n, 2 * n * n
t = n if tile is None else tile
for i0 in range(0, n, t):
for j0 in range(0, n, t):
for k0 in range(0, n, t):
for i in range(i0, min(i0 + t, n)):
for k in range(k0, min(k0 + t, n)):
yield i * n + k
for j in range(j0, min(j0 + t, n)):
yield base_b + k * n + j
yield base_c + i * n + j
def lru_misses(n: int, line: int, cache_words: int, tile: Optional[int] = None) -> int:
"""Misses of a fully associative LRU cache of cache_words words and lines of `line` words,
for the n x n product in loop order i, k, j (tile=None) or with square tiles of side `tile`."""
cap = cache_words // line
cache: OrderedDict[int, None] = OrderedDict()
misses = 0
for addr in _trace(n, tile):
blk = addr // line
if blk in cache:
cache.move_to_end(blk)
else:
misses += 1
cache[blk] = None
if len(cache) > cap:
cache.popitem(last=False)
return misses
test_blocked_matrix_multiplication.py 85 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/blocked-matrix-multiplication/. The authors are listed in prov/4VVM7H.jsonl.
import random
import pytest
from blocked_matrix_multiplication import blocked_matrix_multiplication as mul
from blocked_matrix_multiplication import explicit_transfers, lru_misses
def schoolbook(a, b, p, mod=None):
n, m = len(a), len(b)
c = [[0] * p for _ in range(n)]
for i in range(n):
for j in range(p):
s = 0
for k in range(m):
s += a[i][k] * b[k][j]
c[i][j] = s % mod if mod is not None else s
return c
def rand(rng, r, c, lo=-9, hi=9):
return [[rng.randint(lo, hi) for _ in range(c)] for _ in range(r)]
def test_edges() -> None:
assert mul([], []) == []
assert mul([[5]], [[-3]]) == [[-15]]
assert mul([[], []], [], p=3) == [[0, 0, 0], [0, 0, 0]]
assert mul([[1, 2]], [[3], [4]], tile=1) == [[11]]
with pytest.raises(ValueError):
mul([[1, 2]], [[1, 2]])
with pytest.raises(ValueError):
mul([[1]], [[1]], tile=0)
def test_identity_and_zero() -> None:
rng = random.Random(5)
for n in (1, 2, 5, 7, 16):
a = rand(rng, n, n)
eye = [[int(i == j) for j in range(n)] for i in range(n)]
zero = [[0] * n for _ in range(n)]
assert mul(a, eye, tile=3) == a
assert mul(eye, a, tile=4) == a
assert mul(a, zero, tile=2) == zero
def test_random_against_schoolbook() -> None:
rng = random.Random(11)
for _ in range(600):
n, m, p = rng.randint(1, 9), rng.randint(1, 9), rng.randint(1, 9)
a, b = rand(rng, n, m), rand(rng, m, p)
assert mul(a, b, tile=rng.randint(1, 10)) == schoolbook(a, b, p)
def test_mod_small_primes() -> None:
rng = random.Random(12)
for q in (2, 3, 5, 7):
for _ in range(40):
n, m, p = rng.randint(1, 12), rng.randint(1, 12), rng.randint(1, 12)
a, b = rand(rng, n, m, 0, q - 1), rand(rng, m, p, 0, q - 1)
assert mul(a, b, mod=q, tile=rng.randint(1, 5)) == schoolbook(a, b, p, q)
def test_explicit_transfer_formula() -> None:
# aligned case: 2 n^3 / (b B) + 2 n^2 / B
for n, b, line in ((8, 2, 2), (16, 4, 4), (64, 16, 8), (48, 8, 4)):
assert explicit_transfers(n, n, n, b, line) == 2 * n ** 3 // (b * line) + 2 * n * n // line
# unaligned tiles: each tile row of b words touches at most ceil(b/B) + 1 blocks
for n, b, line in ((10, 3, 4), (13, 5, 4), (17, 6, 8)):
tiles = -(-n // b)
bound = 2 * (tiles ** 3 + tiles ** 2) * b * (-(-b // line) + 1)
t = explicit_transfers(n, n, n, b, line)
assert 2 * n ** 3 // (b * line) <= t <= bound
def test_lru_observed_counts() -> None:
assert lru_misses(64, 8, 768, tile=16) == 6000
assert lru_misses(64, 8, 768, tile=None) == 33792
assert lru_misses(8, 2, 12, tile=2) == 336
# the tiled order makes fewer misses than the i, k, j order once B does not fit
for n in (24, 40):
assert lru_misses(n, 4, 192, tile=8) < lru_misses(n, 4, 192)
What the next level needs
- the tests did not run in this build