From 1f8faacd65b5dc85192d05b68bd75a780962c4a6 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Tue, 6 Oct 2026 11:47:40 +0100 Subject: Batch: filters, if !L.has(k) { a line appending } in a batch's block or in a walk's spread items (PLAN_FILTER; L only grown there, one filter a list, none in a nested LOOP): each round asked as the batch opens, its code run in the rounds it said no to, those rounds kept (k and the appends' cells) and asked again at the end in the rounds' order (L as it is and what the kept rounds before add), the rest added at once (pure_list_add_many) once the lists take them and the memory budget all of them (else, or a key added there already, the batch gives its rounds back); a list such a filter grows read in the block only as no earlier round's item would change (its count in the guard, its first, a place when it has no key, a field by key). Keys kept as bits (pure_list_bits) asked by a loop of lanes, fields of a DENSE list found at their keys (pure_list_dense); an outer lane's affine value in four bytes. The suite's BFS batched: PURE_JIT=0 93 to 39.6 ms (the compiled code 7.7, C 10). module_tier's walks filter d (seeds 1, 2, 3, 5, 23 alike: 5,806 filters decided in seed 23's); GRAPH_SOURCE: a graph built by a filter, searched twice, marks whose filter adds another key (EXISTS), under work and memory budgets --- tests/check.py | 120 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 118 insertions(+), 2 deletions(-) (limited to 'tests') diff --git a/tests/check.py b/tests/check.py index 731c5f8..ddb4a7a 100644 --- a/tests/check.py +++ b/tests/check.py @@ -1390,6 +1390,96 @@ on go { } """ +# Filters in a batch's block (GRAPH_SOURCE): a graph built in the handler +# (an edge added if not there yet, keys from a recurrence), searched breadth +# first twice (a count's loop to the queue's count, each node's edges +# walked and spread, if !seen.has(e.to) { seen.add; queue.append }: asked +# as the batch opens and again in the rounds' order at its end), the depths +# summed by a walk; marks whose filter asks one key and adds another (one +# there already: EXISTS at its round); under work budgets that end within a +# batch and memory budgets that end in the filters' adds. +GRAPH_SOURCE = """record Edge { + link = 0 + to = 0 +} +record Seen { + node = 0 + depth = 0 +} +record Step { + node = 0 +} +state reached = 0 +state hops = -1 +state far = 0 +state marked = 0 +input go = message { + n: u32 + m: u32 + from: u32 + q: u32 +} +on go { + let edges: list of Edge by link + let seen: list of Seen by node + let queue: list of Step + let again: list of Seen by node + let next: list of Step + let marks: list of Seen by node + let order: list of Step + let x = 7 + for k in 0 .. go.m { + x = x * 48271 - floor(x * 48271 / 2147483647) * 2147483647 + let a = x - floor(x / go.n) * go.n + let b = floor(x / 1000) - floor(x / 1000 / go.n) * go.n + if !edges.has(a * 1048576 + b) { edges.add(Edge { link: a * 1048576 + b, to: b }) } + } + queue.append(Step { node: go.from }) + seen.add(Seen { node: go.from }) + for i in 0 .. go.n { + if i >= queue.count { break } + let q = queue.first + i + for e in edges[q.node * 1048576 .. q.node * 1048576 + 1048576] { + if !seen.has(e.to) { + seen.add(Seen { node: e.to, depth: seen[q.node].depth + 1 }) + queue.append(Step { node: e.to }) + } + } + } + next.append(Step { node: go.q }) + again.add(Seen { node: go.q, depth: 1 }) + for i in 0 .. go.n * 2 { + if i >= next.count { break } + let q = next.first + i + for e in edges[q.node * 1048576 + 1 .. q.node * 1048576 + 1048576] { + if !again.has(e.to) { + again.add(Seen { node: e.to, depth: again[q.node].depth * 2 + 0.5 }) + next.append(Step { node: e.to }) + } + } + } + let total = 0 + for s in seen { + total = total + s.depth * (s.node + 1) + } + for s in again { + total = total + s.depth + } + reached = seen.count * 1000 + again.count + hops = if seen.has(go.q) { seen[go.q].depth } else { -1 } + far = total + for v in 0 .. go.n { + order.append(Step { node: v * 7 - floor(v * 7 / go.n) * go.n }) + } + for i in 0 .. go.n { + if i >= order.count { break } + let p = order.first + i + if !marks.has(p.node) { marks.add(Seen { node: p.node * 3 - floor(p.node * 3 / (go.n + go.q)) * (go.n + go.q), depth: i }) } + } + marked = marks.count +} +""" + # Items a batch appends at once (APPEND_SOURCE): keys there already, met # at round q of a DENSE list and at round p of a HASHED one (the # executor's fault at that round); keys past what a DENSE list holds from @@ -1513,6 +1603,20 @@ def module_assumptions(aup, pure, work): assert outputs[name] == outputs['batches'], ('front', name, next((a, b) for a, b in zip(outputs[name].splitlines(), outputs['batches'].splitlines()) if a != b)) faults = [json.loads(line)['fault'] for line in outputs['batches'].splitlines() if line.startswith('{"states"')] assert 'divide' in faults and 'none' in faults and 'work' in faults, faults + source, program = work / 'graph.aup', work / 'graph.pmod' # filters, against the rounds one by one + source.write_text(GRAPH_SOURCE) + compile_module(aup, source, program) + given = ''.join(f'message go {struct.pack("