check_independent_store_reuse.py

#!/usr/bin/env python3
"""Finite semantic audit of bounded per-source independent-store reuse.

The local producer is the separately audited scanned-omission oracle. This
checker independently compares grouped input, lane scheduling and final OR
with direct literal relation composition. Python maps, sets, copied snapshots
and numerical codes are semantic oracles, not a native execution or timing test.
"""
from copy import deepcopy
from random import Random
from collections import Counter
import json
from check_scanned_omission_reuse import Store


class Invalid(ValueError):
    pass


def word(value, width):
    return format(value, '0%db' % width)


def valid_word(value):
    return bool(value) and set(value) <= {'0', '1'}


def encode_store(name):
    if not valid_word(name):
        raise Invalid('store name')
    return ''.join('0' + bit for bit in name) + '11'


def split_middle(encoded):
    """Decode E(b)v, returning the ordinary store and alias spellings."""
    name, pos = [], 0
    while pos + 1 < len(encoded):
        pair = encoded[pos:pos + 2]
        pos += 2
        if pair == '11':
            alias = encoded[pos:]
            if not name or not valid_word(alias):
                raise Invalid('middle spelling')
            return ''.join(name), alias
        if pair not in ('00', '01'):
            raise Invalid('store encoding')
        name.append(pair[1])
    raise Invalid('unterminated store encoding')


class Joined:
    def __init__(self, k, p):
        self.k, self.p = k, p
        self.stores, self.relation, self.sources = {}, {}, {}
        self.current = None
        self.frozen = False
        self.selected = {}
        self.metrics = Counter()

    def right(self, b, v, w):
        assert not self.frozen
        if not all(valid_word(x) for x in (b, v, w)):
            raise Invalid('right spelling')
        assert split_middle(encode_store(b) + v) == (b, v)
        if b != self.current:
            if b in self.stores:
                raise Invalid('regroup')
            if self.current is not None and not self.stores[self.current].gate():
                raise Invalid('previous local final gate')
            self.stores[b] = Store(self.k, 1 << len(v), len(w), len(v))
            self.relation[b] = {}
            self.current = b
        store = self.stores[b]
        if len(v) != store.ell or len(w) != store.n:
            raise Invalid('right local width')
        store.step(int(v, 2), int(w, 2))
        self.relation[b].setdefault(v, set()).add(w)
        self.metrics['right_records'] += 1
        self.metrics['encoded_store_bits'] += len(encode_store(b))

    def freeze(self):
        if self.current is None or not self.stores[self.current].gate():
            raise Invalid('last local final gate')
        self.frozen = True
        # Local identical code values need not identify identical literals.
        code_owner = {}
        for b, store in self.stores.items():
            for code, target in store.inverse.items():
                literal = word(target, store.n)
                if code in code_owner and code_owner[code] != literal:
                    self.metrics['cross_store_code_collisions'] += 1
                code_owner[code] = literal

    def left(self, source, b, v):
        assert self.frozen
        if not valid_word(source):
            raise Invalid('source spelling')
        b, v = split_middle(encode_store(b) + v)
        if b not in self.stores:
            raise Invalid('unknown store')
        store = self.stores[b]
        if len(v) != store.ell:
            raise Invalid('left local width')
        lanes = self.sources.setdefault(source, {})
        if self.p is not None and b not in lanes and len(lanes) == self.p:
            raise Invalid('source dependency cap')
        before = deepcopy(lanes)
        lane = lanes.setdefault(b, dict(committed=None, pending=None))
        if lane['pending'] is not None:
            holes = store.holes(lane['pending'])
            lane['committed'] = holes if lane['committed'] is None else lane['committed'] & holes
            # This models service ownership, not an instruction count.
            assert store.m - store.d <= len(v)
            self.metrics['lane_jobs_completed'] += 1
        alias = int(v, 2)
        lane['pending'] = alias if alias in store.rows else None
        for other, old_lane in before.items():
            if other != b:
                assert lanes[other] == old_lane
                self.metrics['other_lanes_unchanged'] += 1
                if old_lane['pending'] is not None and self.stores[other].ell > len(v):
                    self.metrics['long_jobs_retained_across_short_aliases'] += 1
        self.selected.setdefault(source, []).append((b, v))
        self.metrics['left_records'] += 1
        self.metrics['encoded_store_bits'] += len(encode_store(b))

    def query(self, source, target):
        assert self.frozen and valid_word(target)
        found = False
        for b, lane in self.sources.get(source, {}).items():
            store = self.stores[b]
            if len(target) != store.n:
                continue
            t = int(target, 2)
            B = lane['committed']
            if B is not None and t in store.births and all(t != store.inverse[c] for c in B):
                found = True
            pending = lane['pending']
            if pending is not None:
                row = store.rows[pending]
                found |= t in row['R'] or t == row['p']
        literal_union = set()
        for b, v in self.selected.get(source, []):
            literal_union |= self.relation[b].get(v, set())
        assert (not found) == (target not in literal_union)
        self.metrics['negative_queries'] += 1
        return not found


class Routed(Joined):
    """Unbounded source lanes; a target's actual E(b) selects one lane."""
    def __init__(self, k):
        super().__init__(k, None)

    def right(self, b, v, encoded_target):
        target_store, local = split_middle(encoded_target)
        if target_store != b:
            raise Invalid('right target namespace mismatch')
        super().right(b, v, local)
        self.metrics['encoded_store_bits'] += len(encode_store(b))

    def query(self, source, encoded_target):
        b, target = split_middle(encoded_target)
        lane = self.sources.get(source, {}).get(b)
        found = False
        if lane is not None:
            store = self.stores[b]
            if len(target) == store.n:
                t = int(target, 2)
                B = lane['committed']
                found = B is not None and t in store.births and all(
                    t != store.inverse[c] for c in B)
                if lane['pending'] is not None:
                    row = store.rows[lane['pending']]
                    found |= t in row['R'] or t == row['p']
        literal_union = set()
        for selected_b, v in self.selected.get(source, []):
            if selected_b == b:
                literal_union |= self.relation[b].get(v, set())
        assert (not found) == (target not in literal_union)
        self.metrics['routed_negative_queries'] += 1
        return not found


def audit():
    total = Counter()
    def rejects(action, reason):
        try:
            action()
        except Invalid as error:
            assert str(error) == reason, (str(error), reason)
            total['rejections_checked'] += 1
        else:
            raise AssertionError('Expected rejection: ' + reason)
    for malformed in ('11' + '0', '10' + '110', '00', '0011', '00112'):
        try:
            split_middle(malformed)
        except Invalid:
            total['malformed_middle_checks'] += 1
        else:
            raise AssertionError(malformed)
    for k in (1, 2, 3):
        # Different widths, overlapping/disjoint supports, and identical local
        # code numbers with different meanings. There are more global stores
        # than any one source is allowed to mention.
        for seed in range(20):
            rng = Random(k * 100 + seed)
            joined = Joined(k, 3)
            specs = [('0', 12, 2), ('00', 1, 9), ('101', 3, 3),
                     ('1' * 25, 2, 2), ('1110', 3, 4)]
            for b, ell, n in specs:
                targets = list(range(min(1 << n, 6)))
                aliases = list(range(min(1 << ell, 3)))
                for a in aliases:
                    # Each row is locally dense but need not be globally dense.
                    omitted = {a % len(targets)} if a else set()
                    for t in targets:
                        if t not in omitted:
                            joined.right(b, word(a, ell), word(t, n))
                    joined.right(b, word(0, ell), word(0, n))
            joined.freeze()
            snapshots = []
            sources = ['0', '1', '00101']
            choices = {s: rng.sample(specs, 3) for s in sources}
            # Deterministic cross-width pending-job stress precedes randomness.
            for b, ell, _ in specs[:2] * 3:
                joined.left('111', b, word(0, ell))
            for _ in range(80):
                source = rng.choice(sources)
                b, ell, _ = rng.choice(choices[source])
                alias = rng.randrange(min(1 << ell, 4))
                joined.left(source, b, word(alias, ell))
                queries = {'0', '1', '00000', '1' * 13}
                queries |= {w for rows in joined.relation.values() for row in rows.values() for w in row}
                for q in queries:
                    joined.query(source, q)
                    joined.query('0101010101', q)
                if len(snapshots) < 4:
                    snapshots.append(deepcopy(joined))
            for old in snapshots:
                for s in sources + ['111']:
                    for q in ('0', '00', '000', '000000000', '11111'):
                        old.query(s, q)
                total['retained_join_snapshots'] += 1
            # Missing row still counts as a distinct store mention.
            cap = deepcopy(joined)
            for b, ell, _ in specs[:3]:
                cap.left('1111110', b, word((1 << ell) - 1, ell))
            rejects(lambda: cap.left('1111110', specs[3][0], '11'), 'source dependency cap')
            rejects(lambda: joined.left('0', '10101010101', '0'), 'unknown store')
            rejects(lambda: joined.left('0', '0', '0'), 'left local width')
            total.update(joined.metrics)
            total['histories'] += 1
        # One source selects far more than p=3 stores. Query routing, not
        # global OR, distinguishes equal local targets in different namespaces.
        routed = Routed(k)
        for index in range(17):
            b = word(index, 5)
            ell, n = (1 + index % 4), (1 + index % 5)
            for target in range(min(1 << n, 4)):
                routed.right(b, word(0, ell), encode_store(b) + word(target, n))
        routed.freeze()
        for index in range(17):
            b = word(index, 5)
            ell = 1 + index % 4
            routed.left('010', b, word(0, ell))
            routed.left('010', b, word(0, ell))
            for query_index in range(18):
                qb = word(query_index, 5)
                for q in ('0', '00', '000', '0000', '00000', '111111'):
                    routed.query('010', encode_store(qb) + q)
                    routed.query('0010', encode_store(qb) + q)
        assert len(routed.sources['010']) == 17
        total['routed_histories'] += 1
        total['routed_max_source_dependencies'] = max(
            total['routed_max_source_dependencies'], len(routed.sources['010']))
        total.update(routed.metrics)
        rejects(lambda: Routed(k).right('0', '0', encode_store('1') + '0'),
                'right target namespace mismatch')
        regroup = Joined(k, 1)
        regroup.right('0', '0', '0')
        regroup.right('1', '0', '1')
        rejects(lambda: regroup.right('0', '0', '0'), 'regroup')
        bad = Joined(k, 1)
        for t in range(k + 3):
            bad.right('0', '0', word(t, 4))
        bad.right('0', '1', '0000')
        rejects(bad.freeze, 'last local final gate')
        rejects(lambda: bad.right('1', '0', '0'), 'previous local final gate')
    for required in ('cross_store_code_collisions', 'long_jobs_retained_across_short_aliases',
                     'other_lanes_unchanged', 'lane_jobs_completed'):
        assert total[required] > 0
    return dict(total)


if __name__ == '__main__':
    print(json.dumps({'status': 'pass', 'scope': 'finite semantics and lane ownership; not native timing',
                      'checks': audit()}, indent=2, sort_keys=True))