frontier_code/tests/test_perf.py

"""Tests for the Frontier F19 performance lab (toy VLIW/SIMD machine and batched trees).

Run from frontier_code/:
    python -m unittest discover -s tests -p "test_*.py" -v
Test your own starter code:
    MANUAL_IMPL=starters python -m unittest tests.test_perf -v
"""

import bisect
import importlib
import os
import random
import unittest

PACKAGE = os.environ.get("MANUAL_IMPL", "solutions")
lab = importlib.import_module(f"{PACKAGE}.perf_lab")
Op = lab.Op
SLOTS = lab.TOY_SLOTS


def example_ops():
    """y = (a + b) * (c + d) + e * f, loading six inputs and storing y."""
    return [
        Op("la", "load"), Op("lb", "load"), Op("lc", "load"),
        Op("ld", "load"), Op("le", "load"), Op("lf", "load"),
        Op("s1", "alu", ("la", "lb")), Op("s2", "alu", ("lc", "ld")),
        Op("m1", "alu", ("s1", "s2")), Op("m2", "alu", ("le", "lf")),
        Op("y", "alu", ("m1", "m2")), Op("st", "store", ("y",)),
    ]


class BoundTests(unittest.TestCase):
    def test_critical_path_of_example(self):
        self.assertEqual(lab.critical_path(example_ops()), 5)

    def test_resource_bound_of_example(self):
        self.assertEqual(lab.resource_bound(example_ops(), SLOTS), 6)

    def test_chain_is_its_length(self):
        self.assertEqual(lab.critical_path(lab.independent_chains(1, 7)), 7)

    def test_cycle_is_rejected(self):
        with self.assertRaises(ValueError):
            lab.critical_path([Op("a", "alu", ("b",)), Op("b", "alu", ("a",))])


class ScheduleTests(unittest.TestCase):
    def test_schedule_is_valid(self):
        ops = example_ops()
        lab.verify_schedule(ops, SLOTS, lab.list_schedule(ops, SLOTS))

    def test_example_schedule_is_nine_cycles(self):
        # Six loads need cycles 0-5; the chain after the last load needs 3 more.
        self.assertEqual(len(lab.list_schedule(example_ops(), SLOTS)), 9)

    def test_one_chain_is_latency_bound(self):
        ops = lab.independent_chains(1, 6)
        bundles = lab.list_schedule(ops, SLOTS)
        self.assertEqual(len(bundles), 6)
        self.assertAlmostEqual(lab.utilization(bundles, SLOTS)["valu"], 0.5)

    def test_two_chains_fill_both_vector_slots(self):
        ops = lab.independent_chains(2, 6)
        bundles = lab.list_schedule(ops, SLOTS)
        self.assertEqual(len(bundles), 6)
        self.assertAlmostEqual(lab.utilization(bundles, SLOTS)["valu"], 1.0)

    def test_many_chains_reach_the_resource_bound(self):
        ops = lab.independent_chains(8, 6)
        self.assertEqual(len(lab.list_schedule(ops, SLOTS)), lab.resource_bound(ops, SLOTS))

    def test_verify_catches_same_cycle_dependency(self):
        ops = [Op("a", "alu"), Op("b", "alu", ("a",))]
        with self.assertRaises(ValueError):
            lab.verify_schedule(ops, SLOTS, [{"alu": ["a", "b"]}])

    def test_verify_catches_slot_overflow(self):
        ops = [Op(f"l{i}", "load") for i in range(2)]
        with self.assertRaises(ValueError):
            lab.verify_schedule(ops, SLOTS, [{"load": ["l0", "l1"]}])


class TreeTests(unittest.TestCase):
    def setUp(self):
        rng = random.Random(1)
        self.keys = sorted(rng.sample(range(100_000), 2 ** 10 - 1))
        self.tree = lab.eytzinger(self.keys)
        self.queries = [rng.randrange(-10, 100_010) for _ in range(500)] + self.keys[:20]

    def test_inorder_layout_is_sorted(self):
        tree = lab.eytzinger([10, 20, 30, 40, 50, 60, 70])
        self.assertEqual(tree, [None, 40, 20, 60, 10, 30, 50, 70])

    def test_single_search_matches_bisect(self):
        for q in self.queries:
            self.assertEqual(lab.search_one(self.tree, q), bisect.bisect_left(self.keys, q))

    def test_batch_matches_single(self):
        expected = [bisect.bisect_left(self.keys, q) for q in self.queries]
        self.assertEqual(lab.search_batch_lockstep(self.tree, self.queries), expected)

    def test_rejects_incomplete_tree(self):
        with self.assertRaises(ValueError):
            lab.eytzinger(list(range(10)))

    def test_top_levels_are_shared_by_the_whole_batch(self):
        counts = lab.distinct_nodes_per_level(self.tree, self.queries[:256])
        self.assertEqual(counts[:4], [1, 2, 4, 8])
        self.assertTrue(all(c <= min(2 ** l, 256) for l, c in enumerate(counts)))


class CostModelTests(unittest.TestCase):
    def test_no_caching_loads_every_key_every_level(self):
        cost = lab.traversal_cost(256, 10, 0)
        self.assertEqual(cost.loads, 2560)
        self.assertEqual(cost.valu, 64 * 2 * 10)

    def test_caching_trades_loads_for_selects(self):
        a, b = lab.traversal_cost(256, 10, 2), lab.traversal_cost(256, 10, 3)
        self.assertLess(b.loads, a.loads)
        self.assertGreater(b.valu, a.valu)

    def test_best_cached_levels_is_the_minimum(self):
        c, cost = lab.best_cached_levels(256, 10)
        self.assertEqual(cost.bound, min(lab.traversal_cost(256, 10, k).bound for k in range(11)))
        self.assertEqual(c, 5)


if __name__ == "__main__":
    unittest.main()