Algorithm Search⌘K

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
blocked_matrix_multiplication.pyRawGitHub
# 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

Read the full page about Blocked matrix multiplication