| 1 | """Linearizability checker for branch-head histories (suites/robustness)."""
|
| 2 |
|
| 3 | from __future__ import annotations
|
| 4 |
|
| 5 | import math
|
| 6 | import sys
|
| 7 | import unittest
|
| 8 | from pathlib import Path
|
| 9 |
|
| 10 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "suites" / "robustness"))
|
| 11 |
|
| 12 | from robustness.lincheck import (OK, REJECTED, UNKNOWN, Op, ancestry_from_parents, check, # noqa: E402
|
| 13 | check_partitioned, lost_acknowledged)
|
| 14 |
|
| 15 | # base <- a <- ab ; base <- b (a and b are siblings)
|
| 16 | PARENTS = {"base": (), "a": ("base",), "b": ("base",), "ab": ("a", "b")}
|
| 17 | IS_ANC = ancestry_from_parents(PARENTS)
|
| 18 |
|
| 19 |
|
| 20 | class LincheckTests(unittest.TestCase):
|
| 21 | def test_sequential_fast_forwards_linearize(self):
|
| 22 | ops = [Op("w0", "push", "a", 0, 1), Op("w1", "push", "ab", 2, 3), Op("r", "read", "ab", 4, 4)]
|
| 23 | self.assertTrue(check(ops, IS_ANC, "base").linearizable)
|
| 24 |
|
| 25 | def test_two_acknowledged_sibling_pushes_are_not_linearizable(self):
|
| 26 | # The lost-update shape: both pushes of siblings acknowledged concurrently.
|
| 27 | ops = [Op("w0", "push", "a", 0, 2), Op("w1", "push", "b", 1, 3), Op("r", "read", "b", 4, 4)]
|
| 28 | result = check(ops, IS_ANC, "base")
|
| 29 | self.assertFalse(result.linearizable)
|
| 30 | self.assertEqual([op.value for op in lost_acknowledged(ops, {"main": "b"}, IS_ANC)], ["a"])
|
| 31 |
|
| 32 | def test_rejection_has_no_effect(self):
|
| 33 | ops = [Op("w0", "push", "a", 0, 2), Op("w1", "push", "b", 1, 3, REJECTED), Op("r", "read", "a", 4, 4)]
|
| 34 | self.assertTrue(check(ops, IS_ANC, "base").linearizable)
|
| 35 |
|
| 36 | def test_unknown_outcome_may_or_may_not_apply(self):
|
| 37 | applied = [Op("w0", "push", "a", 0, math.inf, UNKNOWN), Op("r", "read", "a", 5, 5)]
|
| 38 | not_applied = [Op("w0", "push", "a", 0, math.inf, UNKNOWN), Op("r", "read", "base", 5, 5)]
|
| 39 | self.assertTrue(check(applied, IS_ANC, "base").linearizable)
|
| 40 | self.assertTrue(check(not_applied, IS_ANC, "base").linearizable)
|
| 41 |
|
| 42 | def test_real_time_order_is_respected(self):
|
| 43 | # read of "a" completes before the push of "a" starts: impossible.
|
| 44 | ops = [Op("r", "read", "a", 0, 1), Op("w0", "push", "a", 2, 3)]
|
| 45 | self.assertFalse(check(ops, IS_ANC, "base").linearizable)
|
| 46 |
|
| 47 | def test_head_regression_is_not_linearizable(self):
|
| 48 | ops = [Op("w0", "push", "ab", 0, 1), Op("w1", "push", "a", 2, 3), Op("r", "read", "a", 4, 4)]
|
| 49 | self.assertFalse(check(ops, IS_ANC, "base").linearizable)
|
| 50 |
|
| 51 | def test_partitioned_by_branch(self):
|
| 52 | ops = [Op("w0", "push", "a", 0, 1, branch="x"), Op("w1", "push", "b", 0, 1, branch="y")]
|
| 53 | results = check_partitioned(ops, IS_ANC, {"x": "base", "y": "base"})
|
| 54 | self.assertEqual({k: v.linearizable for k, v in results.items()}, {"x": True, "y": True})
|
| 55 |
|
| 56 | def test_budget_exhaustion_is_unmeasured_not_a_verdict(self):
|
| 57 | ops = [Op(f"w{i}", "push", "a", 0, 10, REJECTED) for i in range(12)] + [Op("r", "read", "zz", 11, 11)]
|
| 58 | result = check(ops, IS_ANC, "base", budget=50)
|
| 59 | self.assertIsNone(result.linearizable)
|
| 60 |
|
| 61 |
|
| 62 | if __name__ == "__main__":
|
| 63 | unittest.main()
|