oak/benchmarks

"""Linearizability checker for branch-head histories (suites/robustness)."""

from __future__ import annotations

import math

import sys

import unittest

from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "suites" / "robustness"))

from robustness.lincheck import (OK, REJECTED, UNKNOWN, Op, ancestry_from_parents, check,  # noqa: E402

                                 check_partitioned, lost_acknowledged)

# base <- a <- ab ; base <- b (a and b are siblings)

PARENTS = {"base": (), "a": ("base",), "b": ("base",), "ab": ("a", "b")}

IS_ANC = ancestry_from_parents(PARENTS)

class LincheckTests(unittest.TestCase):

    def test_sequential_fast_forwards_linearize(self):

        ops = [Op("w0", "push", "a", 0, 1), Op("w1", "push", "ab", 2, 3), Op("r", "read", "ab", 4, 4)]

        self.assertTrue(check(ops, IS_ANC, "base").linearizable)

    def test_two_acknowledged_sibling_pushes_are_not_linearizable(self):

        # The lost-update shape: both pushes of siblings acknowledged concurrently.

        ops = [Op("w0", "push", "a", 0, 2), Op("w1", "push", "b", 1, 3), Op("r", "read", "b", 4, 4)]

        result = check(ops, IS_ANC, "base")

        self.assertFalse(result.linearizable)

        self.assertEqual([op.value for op in lost_acknowledged(ops, {"main": "b"}, IS_ANC)], ["a"])

    def test_rejection_has_no_effect(self):

        ops = [Op("w0", "push", "a", 0, 2), Op("w1", "push", "b", 1, 3, REJECTED), Op("r", "read", "a", 4, 4)]

        self.assertTrue(check(ops, IS_ANC, "base").linearizable)

    def test_unknown_outcome_may_or_may_not_apply(self):

        applied = [Op("w0", "push", "a", 0, math.inf, UNKNOWN), Op("r", "read", "a", 5, 5)]

        not_applied = [Op("w0", "push", "a", 0, math.inf, UNKNOWN), Op("r", "read", "base", 5, 5)]

        self.assertTrue(check(applied, IS_ANC, "base").linearizable)

        self.assertTrue(check(not_applied, IS_ANC, "base").linearizable)

    def test_real_time_order_is_respected(self):

        # read of "a" completes before the push of "a" starts: impossible.

        ops = [Op("r", "read", "a", 0, 1), Op("w0", "push", "a", 2, 3)]

        self.assertFalse(check(ops, IS_ANC, "base").linearizable)

    def test_head_regression_is_not_linearizable(self):

        ops = [Op("w0", "push", "ab", 0, 1), Op("w1", "push", "a", 2, 3), Op("r", "read", "a", 4, 4)]

        self.assertFalse(check(ops, IS_ANC, "base").linearizable)

    def test_partitioned_by_branch(self):

        ops = [Op("w0", "push", "a", 0, 1, branch="x"), Op("w1", "push", "b", 0, 1, branch="y")]

        results = check_partitioned(ops, IS_ANC, {"x": "base", "y": "base"})

        self.assertEqual({k: v.linearizable for k, v in results.items()}, {"x": True, "y": True})

    def test_budget_exhaustion_is_unmeasured_not_a_verdict(self):

        ops = [Op(f"w{i}", "push", "a", 0, 10, REJECTED) for i in range(12)] + [Op("r", "read", "zz", 11, 11)]

        result = check(ops, IS_ANC, "base", budget=50)

        self.assertIsNone(result.linearizable)

if __name__ == "__main__":

    unittest.main()