check_cosingleton_reuse.py

#!/usr/bin/env python3
"""Finite oracle for co-singleton reuse; not a native-cost implementation.

Python sets, maps, integer arithmetic and copied snapshots are oracle machinery.
The written proof separately charges literal routing and all local operations.
"""
from itertools import product
from random import Random


def signature(row):
    xor = 0
    for value in row:
        xor ^= value
    return len(row), xor


class Oracle:
    def __init__(self, alias_count):
        self.rows = [set() for _ in range(alias_count)]
        self.classes = {}
        self.codes = {}
        self.free = set(range(alias_count))
        self.histogram = {}
        self.minimum = 0
        self.support = set()
        self.full_pool_reuses = 0

    def snapshot(self):
        return (tuple(frozenset(row) for row in self.rows),
                tuple(sorted(self.codes.items())),
                tuple(sorted(self.classes.items())))

    def insert(self, alias, target):
        row = self.rows[alias]
        if target in row:
            return
        degree = len(row)
        old_signature = signature(row)
        new_signature = signature(row | {target})
        candidate = min(self.free) if self.free else self.codes.get(alias)
        was_full = not self.free
        new_class = new_signature not in self.classes
        if degree:
            population, old_code = self.classes[old_signature]
            if population == 1:
                del self.classes[old_signature]
                self.free.add(old_code)
            else:
                self.classes[old_signature] = population - 1, old_code
            self.histogram[degree] -= 1
            if not self.histogram[degree]:
                del self.histogram[degree]
        row.add(target)
        if new_signature in self.classes:
            population, code = self.classes[new_signature]
            self.classes[new_signature] = population + 1, code
        else:
            assert candidate is not None and candidate in self.free
            code = candidate
            self.free.remove(code)
            self.classes[new_signature] = 1, code
            if was_full:
                assert degree and new_class
                self.full_pool_reuses += 1
        self.codes[alias] = code
        self.histogram[degree + 1] = self.histogram.get(degree + 1, 0) + 1
        if degree == 0:
            self.minimum = 1
        elif degree == self.minimum and degree not in self.histogram:
            self.minimum += 1
        self.support.add(target)
        self.check()

    def gate(self):
        return self.minimum >= len(self.support) - 1

    def check(self):
        assert self.minimum == min((len(row) for row in self.rows if row), default=0)
        degrees = {len(row) for row in self.rows if row}
        assert self.histogram == {d: sum(len(row) == d for row in self.rows) for d in degrees}
        expected = {}
        for alias, row in enumerate(self.rows):
            if row:
                sig = signature(row)
                expected[sig] = expected.get(sig, 0) + 1
                assert self.codes[alias] == self.classes[sig][1]
        assert expected == {sig: pop for sig, (pop, code) in self.classes.items()}
        live_codes = {code for pop, code in self.classes.values()}
        assert len(live_codes) == len(self.classes)
        assert self.free == set(range(len(self.rows))) - live_codes
        assert self.gate() == all(len(self.support - row) <= 1 for row in self.rows if row)


def check_snapshot(snapshot):
    rows, codes, classes = snapshot
    codes, classes = dict(codes), dict(classes)
    for alias, row in enumerate(rows):
        if row:
            assert codes[alias] == classes[signature(row)][1]
    for sig, (population, code) in classes.items():
        assert population == sum(signature(row) == sig for row in rows if row)


def pending_queries(oracle, selections, width):
    """Commit the previous pending class during the next selection."""
    assert oracle.gate()
    committed = None
    pending = None
    full = False
    actual = set()
    checks = 0
    for alias in selections:
        if pending is not None and not full:
            if committed is None:
                committed = pending
            elif oracle.codes[committed] != oracle.codes[pending]:
                full = True
        pending = alias if alias < len(oracle.rows) and oracle.rows[alias] else None
        if pending is not None:
            actual |= oracle.rows[pending]
        predicted = set(oracle.support) if full else set()
        if not full and committed is not None:
            predicted |= oracle.rows[committed]
        if pending is not None:
            predicted |= oracle.rows[pending]
        for query_width in (width - 1, width, width + 1):
            for target in range(1 << width):
                valid = query_width == width
                # Exact negative observation is SAT under the ordinary pins.
                assert (not (valid and target in predicted)) == (not (valid and target in actual))
                checks += 1
    return checks


def exhaustive():
    subsets = [{j for j in range(4) if mask >> j & 1} for mask in range(16)]
    states = pairs = 0
    for rows in product(subsets, repeat=4):
        support = set().union(*rows)
        observed = [row for row in rows if row]
        if any(len(support - row) > 1 for row in observed):
            continue
        states += 1
        for left in observed:
            for right in observed:
                assert (signature(left) == signature(right)) == (left == right)
                if signature(left) != signature(right):
                    assert left | right == support
                pairs += 1
    # The zero omission collides in XOR with FULL, but not in count.
    assert signature({0, 1, 2, 3})[1] == signature({1, 2, 3})[1]
    assert signature({0, 1, 2, 3}) != signature({1, 2, 3})
    return states, pairs


def streaming():
    steps = snapshots = queries = recoveries = reuses = 0
    for seed in range(400):
        rng = Random(seed)
        oracle = Oracle(8)
        retained = []
        previous_gate = True
        # Arbitrary interleaving, then complete every row to guarantee recovery.
        records = [(rng.randrange(8), rng.randrange(16)) for _ in range(100)]
        records += [(a, t) for a in range(8) for t in range(16)]
        for alias, target in records:
            retained.append(oracle.snapshot())
            oracle.insert(alias, target)
            steps += 1
            if oracle.gate() and not previous_gate:
                recoveries += 1
            previous_gate = oracle.gate()
            if oracle.gate():
                selections = [rng.randrange(9) for _ in range(6)]
                queries += pending_queries(oracle, selections, 4)
        for snapshot in retained:
            check_snapshot(snapshot)
            snapshots += 1
        reuses += oracle.full_pool_reuses
    # Explicitly saturate the slot pool before replacing a singleton class.
    oracle = Oracle(8)
    for alias in range(8):
        oracle.insert(alias, alias)
    snapshot = oracle.snapshot()
    oracle.insert(0, 1)
    assert oracle.full_pool_reuses == 1
    oracle.insert(1, 0)  # Join an extant signature class and release a slot.
    check_snapshot(snapshot)
    assert oracle.codes[0] == oracle.codes[1]
    return steps, snapshots, queries, recoveries, reuses + 1


def main():
    states, pairs = exhaustive()
    steps, snapshots, queries, recoveries, reuses = streaming()
    print({"gated_states": states, "row_pairs": pairs,
           "stream_records": steps, "retained_snapshots": snapshots + 1,
           "negative_query_checks": queries, "gate_recoveries": recoveries,
           "full_pool_reuses": reuses})


if __name__ == "__main__":
    main()