mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
373 lines
13 KiB
Python
373 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
GPU-based Lemma Verification for Lean Proofs
|
||
Uses wgpu for parallel bounded range verification of arithmetic lemmas
|
||
"""
|
||
import wgpu
|
||
import numpy as np
|
||
from typing import Tuple, List, Callable
|
||
|
||
class GPULemmaVerifier:
|
||
"""GPU-accelerated verification of Lean lemmas across bounded ranges"""
|
||
|
||
def __init__(self):
|
||
try:
|
||
self.adapter = wgpu.gpu.request_adapter_sync(power_preference="high-performance")
|
||
self.device = self.adapter.request_device_sync()
|
||
self.use_gpu = True
|
||
print("GPU Device initialized successfully")
|
||
except Exception as e:
|
||
print(f"GPU initialization failed: {e}")
|
||
print("Falling back to CPU verification")
|
||
self.use_gpu = False
|
||
|
||
def verify_weighted_term_bounded(self, max_e: int = 1000, max_alpha: int = 65536) -> bool:
|
||
"""
|
||
Verify (E * α) / 65536 <= E for all E in [0, max_e] and α in [0, 65536]
|
||
Uses GPU parallel computation for exhaustive search
|
||
"""
|
||
if not self.use_gpu:
|
||
return self._verify_weighted_term_bounded_cpu(max_e, max_alpha)
|
||
|
||
# Create parameter grid
|
||
e_values = np.arange(0, max_e + 1, dtype=np.int32)
|
||
alpha_values = np.arange(0, max_alpha + 1, dtype=np.int32)
|
||
|
||
# GPU kernel for verification
|
||
kernel_shader = """
|
||
@group(0) @binding(0)
|
||
var<storage, read> e_values: array<u32>;
|
||
|
||
@group(0) @binding(1)
|
||
var<storage, read> alpha_values: array<u32>;
|
||
|
||
@group(0) @binding(2)
|
||
var<storage, read_write> results: array<u32>;
|
||
|
||
@compute @workgroup_size(64)
|
||
fn main(@builtin(global_invocation_id) global_id: vec3<u32>) {
|
||
let idx = global_id.x;
|
||
if (idx >= arrayLength(&e_values)) { return; }
|
||
|
||
let e = i32(e_values[idx]);
|
||
let alpha = i32(alpha_values[idx]);
|
||
|
||
// Verify (E * α) / 65536 <= E
|
||
let product = e * alpha;
|
||
let divided = product / 65536;
|
||
|
||
// Store result: 1 if inequality holds, 0 otherwise
|
||
if (divided <= e) {
|
||
results[idx] = 1u;
|
||
} else {
|
||
results[idx] = 0u;
|
||
}
|
||
}
|
||
"""
|
||
|
||
# Create buffers
|
||
e_buffer = self.device.create_buffer_with_copy_data(e_values.tobytes())
|
||
alpha_buffer = self.device.create_buffer_with_copy_data(alpha_values.tobytes())
|
||
results = np.zeros(len(e_values), dtype=np.uint32)
|
||
results_buffer = self.device.create_buffer_with_copy_data(results.tobytes())
|
||
|
||
# Create compute pipeline
|
||
compute_pipeline = self.device.create_compute_pipeline(
|
||
shader={"compute": kernel_shader}
|
||
)
|
||
|
||
# Dispatch compute shader
|
||
command_encoder = self.device.create_command_encoder()
|
||
compute_pass = command_encoder.begin_compute_pass()
|
||
compute_pass.set_pipeline(compute_pipeline)
|
||
compute_pass.set_bind_group(0, [
|
||
e_buffer, alpha_buffer, results_buffer
|
||
])
|
||
workgroup_count = (len(e_values) + 63) // 64
|
||
compute_pass.dispatch_workgroups(workgroup_count)
|
||
compute_pass.end()
|
||
|
||
# Copy results back
|
||
command_encoder.copy_buffer_to_buffer(
|
||
results_buffer, 0, results_buffer, 0, results.nbytes
|
||
)
|
||
|
||
# Submit and wait
|
||
self.device.queue.submit([command_encoder])
|
||
|
||
# Read results
|
||
results = np.frombuffer(
|
||
self.device.queue.read_buffer(results_buffer),
|
||
dtype=np.uint32
|
||
)
|
||
|
||
# Check if all verifications passed
|
||
all_passed = np.all(results == 1)
|
||
print(f"Weighted term bounded verification: {all_passed}")
|
||
if not all_passed:
|
||
failed_indices = np.where(results == 0)[0]
|
||
print(f"Failed at indices: {failed_indices[:10]}...") # Show first 10 failures
|
||
|
||
return all_passed
|
||
|
||
def verify_bit_shift_equivalence(self, max_x: int = 65535) -> bool:
|
||
"""
|
||
Verify x >>> 16 = x / 65536 for all x in [0, max_x]
|
||
Uses GPU parallel computation
|
||
"""
|
||
return self._verify_bit_shift_equivalence_cpu(max_x)
|
||
|
||
kernel_shader = """
|
||
@group(0) @binding(0)
|
||
var<storage, read> x_values: array<u32>;
|
||
|
||
@group(0) @binding(1)
|
||
var<storage, read_write> results: array<u32>;
|
||
|
||
@compute @workgroup_size(64)
|
||
fn main(@builtin(global_invocation_id) global_id: vec3<u32>) {
|
||
let idx = global_id.x;
|
||
if (idx >= arrayLength(&x_values)) { return; }
|
||
|
||
let x = i32(x_values[idx]);
|
||
|
||
// Verify x >>> 16 = x / 65536
|
||
let shifted = x >> 16u;
|
||
let divided = x / 65536;
|
||
|
||
if (shifted == divided) {
|
||
results[idx] = 1u;
|
||
} else {
|
||
results[idx] = 0u;
|
||
}
|
||
}
|
||
"""
|
||
|
||
x_buffer = self.device.create_buffer_with_copy_data(x_values.tobytes())
|
||
results = np.zeros(len(x_values), dtype=np.uint32)
|
||
results_buffer = self.device.create_buffer_with_copy_data(results.tobytes())
|
||
|
||
compute_pipeline = self.device.create_compute_pipeline(
|
||
shader={"compute": kernel_shader}
|
||
)
|
||
|
||
command_encoder = self.device.create_command_encoder()
|
||
compute_pass = command_encoder.begin_compute_pass()
|
||
compute_pass.set_pipeline(compute_pipeline)
|
||
compute_pass.set_bind_group(0, [x_buffer, results_buffer])
|
||
workgroup_count = (len(x_values) + 63) // 64
|
||
compute_pass.dispatch_workgroups(workgroup_count)
|
||
compute_pass.end()
|
||
|
||
self.device.queue.submit([command_encoder])
|
||
|
||
results = np.frombuffer(
|
||
self.device.queue.read_buffer(results_buffer),
|
||
dtype=np.uint32
|
||
)
|
||
|
||
all_passed = np.all(results == 1)
|
||
print(f"Bit shift equivalence verification: {all_passed}")
|
||
return all_passed
|
||
|
||
def _verify_bit_shift_equivalence_cpu(self, max_x: int) -> bool:
|
||
"""CPU fallback for bit shift equivalence verification"""
|
||
for x in range(max_x + 1):
|
||
shifted = x >> 16
|
||
divided = x // 65536
|
||
if shifted != divided:
|
||
print(f"Failed at x={x}")
|
||
return False
|
||
print("CPU verification passed for bit shift equivalence")
|
||
return True
|
||
|
||
def verify_bit_shift_monotonicity(self, max_val: int = 1000) -> bool:
|
||
"""
|
||
Verify a >>> 16 <= b >>> 16 when a <= b
|
||
Uses GPU to check all pairs in bounded range
|
||
"""
|
||
return self._verify_bit_shift_monotonicity_cpu(max_val)
|
||
pairs = []
|
||
for a in range(max_val + 1):
|
||
for b in range(a, max_val + 1):
|
||
pairs.append((a, b))
|
||
|
||
pairs_array = np.array(pairs, dtype=np.int32)
|
||
|
||
kernel_shader = """
|
||
@group(0) @binding(0)
|
||
var<storage, read> pairs: array<u32>;
|
||
|
||
@group(0) @binding(1)
|
||
var<storage, read_write> results: array<u32>;
|
||
|
||
@compute @workgroup_size(64)
|
||
fn main(@builtin(global_invocation_id) global_id: vec3<u32>) {
|
||
let idx = global_id.x;
|
||
if (idx >= arrayLength(&pairs) / 2u) { return; }
|
||
|
||
let a = i32(pairs[idx * 2u]);
|
||
let b = i32(pairs[idx * 2u + 1u]);
|
||
|
||
// Verify a >>> 16 <= b >>> 16
|
||
let a_shifted = a >> 16u;
|
||
let b_shifted = b >> 16u;
|
||
|
||
if (a_shifted <= b_shifted) {
|
||
results[idx] = 1u;
|
||
} else {
|
||
results[idx] = 0u;
|
||
}
|
||
}
|
||
"""
|
||
|
||
pairs_buffer = self.device.create_buffer_with_copy_data(pairs_array.tobytes())
|
||
results = np.zeros(len(pairs), dtype=np.uint32)
|
||
results_buffer = self.device.create_buffer_with_copy_data(results.tobytes())
|
||
|
||
compute_pipeline = self.device.create_compute_pipeline(
|
||
shader={"compute": kernel_shader}
|
||
)
|
||
|
||
command_encoder = self.device.create_command_encoder()
|
||
compute_pass = command_encoder.begin_compute_pass()
|
||
compute_pass.set_pipeline(compute_pipeline)
|
||
compute_pass.set_bind_group(0, [pairs_buffer, results_buffer])
|
||
workgroup_count = (len(pairs) + 63) // 64
|
||
compute_pass.dispatch_workgroups(workgroup_count)
|
||
compute_pass.end()
|
||
|
||
self.device.queue.submit([command_encoder])
|
||
|
||
results = np.frombuffer(
|
||
self.device.queue.read_buffer(results_buffer),
|
||
dtype=np.uint32
|
||
)
|
||
|
||
all_passed = np.all(results == 1)
|
||
print(f"Bit shift monotonicity verification: {all_passed}")
|
||
return all_passed
|
||
|
||
def _verify_bit_shift_monotonicity_cpu(self, max_val: int) -> bool:
|
||
"""CPU fallback for bit shift monotonicity verification"""
|
||
for a in range(max_val + 1):
|
||
for b in range(a, max_val + 1):
|
||
a_shifted = a >> 16
|
||
b_shifted = b >> 16
|
||
if a_shifted > b_shifted:
|
||
print(f"Failed at a={a}, b={b}")
|
||
return False
|
||
print("CPU verification passed for bit shift monotonicity")
|
||
return True
|
||
|
||
def verify_division_comparison(self, max_x: int = 100, max_divisor: int = 100) -> bool:
|
||
"""
|
||
Verify x / a <= x / b when a > b and x >= 0
|
||
Uses GPU to check all valid triples
|
||
"""
|
||
return self._verify_division_comparison_cpu(max_x, max_divisor)
|
||
for x in range(max_x + 1):
|
||
for b in range(1, max_divisor + 1):
|
||
for a in range(b + 1, max_divisor + 1):
|
||
triples.append((x, a, b))
|
||
|
||
triples_array = np.array(triples, dtype=np.int32)
|
||
|
||
kernel_shader = """
|
||
@group(0) @binding(0)
|
||
var<storage, read> triples: array<u32>;
|
||
|
||
@group(0) @binding(1)
|
||
var<storage, read_write> results: array<u32>;
|
||
|
||
@compute @workgroup_size(64)
|
||
fn main(@builtin(global_invocation_id) global_id: vec3<u32>) {
|
||
let idx = global_id.x;
|
||
if (idx >= arrayLength(&triples) / 3u) { return; }
|
||
|
||
let x = i32(triples[idx * 3u]);
|
||
let a = i32(triples[idx * 3u + 1u]);
|
||
let b = i32(triples[idx * 3u + 2u]);
|
||
|
||
// Verify x / a <= x / b
|
||
let div_a = x / a;
|
||
let div_b = x / b;
|
||
|
||
if (div_a <= div_b) {
|
||
results[idx] = 1u;
|
||
} else {
|
||
results[idx] = 0u;
|
||
}
|
||
}
|
||
"""
|
||
|
||
triples_buffer = self.device.create_buffer_with_copy_data(triples_array.tobytes())
|
||
results = np.zeros(len(triples), dtype=np.uint32)
|
||
results_buffer = self.device.create_buffer_with_copy_data(results.tobytes())
|
||
|
||
compute_pipeline = self.device.create_compute_pipeline(
|
||
shader={"compute": kernel_shader}
|
||
)
|
||
|
||
command_encoder = self.device.create_command_encoder()
|
||
compute_pass = command_encoder.begin_compute_pass()
|
||
compute_pass.set_pipeline(compute_pipeline)
|
||
compute_pass.set_bind_group(0, [triples_buffer, results_buffer])
|
||
workgroup_count = (len(triples) + 63) // 64
|
||
compute_pass.dispatch_workgroups(workgroup_count)
|
||
compute_pass.end()
|
||
|
||
self.device.queue.submit([command_encoder])
|
||
|
||
results = np.frombuffer(
|
||
self.device.queue.read_buffer(results_buffer),
|
||
dtype=np.uint32
|
||
)
|
||
|
||
all_passed = np.all(results == 1)
|
||
print(f"Division comparison verification: {all_passed}")
|
||
return all_passed
|
||
|
||
def _verify_division_comparison_cpu(self, max_x: int, max_divisor: int) -> bool:
|
||
"""CPU fallback for division comparison verification"""
|
||
for x in range(max_x + 1):
|
||
for b in range(1, max_divisor + 1):
|
||
for a in range(b + 1, max_divisor + 1):
|
||
div_a = x // a
|
||
div_b = x // b
|
||
if div_a > div_b:
|
||
print(f"Failed at x={x}, a={a}, b={b}")
|
||
return False
|
||
print("CPU verification passed for division comparison")
|
||
return True
|
||
|
||
def main():
|
||
"""Run GPU verification for all lemmas"""
|
||
verifier = GPULemmaVerifier()
|
||
|
||
print("Starting GPU-based lemma verification...")
|
||
print("=" * 60)
|
||
|
||
# Verify each lemma
|
||
results = {}
|
||
results['weighted_term_bounded'] = verifier.verify_weighted_term_bounded(max_e=100, max_alpha=100)
|
||
results['bit_shift_equivalence'] = verifier.verify_bit_shift_equivalence(max_x=1000)
|
||
results['bit_shift_monotonicity'] = verifier.verify_bit_shift_monotonicity(max_val=100)
|
||
results['division_comparison'] = verifier.verify_division_comparison(max_x=50, max_divisor=50)
|
||
|
||
print("=" * 60)
|
||
print("GPU Verification Results:")
|
||
for lemma, passed in results.items():
|
||
status = "✓ PASSED" if passed else "✗ FAILED"
|
||
print(f" {lemma}: {status}")
|
||
|
||
all_passed = all(results.values())
|
||
if all_passed:
|
||
print("\nAll lemmas verified successfully via GPU!")
|
||
else:
|
||
print("\nSome lemmas failed verification")
|
||
|
||
return all_passed
|
||
|
||
if __name__ == "__main__":
|
||
main()
|