diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml new file mode 100644 index 0000000..a93a456 --- /dev/null +++ b/.github/workflows/python-tests.yml @@ -0,0 +1,48 @@ +name: Python Tests + +on: + push: + branches: [main, corrina, dev] + paths: + - 'driver/python/**' + - 'test/quantum/**' + - 'api_server.py' + - 'pyproject.toml' + - 'requirements.txt' + pull_request: + branches: [main, corrina] + paths: + - 'driver/python/**' + - 'test/quantum/**' + - 'api_server.py' + - 'pyproject.toml' + - 'requirements.txt' + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.11", "3.12"] + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + + - name: Run quantum driver tests + run: | + python -m pytest test/quantum/ -v --tb=short + + - name: Run demo script + run: | + python demo_accl_q.py diff --git a/INSTALL.md b/INSTALL.md index 50da906..e285121 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -1,12 +1,42 @@ -# Installation instructions +# Installation Instructions -## Pull project +## ACCL-Q Python Driver (Quick Start) + +```sh +git clone https://github.com/The-AI-Cowboys-Projects/ACCL_NEW.git +cd ACCL_NEW + +# Install Python dependencies +pip install -r requirements.txt + +# Run the demo +python demo_accl_q.py + +# Run the test suite +python -m pytest test/quantum/ -v + +# Start the API server +python -m uvicorn api_server:app --host 0.0.0.0 --port 8080 +``` + +### Requirements + +- Python 3.11+ +- numpy >= 1.24 +- fastapi, uvicorn, pydantic (for API server) +- pytest, pytest-asyncio, httpx (for tests) + +--- + +## ACCL Hardware Build (Original) + +### Pull project ```sh git clone https://github.com/Xilinx/ACCL.git git submodule update --init --recursive ``` -## Install dependencies +### Install dependencies The project has been tested with Xilinx Vitis 2022.1 on Ubuntu 20.04. ```sh sudo apt update diff --git a/README.md b/README.md index 3ff8dad..7faf48f 100644 --- a/README.md +++ b/README.md @@ -133,7 +133,8 @@ This demonstrates sufficient margin for multi-round QEC within coherence limits. │ Operation Modes │ │ ├── STANDARD - Default operation │ │ ├── DETERMINISTIC - Hardware-synchronized, minimal jitter │ -│ └── LOW_LATENCY - Optimized for speed over consistency │ +│ ├── LOW_LATENCY - Optimized for speed over consistency │ +│ └── ULTRA_LOW_LATENCY - Hardware-autonomous sub-50ns feedback │ ├─────────────────────────────────────────────────────────────────┤ │ Infrastructure │ │ └── IBM Cloud Code Engine (Serverless Container) │ @@ -181,6 +182,27 @@ curl -X POST ".../qec/syndrome" \ -d '{"num_ranks": 8, "syndrome_bits": 4}' ``` +### ULL Pipeline + +```bash +# Configure ULL pipeline +curl -X POST ".../ull/configure" \ + -H "Content-Type: application/json" \ + -d '{"syndrome_bits": 16, "coherence_time_us": 50.0}' + +# Run autonomous feedback cycle +curl -X POST ".../ull/feedback?num_cycles=1" + +# Run 100 continuous cycles +curl -X POST ".../ull/feedback?num_cycles=100" + +# Check ULL status +curl ".../ull/status" + +# Disarm pipeline +curl -X POST ".../ull/disarm" +``` + ### Qubit Emulator ```bash @@ -266,23 +288,34 @@ docker run -p 8080:8080 accl-q ``` ACCL_NEW/ -├── api_server.py # FastAPI REST API server -├── demo_accl_q.py # Comprehensive demo script +├── api_server.py # FastAPI REST API (includes ULL endpoints) +├── demo_accl_q.py # Comprehensive demo (6 demos incl. ULL) +├── pyproject.toml # Python packaging configuration +├── requirements.txt # Python dependencies ├── Dockerfile # Production container definition -├── .dockerignore # Docker build exclusions ├── driver/ │ └── python/ │ └── accl_quantum/ # Core ACCL-Q driver │ ├── __init__.py # Package exports │ ├── driver.py # ACCLQuantum main class +│ ├── constants.py # Enums, ULL config, latency budgets +│ ├── hardware_accel.py # DMA pool, LUT decoder, FPGA regs +│ ├── feedback.py # Feedback pipelines (std + ULL) │ ├── emulator.py # RealisticQubitEmulator -│ ├── feedback.py # MeasurementFeedbackPipeline +│ ├── profiler.py # Critical path profiler │ ├── stats.py # LatencyMonitor -│ └── constants.py # Enums and configuration +│ ├── deployment.py # Multi-board RFSoC deployment +│ ├── integrations.py # QubiC/QICK integrations +│ └── docs/ # Documentation +├── kernels/cclo/hls/quantum/ # HLS constants +│ └── quantum_hls_constants.h ├── test/ -│ └── quantum/ # Test suite +│ └── quantum/ # Test suite (~200 tests) │ ├── test_collective_ops.py │ ├── test_integration.py +│ ├── test_ull_optimization.py +│ ├── test_ull_latency_validation.py +│ ├── test_module_coverage.py │ └── test_latency_validation.py └── README.md # This file ``` @@ -303,6 +336,38 @@ This experimental deployment validates that FPGA-based collective communication --- +## Ultra-Low-Latency (ULL) Mode + +ACCL-Q v0.3.0 introduces ULL mode for hardware-autonomous feedback execution targeting **<50ns latency** (0.1% of 50us coherence time) — a 10x improvement over standard feedback. + +| Component | Standard | ULL Target | +|-----------|----------|-----------| +| Multicast | 300ns | 10ns | +| Reduce | 400ns | 4ns | +| Decode | 50-200ns | 8ns | +| Trigger | 50ns | 2ns | +| **Total** | **~500ns** | **~34ns** | + +```python +from accl_quantum import ACCLQuantum, ACCLMode +from accl_quantum.feedback import HardwareFeedbackEngine +from accl_quantum.constants import ULLPipelineConfig + +# Zero-copy ULL collectives +accl = ACCLQuantum(num_ranks=4, local_rank=0) +accl.configure(mode=ACCLMode.ULTRA_LOW_LATENCY) +result = accl.broadcast(data, root=0) # 10ns, zero-copy + +# Autonomous hardware feedback +engine = HardwareFeedbackEngine(ULLPipelineConfig()) +engine.program_pipeline(decoder_fn=my_decoder, syndrome_bits=16) +result = engine.run_autonomous_cycle() # ~34ns per cycle +``` + +See [Performance Tuning Guide](driver/python/accl_quantum/docs/performance_tuning.md) for details. + +--- + ## Future Work - Integration with IBM Quantum systems via Qiskit diff --git a/api_server.py b/api_server.py index 6a9a631..0484c90 100644 --- a/api_server.py +++ b/api_server.py @@ -32,7 +32,9 @@ NoiseParameters, GateType, ) -from accl_quantum.feedback import MeasurementFeedbackPipeline, FeedbackConfig +from accl_quantum.feedback import MeasurementFeedbackPipeline, FeedbackConfig, HardwareFeedbackEngine +from accl_quantum.hardware_accel import HardwareAccelerator +from accl_quantum.constants import ULLPipelineConfig, ULL_TARGET_TOTAL_NS # Constants MAX_EMULATORS = 50 @@ -48,6 +50,7 @@ "standard": ACCLMode.STANDARD, "deterministic": ACCLMode.DETERMINISTIC, "low_latency": ACCLMode.LOW_LATENCY, + "ultra_low_latency": ACCLMode.ULTRA_LOW_LATENCY, } # Global instances @@ -59,7 +62,7 @@ # Request/Response models class CreateClusterRequest(BaseModel): num_ranks: int = Field(default=4, ge=2, le=16, description="Number of FPGA ranks to simulate") - mode: str = Field(default="deterministic", description="Operation mode: standard, deterministic, low_latency") + mode: str = Field(default="deterministic", description="Operation mode: standard, deterministic, low_latency, ultra_low_latency") class BroadcastRequest(BaseModel): @@ -116,8 +119,8 @@ async def lifespan(app: FastAPI): app = FastAPI( title="ACCL-Q API", - description="Quantum Collective Communication Emulator API", - version="0.2.0", + description="Quantum Collective Communication Emulator API with Ultra-Low-Latency support", + version="0.3.0", lifespan=lifespan ) @@ -142,7 +145,7 @@ async def root(): """API info.""" return { "service": "ACCL-Q Quantum Emulator", - "version": "0.2.0", + "version": "0.3.0", "endpoints": { "/health": "Health check", "/cluster": "Create/manage ACCL cluster", @@ -151,6 +154,8 @@ async def root(): "/collective/allreduce": "Allreduce operation", "/collective/barrier": "Barrier synchronization", "/qec/syndrome": "QEC syndrome aggregation demo", + "/ull/status": "ULL pipeline status", + "/ull/feedback": "Run ULL autonomous feedback cycle", "/emulator": "Create qubit emulator", "/emulator/{id}/gate": "Apply quantum gate", "/emulator/{id}/measure": "Measure qubits", @@ -508,6 +513,104 @@ async def delete_emulator(emulator_id: str): return {"success": True, "message": f"Emulator {emulator_id} deleted"} +# ULL Pipeline Endpoints +_ull_engine: Optional[HardwareFeedbackEngine] = None + + +class ULLConfigRequest(BaseModel): + syndrome_bits: int = Field(default=16, ge=1, le=512) + coherence_time_us: float = Field(default=50.0, gt=0) + fiber_length_m: float = Field(default=1.0, ge=0) + + +@app.get("/ull/status") +async def ull_status(): + """Get ULL pipeline status.""" + if _ull_engine is None: + return { + "initialized": False, + "message": "ULL engine not initialized. POST /ull/configure first." + } + + stats = _ull_engine.get_stats() + return { + "initialized": True, + "armed": _ull_engine._armed, + "stats": stats, + "latency_budget_ns": ULL_TARGET_TOTAL_NS, + } + + +@app.post("/ull/configure") +async def ull_configure(request: ULLConfigRequest): + """Configure and arm the ULL hardware feedback pipeline.""" + global _ull_engine + + config = ULLPipelineConfig( + max_syndrome_bits=request.syndrome_bits, + coherence_time_us=request.coherence_time_us, + fiber_length_m=request.fiber_length_m, + ) + + _ull_engine = HardwareFeedbackEngine(config) + + def simple_decoder(syndrome): + return syndrome + + entries = _ull_engine.program_pipeline( + decoder_fn=simple_decoder, + syndrome_bits=request.syndrome_bits, + ) + + return { + "success": True, + "lut_entries": entries, + "estimated_latency_ns": _ull_engine._hw_accel.estimate_latency_ns(), + "message": f"ULL pipeline armed with {entries} LUT entries" + } + + +@app.post("/ull/feedback") +async def ull_feedback(num_cycles: int = 1): + """Run ULL autonomous feedback cycle(s).""" + if _ull_engine is None: + raise HTTPException( + status_code=400, + detail="ULL engine not initialized. POST /ull/configure first." + ) + + if num_cycles == 1: + result = _ull_engine.run_autonomous_cycle() + return { + "success": result.success, + "total_latency_ns": result.total_latency_ns, + "within_budget": result.within_budget, + "phases": result.phases, + } + else: + results = _ull_engine.run_continuous(num_cycles=min(num_cycles, 1000)) + violations = sum(1 for r in results if not r.within_budget) + latencies = [r.total_latency_ns for r in results] + return { + "success": all(r.success for r in results), + "num_cycles": len(results), + "violations": violations, + "mean_latency_ns": float(np.mean(latencies)), + "max_latency_ns": float(np.max(latencies)), + "min_latency_ns": float(np.min(latencies)), + } + + +@app.post("/ull/disarm") +async def ull_disarm(): + """Disarm the ULL pipeline.""" + global _ull_engine + if _ull_engine is None: + raise HTTPException(status_code=400, detail="ULL engine not initialized") + _ull_engine.disarm() + return {"success": True, "message": "ULL pipeline disarmed"} + + if __name__ == "__main__": import uvicorn port = int(os.environ.get("PORT", 8080)) diff --git a/demo_accl_q.py b/demo_accl_q.py index bd56d8c..9e55d15 100644 --- a/demo_accl_q.py +++ b/demo_accl_q.py @@ -9,6 +9,7 @@ 3. Measurement feedback pipeline 4. Realistic qubit emulation with noise 5. Latency monitoring and profiling +6. Ultra-Low-Latency (ULL) hardware-autonomous feedback """ import numpy as np @@ -32,6 +33,16 @@ ) from accl_quantum.feedback import MeasurementFeedbackPipeline, FeedbackConfig from accl_quantum.integrations import UnifiedQuantumControl +from accl_quantum.feedback import HardwareFeedbackEngine, ULLFeedbackResult +from accl_quantum.hardware_accel import HardwareAccelerator, DMABufferPool, LUTDecoder +from accl_quantum.constants import ( + ULLPipelineConfig, + ULL_TARGET_TOTAL_NS, + ULL_TARGET_MULTICAST_NS, + ULL_TARGET_REDUCE_NS, + ULL_TARGET_DECODE_NS, + ULL_TARGET_TRIGGER_NS, +) def print_header(title: str): @@ -445,6 +456,115 @@ def demo_latency_monitoring(): print(monitor.summary()) +def demo_ull_feedback(): + """Demo 6: Ultra-Low-Latency hardware-autonomous feedback.""" + print_header("Demo 6: Ultra-Low-Latency (ULL) Hardware Feedback") + + print("\nULL mode shifts feedback execution from software to hardware.") + print("Target: <50ns total feedback latency (0.1% of 50us coherence time)") + print() + + # Show latency budget breakdown + print("[1] ULL Latency Budget:") + print(f" Multicast: {ULL_TARGET_MULTICAST_NS}ns (simplified Aurora, on-board links)") + print(f" Reduce: {ULL_TARGET_REDUCE_NS}ns (combinational XOR, 1-2 cycles)") + print(f" Decode: {ULL_TARGET_DECODE_NS}ns (BRAM LUT, 4 cycles)") + print(f" Trigger: {ULL_TARGET_TRIGGER_NS}ns (hardware register, 1 cycle)") + print(f" Budget: {ULL_TARGET_TOTAL_NS}ns (0.1% of 50us coherence)") + + # Configure ULL pipeline + config = ULLPipelineConfig( + max_syndrome_bits=16, + coherence_time_us=50.0, + fiber_length_m=1.0, + ) + + print(f"\n[2] Configuring ULL pipeline:") + print(f" Syndrome bits: {config.max_syndrome_bits}") + print(f" Coherence time: {config.coherence_time_us}us") + print(f" Fiber length: {config.fiber_length_m}m") + + # Create and program the hardware feedback engine + engine = HardwareFeedbackEngine(config) + + def surface_code_decoder(syndrome): + """Simple majority-vote decoder for demo.""" + return syndrome # Identity decoder + + entries = engine.program_pipeline( + decoder_fn=surface_code_decoder, + syndrome_bits=16, + ) + + print(f" LUT entries programmed: {entries}") + print(f" Estimated latency: {engine._hw_accel.estimate_latency_ns():.1f}ns") + + # Validate configuration + warnings = engine._hw_accel.validate_config() + if warnings: + for w in warnings: + print(f" WARNING: {w}") + else: + print(" Config validated: all clear") + + # Run single autonomous cycle + print(f"\n[3] Running single autonomous feedback cycle:") + result = engine.run_autonomous_cycle() + + print(f" Success: {result.success}") + print(f" Total latency: {result.total_latency_ns:.1f}ns") + print(f" Within budget: {result.within_budget}") + print(f" Phase breakdown:") + for phase, ns in result.phases.items(): + print(f" {phase}: {ns:.1f}ns") + + # Run continuous cycles + num_cycles = 1000 + print(f"\n[4] Running {num_cycles} continuous feedback cycles:") + results = engine.run_continuous(num_cycles=num_cycles) + + latencies = [r.total_latency_ns for r in results] + violations = sum(1 for r in results if not r.within_budget) + + print(f" Cycles completed: {len(results)}") + print(f" Mean latency: {np.mean(latencies):.1f}ns") + print(f" Min latency: {np.min(latencies):.1f}ns") + print(f" Max latency: {np.max(latencies):.1f}ns") + print(f" Budget violations: {violations}/{num_cycles} ({violations/num_cycles*100:.1f}%)") + + # DMA buffer pool demo + print(f"\n[5] DMA Buffer Pool:") + pool = engine._hw_accel.pool + print(f" Total buffers: {pool.total}") + print(f" Available: {pool.available}") + + buf = pool.acquire() + print(f" Acquired buffer: {buf.shape}, dtype={buf.dtype}") + pool.release(buf) + print(f" Released. Available: {pool.available}") + + # ULL driver mode demo + print(f"\n[6] ULL Driver Mode (zero-copy):") + accl = ACCLQuantum(num_ranks=4, local_rank=0) + accl.configure(mode=ACCLMode.ULTRA_LOW_LATENCY) + + data = np.array([1, 0, 1, 0, 1, 1, 0, 0], dtype=np.uint8) + result = accl.broadcast(data, root=0) + + print(f" Input data: {data}") + print(f" Output data: {result.data}") + print(f" Latency: {result.latency_ns:.1f}ns") + print(f" Zero-copy verified: {result.data is data}") + + # Improvement summary + print(f"\n[7] Performance Comparison:") + print(f" Standard feedback: ~500ns (software-mediated)") + print(f" ULL feedback: ~{np.mean(latencies):.0f}ns (hardware-autonomous)") + improvement = 500 / np.mean(latencies) + print(f" Improvement: {improvement:.1f}x faster") + print(f" Coherence budget: {np.mean(latencies) / 50000 * 100:.4f}% of 50us") + + def main(): """Run all demos.""" print("\n" + "=" * 60) @@ -460,6 +580,7 @@ def main(): demo_qubit_emulator() demo_feedback_pipeline() demo_latency_monitoring() + demo_ull_feedback() print_header("Demo Complete!") print("\nAll demos completed successfully.") @@ -469,6 +590,7 @@ def main(): print(" - Hardware-synchronized barriers with <10ns jitter") print(" - Measurement feedback within 500ns budget") print(" - Realistic qubit emulation with T1/T2 noise") + print(" - ULL hardware-autonomous feedback in <50ns") print("\nFor more information, see the ACCL-Q documentation.") except Exception as e: diff --git a/driver/python/accl_quantum/docs/performance_tuning.md b/driver/python/accl_quantum/docs/performance_tuning.md index b26ba55..7eb430b 100644 --- a/driver/python/accl_quantum/docs/performance_tuning.md +++ b/driver/python/accl_quantum/docs/performance_tuning.md @@ -436,6 +436,133 @@ barrier: mean=89.2ns, p99=98.4ns, jitter=1.8ns [PASS] --- +## Ultra-Low-Latency (ULL) Mode + +### Overview + +ULL mode shifts feedback execution from software-mediated (~500ns) to hardware-autonomous (<50ns). Python handles setup and monitoring while the FPGA runs the feedback loop independently. + +### When to Use ULL + +- Feedback loops that must complete within 0.1% of coherence time +- Surface code QEC with tight latency budgets +- Short inter-board links (1-2m fiber) +- Syndrome sizes up to 512 bits + +### ULL Latency Budget + +| Component | Target | Clock Cycles | How | +|-----------|--------|-------------|-----| +| Multicast | 10ns | 5 | Simplified Aurora, on-board links | +| XOR Reduce | 4ns | 2 | Combinational logic | +| LUT Decode | 8ns | 4 | BRAM lookup table | +| Trigger | 2ns | 1 | Hardware register write | +| **Total** | **<50ns** | **<25** | Within 0.1% of 50us coherence | + +### Quick Start + +```python +from accl_quantum import ACCLQuantum, ACCLMode +from accl_quantum.feedback import HardwareFeedbackEngine +from accl_quantum.constants import ULLPipelineConfig + +# Option 1: ULL driver mode (zero-copy collectives) +accl = ACCLQuantum(num_ranks=4, local_rank=0) +accl.configure(mode=ACCLMode.ULTRA_LOW_LATENCY) +result = accl.broadcast(data, root=0) +# result.data is data (zero-copy identity) + +# Option 2: Hardware feedback engine (autonomous cycles) +config = ULLPipelineConfig( + max_syndrome_bits=16, + coherence_time_us=50.0, + fiber_length_m=1.0, +) +engine = HardwareFeedbackEngine(config) +engine.program_pipeline(decoder_fn=my_decoder, syndrome_bits=16) + +result = engine.run_autonomous_cycle() +print(f"Latency: {result.total_latency_ns}ns, within budget: {result.within_budget}") +``` + +### ULL Configuration + +```python +from accl_quantum.constants import ULLPipelineConfig + +config = ULLPipelineConfig( + max_syndrome_bits=512, # Max syndrome width + decoder_type='lut', # 'lut' (BRAM) or 'combinational' + lut_depth=4096, # LUT entries + use_hardware_multicast=True, # Hardware multicast fan-out + use_combinational_reduce=True, # Single-cycle XOR + coherence_time_us=50.0, # Qubit coherence time + auto_trigger=True, # Hardware trigger assertion + bypass_monitoring=True, # Skip profiling in hot path + dma_buffer_count=16, # Pre-allocated DMA buffers + fiber_length_m=1.0, # Short links for ULL +) +``` + +### Hardware Accelerator Components + +**DMA Buffer Pool**: Pre-allocated, cache-line-aligned buffers for zero-copy transfers. + +```python +from accl_quantum.hardware_accel import DMABufferPool + +pool = DMABufferPool(num_buffers=16, buffer_size_bytes=64) +buf = pool.acquire() # Zero-allocation in hot path +pool.release(buf) # Return to pool +``` + +**LUT Decoder**: BRAM-based syndrome-to-correction lookup. + +```python +from accl_quantum.hardware_accel import LUTDecoder + +decoder = LUTDecoder(num_syndrome_bits=16, lut_depth=4096) +entries = decoder.program(my_decoder_function) +correction = decoder.lookup(syndrome_array) +bram_image = decoder.get_bram_image() # For FPGA programming +``` + +**FPGA Register Interface**: Simulated register map for ULL pipeline control. + +```python +from accl_quantum.hardware_accel import FPGARegisterInterface + +regs = FPGARegisterInterface() +regs.arm_ull_pipeline() +assert regs.is_pipeline_active() +regs.disarm_ull_pipeline() +``` + +### Validating ULL Configuration + +```python +from accl_quantum.hardware_accel import HardwareAccelerator + +accel = HardwareAccelerator(config) +warnings = accel.validate_config() +for w in warnings: + print(f"WARNING: {w}") + +estimated = accel.estimate_latency_ns() +print(f"Estimated latency: {estimated:.1f}ns") +``` + +### Common ULL Issues + +| Issue | Cause | Fix | +|-------|-------|-----| +| Latency > 50ns | Fiber too long | Reduce `fiber_length_m` to 1-2m | +| LUT miss | Syndrome not in table | Increase `lut_depth` or use weight-3 enumeration | +| Buffer exhaustion | Too many concurrent ops | Increase `dma_buffer_count` | +| Jitter > 2ns | Software in hot path | Set `bypass_monitoring=True` | + +--- + ## See Also - [API Reference](api_reference.md) - Complete API documentation diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..8f68fae --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,55 @@ +[build-system] +requires = ["setuptools>=68.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "accl-quantum" +version = "0.3.0" +description = "Quantum-Optimized Alveo Collective Communication Library" +readme = "README.md" +license = {text = "Apache-2.0"} +requires-python = ">=3.11" +authors = [ + {name = "The AI Cowboys Projects", email = "ai-cowboys@example.com"}, +] +keywords = ["quantum", "fpga", "collective-communication", "qec", "accl"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Science/Research", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Scientific/Engineering :: Physics", +] +dependencies = [ + "numpy>=1.24", +] + +[project.optional-dependencies] +server = [ + "fastapi>=0.100", + "uvicorn[standard]>=0.23", + "pydantic>=2.0", +] +dev = [ + "pytest>=7.0", + "pytest-asyncio>=0.21", + "httpx>=0.24", +] +all = ["accl-quantum[server,dev]"] + +[project.urls] +Homepage = "https://github.com/The-AI-Cowboys-Projects/ACCL_NEW" +Repository = "https://github.com/The-AI-Cowboys-Projects/ACCL_NEW" + +[tool.setuptools.packages.find] +where = ["driver/python"] + +[tool.pytest.ini_options] +testpaths = ["test/quantum"] +asyncio_mode = "auto" +addopts = "-v --tb=short" + +[tool.setuptools.package-data] +accl_quantum = ["docs/*.md"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..94e9b16 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,12 @@ +# Core dependency +numpy>=1.24 + +# API server +fastapi>=0.100 +uvicorn[standard]>=0.23 +pydantic>=2.0 + +# Development/testing +pytest>=7.0 +pytest-asyncio>=0.21 +httpx>=0.24 diff --git a/test/quantum/test_module_coverage.py b/test/quantum/test_module_coverage.py new file mode 100644 index 0000000..276f29b --- /dev/null +++ b/test/quantum/test_module_coverage.py @@ -0,0 +1,769 @@ +""" +Tests for ACCL-Q modules: profiler, stats, deployment, constants. + +Fills coverage gaps for modules that previously lacked dedicated tests. +""" + +import sys +import os +import json +import time +import tempfile +from pathlib import Path + +import numpy as np +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'driver', 'python')) + +from accl_quantum.constants import ( + ACCLMode, ACCLConfig, ReduceOp, SyncMode, CollectiveOp, + OperationStatus, QuantumMsgType, LatencyBudget, ULLPipelineConfig, + CLOCK_PERIOD_NS, MAX_RANKS, TARGET_P2P_LATENCY_NS, + TARGET_BROADCAST_LATENCY_NS, TARGET_REDUCE_LATENCY_NS, + ULL_TARGET_TOTAL_NS, ULL_TARGET_MULTICAST_NS, + ULL_TARGET_REDUCE_NS, ULL_TARGET_DECODE_NS, ULL_TARGET_TRIGGER_NS, + ULL_MAX_SYNDROME_BITS, ULL_LUT_DECODER_DEPTH, + FEEDBACK_LATENCY_BUDGET_NS, MAX_JITTER_NS, +) +from accl_quantum.stats import ( + LatencyStats, LatencyRecord, LatencyMonitor, LatencyProfiler, +) +from accl_quantum.profiler import ( + CriticalPathProfiler, BottleneckAnalyzer, OptimizationAdvisor, + PerformanceRegressor, LatencyVisualizer, ProfilingSession, + LatencyBreakdown, Bottleneck, Recommendation, + BottleneckType, OptimizationCategory, ProfileSample, +) +from accl_quantum.deployment import ( + BoardType, NetworkTopology, DeploymentState, + BoardConfig, LinkConfig, DeploymentConfig, + TopologyBuilder, DeploymentManager, + create_default_deployment, +) + + +# ============================================================================ +# Constants Tests +# ============================================================================ + +class TestConstants: + """Test constants module values and consistency.""" + + def test_clock_period(self): + assert CLOCK_PERIOD_NS == 2 + assert 1000 / CLOCK_PERIOD_NS == 500 # 500 MHz + + def test_max_ranks(self): + assert MAX_RANKS == 16 + + def test_latency_targets_ordered(self): + """P2P < Broadcast < Reduce makes physical sense.""" + assert TARGET_P2P_LATENCY_NS < TARGET_BROADCAST_LATENCY_NS + assert TARGET_BROADCAST_LATENCY_NS <= TARGET_REDUCE_LATENCY_NS + + def test_ull_targets_sum_within_budget(self): + component_sum = ( + ULL_TARGET_MULTICAST_NS + ULL_TARGET_REDUCE_NS + + ULL_TARGET_DECODE_NS + ULL_TARGET_TRIGGER_NS + ) + assert component_sum <= ULL_TARGET_TOTAL_NS + + def test_ull_constants_positive(self): + assert ULL_MAX_SYNDROME_BITS > 0 + assert ULL_LUT_DECODER_DEPTH > 0 + + def test_feedback_budget(self): + assert FEEDBACK_LATENCY_BUDGET_NS == 500 + + +class TestACCLMode: + """Test ACCLMode enum.""" + + def test_modes_distinct(self): + modes = [ACCLMode.STANDARD, ACCLMode.DETERMINISTIC, + ACCLMode.LOW_LATENCY, ACCLMode.ULTRA_LOW_LATENCY] + assert len(set(modes)) == 4 + + def test_ull_mode_value(self): + assert ACCLMode.ULTRA_LOW_LATENCY == 3 + + def test_mode_ordering(self): + assert ACCLMode.STANDARD < ACCLMode.ULTRA_LOW_LATENCY + + +class TestACCLConfig: + """Test ACCLConfig dataclass.""" + + def test_default_values(self): + config = ACCLConfig(num_ranks=4, local_rank=0) + assert config.mode == ACCLMode.DETERMINISTIC + assert config.sync_mode == SyncMode.HARDWARE + assert config.enable_latency_monitoring is True + + def test_validate_valid(self): + config = ACCLConfig(num_ranks=8, local_rank=3) + assert config.validate() is True + + def test_validate_invalid_ranks(self): + config = ACCLConfig(num_ranks=0, local_rank=0) + with pytest.raises(ValueError): + config.validate() + + def test_validate_invalid_local_rank(self): + config = ACCLConfig(num_ranks=4, local_rank=5) + with pytest.raises(ValueError): + config.validate() + + def test_validate_max_ranks(self): + config = ACCLConfig(num_ranks=MAX_RANKS + 1, local_rank=0) + with pytest.raises(ValueError): + config.validate() + + +class TestULLPipelineConfig: + """Test ULLPipelineConfig defaults.""" + + def test_defaults(self): + config = ULLPipelineConfig() + assert config.max_syndrome_bits == ULL_MAX_SYNDROME_BITS + assert config.decoder_type == 'lut' + assert config.coherence_time_us == 50.0 + assert config.auto_trigger is True + assert config.bypass_monitoring is True + + def test_custom_config(self): + config = ULLPipelineConfig( + max_syndrome_bits=64, + fiber_length_m=0.5, + coherence_time_us=100.0, + ) + assert config.max_syndrome_bits == 64 + assert config.fiber_length_m == 0.5 + + +class TestLatencyBudget: + """Test LatencyBudget factory methods.""" + + def test_for_qec_cycle_default(self): + budget = LatencyBudget.for_qec_cycle() + assert budget.total_budget_ns == 10000 # 100us * 1000 * 10% + assert budget.communication_budget_ns == 6000 + assert budget.computation_budget_ns == 3000 + assert budget.margin_ns == 1000 + + def test_for_qec_cycle_custom_pct(self): + budget = LatencyBudget.for_qec_cycle(coherence_time_us=50.0, coherence_budget_pct=1.0) + assert budget.total_budget_ns == 500 # 50 * 1000 * 1% + + def test_for_feedback(self): + budget = LatencyBudget.for_feedback() + assert budget.total_budget_ns == FEEDBACK_LATENCY_BUDGET_NS + assert budget.communication_budget_ns == 300 + assert budget.computation_budget_ns == 150 + assert budget.margin_ns == 50 + + def test_for_ull_feedback(self): + budget = LatencyBudget.for_ull_feedback(coherence_time_us=50.0) + assert budget.total_budget_ns == 50.0 # 50us * 1000 * 0.1% + + def test_for_ull_feedback_custom_coherence(self): + budget = LatencyBudget.for_ull_feedback(coherence_time_us=100.0) + assert budget.total_budget_ns == 100.0 + + +class TestEnumerations: + """Test enum values match HLS constants.""" + + def test_reduce_ops(self): + assert ReduceOp.XOR == 0 + assert ReduceOp.ADD == 1 + assert ReduceOp.MAX == 2 + assert ReduceOp.MIN == 3 + + def test_collective_ops(self): + assert CollectiveOp.BROADCAST == 0 + assert CollectiveOp.BARRIER == 6 + + def test_operation_status(self): + assert OperationStatus.SUCCESS == 0 + assert OperationStatus.UNKNOWN_ERROR == 255 + + def test_quantum_msg_types(self): + assert QuantumMsgType.MEASUREMENT_DATA == 0x10 + assert QuantumMsgType.CONDITIONAL_OP == 0x14 + + +# ============================================================================ +# Stats Tests +# ============================================================================ + +class TestLatencyStats: + """Test LatencyStats dataclass.""" + + def test_from_samples(self): + samples = [100.0, 200.0, 300.0, 400.0, 500.0] + stats = LatencyStats.from_samples(samples) + assert stats.count == 5 + assert stats.mean_ns == 300.0 + assert stats.min_ns == 100.0 + assert stats.max_ns == 500.0 + + def test_from_empty_samples(self): + stats = LatencyStats.from_samples([]) + assert stats.count == 0 + assert stats.mean_ns == 0.0 + + def test_meets_target_pass(self): + stats = LatencyStats.from_samples([100.0, 102.0, 98.0, 101.0, 99.0]) + assert stats.meets_target(target_ns=200, jitter_target_ns=10) + + def test_meets_target_fail_mean(self): + stats = LatencyStats.from_samples([500.0, 600.0, 700.0]) + assert not stats.meets_target(target_ns=200, jitter_target_ns=100) + + def test_str_representation(self): + stats = LatencyStats.from_samples([100.0]) + s = str(stats) + assert "LatencyStats" in s + assert "mean=" in s + + +class TestLatencyMonitor: + """Test LatencyMonitor functionality.""" + + def test_record_and_stats(self): + monitor = LatencyMonitor() + for i in range(10): + monitor.record(CollectiveOp.BROADCAST, 200 + i, num_ranks=4) + + stats = monitor.get_stats() + assert CollectiveOp.BROADCAST in stats + assert stats[CollectiveOp.BROADCAST].count == 10 + + def test_violations(self): + monitor = LatencyMonitor() + # Broadcast target is 300ns + monitor.record(CollectiveOp.BROADCAST, 500, num_ranks=4) + monitor.record(CollectiveOp.BROADCAST, 100, num_ranks=4) + + violations = monitor.get_violations() + assert violations[CollectiveOp.BROADCAST] == 1 + + def test_violation_rate(self): + monitor = LatencyMonitor() + monitor.record(CollectiveOp.BROADCAST, 500, num_ranks=4) + monitor.record(CollectiveOp.BROADCAST, 100, num_ranks=4) + + rate = monitor.get_violation_rate(CollectiveOp.BROADCAST) + assert rate == 0.5 + + def test_histogram(self): + monitor = LatencyMonitor() + for _ in range(100): + monitor.record(CollectiveOp.BROADCAST, 200 + np.random.normal(0, 5), num_ranks=4) + + counts, edges = monitor.get_histogram(CollectiveOp.BROADCAST) + assert len(counts) > 0 + assert len(edges) == len(counts) + 1 + + def test_empty_histogram(self): + monitor = LatencyMonitor() + counts, edges = monitor.get_histogram(CollectiveOp.BROADCAST) + assert len(counts) == 0 + + def test_clear(self): + monitor = LatencyMonitor() + monitor.record(CollectiveOp.BROADCAST, 200, num_ranks=4) + monitor.clear() + stats = monitor.get_stats() + assert len(stats) == 0 + + def test_export_history(self): + monitor = LatencyMonitor() + monitor.record(CollectiveOp.REDUCE, 350, num_ranks=8, root_rank=0) + history = monitor.export_history() + assert len(history) == 1 + assert history[0]['operation'] == 'REDUCE' + assert history[0]['latency_ns'] == 350 + + def test_alert_callback(self): + monitor = LatencyMonitor() + alerts = [] + monitor.add_alert_callback(lambda op, lat, target: alerts.append((op, lat))) + + monitor.record(CollectiveOp.BROADCAST, 500, num_ranks=4) + assert len(alerts) == 1 + + def test_summary(self): + monitor = LatencyMonitor() + monitor.record(CollectiveOp.BROADCAST, 200, num_ranks=4) + summary = monitor.summary() + assert "BROADCAST" in summary + + def test_get_stats_single_op(self): + monitor = LatencyMonitor() + monitor.record(CollectiveOp.BROADCAST, 200, num_ranks=4) + monitor.record(CollectiveOp.REDUCE, 350, num_ranks=4) + stats = monitor.get_stats(CollectiveOp.BROADCAST) + assert CollectiveOp.BROADCAST in stats + assert CollectiveOp.REDUCE not in stats + + +class TestLatencyProfiler: + """Test LatencyProfiler context manager.""" + + def test_profiler_context(self): + monitor = LatencyMonitor() + with LatencyProfiler(monitor, CollectiveOp.BROADCAST, num_ranks=4): + time.sleep(0.001) + + stats = monitor.get_stats() + assert CollectiveOp.BROADCAST in stats + assert stats[CollectiveOp.BROADCAST].count == 1 + assert stats[CollectiveOp.BROADCAST].mean_ns > 0 + + +# ============================================================================ +# Profiler Tests +# ============================================================================ + +class TestCriticalPathProfiler: + """Test CriticalPathProfiler.""" + + def test_start_end_operation(self): + profiler = CriticalPathProfiler() + op_id = profiler.start_operation('broadcast') + time.sleep(0.001) + duration = profiler.end_operation(op_id) + assert duration is not None + assert duration > 0 + + def test_end_unknown_operation(self): + profiler = CriticalPathProfiler() + assert profiler.end_operation('nonexistent') is None + + def test_record_phase(self): + profiler = CriticalPathProfiler() + profiler.record_phase('broadcast', 'tree_down', 180.0) + profiler.record_phase('broadcast', 'serialize', 50.0) + + breakdown = profiler.get_breakdown('broadcast') + assert 'tree_down' in breakdown.phases + assert breakdown.phases['tree_down'] == 180.0 + + def test_get_breakdown_empty(self): + profiler = CriticalPathProfiler() + breakdown = profiler.get_breakdown('nonexistent') + assert breakdown.total_ns == 0 + + def test_get_critical_path(self): + profiler = CriticalPathProfiler() + profiler.record_phase('broadcast', 'tree_down', 180.0) + profiler.record_phase('broadcast', 'serialize', 50.0) + + path = profiler.get_critical_path('broadcast') + assert len(path) == 2 + assert path[0][0] == 'tree_down' # Highest first + + def test_clear(self): + profiler = CriticalPathProfiler() + profiler.record_phase('broadcast', 'tree_down', 180.0) + profiler.clear() + breakdown = profiler.get_breakdown('broadcast') + assert breakdown.total_ns == 0 + + def test_ull_phases_defined(self): + profiler = CriticalPathProfiler() + assert 'ull_feedback' in profiler._operation_phases + assert 'readout' in profiler._operation_phases['ull_feedback'] + assert 'trigger' in profiler._operation_phases['ull_feedback'] + + def test_operation_phases_completeness(self): + profiler = CriticalPathProfiler() + expected_ops = ['broadcast', 'reduce', 'allreduce', 'barrier', + 'scatter', 'gather', 'feedback', + 'ull_feedback', 'ull_broadcast', 'ull_reduce'] + for op in expected_ops: + assert op in profiler._operation_phases + + +class TestLatencyBreakdown: + """Test LatencyBreakdown dataclass.""" + + def test_overhead(self): + bd = LatencyBreakdown(total_ns=300.0, phases={'a': 100.0, 'b': 150.0}) + assert bd.overhead_ns == 50.0 + + def test_percentage(self): + bd = LatencyBreakdown(total_ns=200.0, phases={'a': 100.0}) + assert bd.percentage('a') == 50.0 + assert bd.percentage('nonexistent') == 0.0 + + def test_percentage_zero_total(self): + bd = LatencyBreakdown(total_ns=0) + assert bd.percentage('anything') == 0.0 + + def test_to_dict(self): + bd = LatencyBreakdown(total_ns=100.0, phases={'x': 60.0}) + d = bd.to_dict() + assert d['total_ns'] == 100.0 + assert d['phases']['x'] == 60.0 + assert d['overhead_ns'] == 40.0 + + +class TestBottleneckAnalyzer: + """Test BottleneckAnalyzer.""" + + def test_no_data_no_bottlenecks(self): + profiler = CriticalPathProfiler() + analyzer = BottleneckAnalyzer(profiler) + assert len(analyzer.analyze()) == 0 + + def test_network_bottleneck_detection(self): + profiler = CriticalPathProfiler() + # Record total + op_id = profiler.start_operation('broadcast') + profiler._samples.append(ProfileSample( + timestamp_ns=0, operation='broadcast', phase='total', duration_ns=300.0)) + profiler.record_phase('broadcast', 'tree_down', 250.0) + profiler.record_phase('broadcast', 'serialize', 30.0) + + analyzer = BottleneckAnalyzer(profiler) + bottlenecks = analyzer.analyze() + network_bns = [b for b in bottlenecks if b.type == BottleneckType.NETWORK_LATENCY] + assert len(network_bns) > 0 + + def test_get_summary(self): + profiler = CriticalPathProfiler() + analyzer = BottleneckAnalyzer(profiler) + summary = analyzer.get_summary() + assert 'total_bottlenecks' in summary + + +class TestOptimizationAdvisor: + """Test OptimizationAdvisor.""" + + def test_no_recommendations_when_clean(self): + profiler = CriticalPathProfiler() + analyzer = BottleneckAnalyzer(profiler) + advisor = OptimizationAdvisor(analyzer) + recs = advisor.get_recommendations() + assert len(recs) == 0 + + def test_top_recommendations(self): + profiler = CriticalPathProfiler() + analyzer = BottleneckAnalyzer(profiler) + advisor = OptimizationAdvisor(analyzer) + top = advisor.get_top_recommendations(n=3) + assert len(top) <= 3 + + +class TestPerformanceRegressor: + """Test PerformanceRegressor.""" + + def test_no_regressions_no_baseline(self): + regressor = PerformanceRegressor() + regressor.update_current('broadcast', LatencyStats.from_samples([200, 210, 190])) + assert len(regressor.check_regressions()) == 0 + + def test_detect_regression(self): + regressor = PerformanceRegressor() + regressor._baseline['broadcast'] = LatencyStats.from_samples([200, 210, 190]) + regressor.update_current('broadcast', LatencyStats.from_samples([400, 410, 390])) + regressions = regressor.check_regressions() + assert len(regressions) > 0 + assert regressions[0]['operation'] == 'broadcast' + + def test_save_load_baseline(self): + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "baseline.json" + + # Save baseline first (file doesn't exist yet) + regressor = PerformanceRegressor(baseline_path=path) + regressor.update_current('broadcast', LatencyStats.from_samples([200, 210, 190])) + regressor.save_baseline() + + # Load in new instance (now file exists with valid JSON) + regressor2 = PerformanceRegressor(baseline_path=path) + assert 'broadcast' in regressor2._baseline + + def test_get_comparison(self): + regressor = PerformanceRegressor() + regressor._baseline['broadcast'] = LatencyStats.from_samples([200]) + regressor.update_current('broadcast', LatencyStats.from_samples([250])) + comparison = regressor.get_comparison() + assert 'broadcast' in comparison + assert 'changes' in comparison['broadcast'] + + def test_update_from_monitor(self): + monitor = LatencyMonitor() + monitor.record(CollectiveOp.BROADCAST, 200, num_ranks=4) + regressor = PerformanceRegressor() + regressor.update_from_monitor(monitor) + assert 'BROADCAST' in regressor._current + + +class TestLatencyVisualizer: + """Test LatencyVisualizer.""" + + def test_breakdown_bar(self): + bd = LatencyBreakdown(total_ns=300.0, phases={'net': 200.0, 'serial': 80.0}) + result = LatencyVisualizer.breakdown_bar(bd) + assert "300.0ns" in result + assert "net" in result + + def test_breakdown_bar_no_data(self): + bd = LatencyBreakdown(total_ns=0) + assert "No data" in LatencyVisualizer.breakdown_bar(bd) + + def test_histogram(self): + samples = [100 + np.random.normal(0, 10) for _ in range(100)] + result = LatencyVisualizer.histogram(samples) + assert "n=100" in result + + def test_histogram_no_data(self): + assert "No data" in LatencyVisualizer.histogram([]) + + def test_comparison_table(self): + comparison = { + 'broadcast': { + 'baseline': {'mean_ns': 200}, + 'current': {'mean_ns': 250}, + 'changes': {'mean_percent': 25.0}, + } + } + result = LatencyVisualizer.comparison_table(comparison) + assert "broadcast" in result + + +class TestProfilingSession: + """Test ProfilingSession.""" + + def test_profile_operation_context(self): + session = ProfilingSession() + with session.profile_operation('broadcast'): + time.sleep(0.001) + + breakdown = session.profiler.get_breakdown('broadcast') + assert breakdown.total_ns > 0 + + def test_analyze(self): + session = ProfilingSession() + result = session.analyze() + assert 'session_duration_ns' in result + assert 'bottlenecks' in result + assert 'recommendations' in result + + def test_generate_report(self): + session = ProfilingSession() + report = session.generate_report() + assert "ACCL-Q PERFORMANCE PROFILING REPORT" in report + + +# ============================================================================ +# Deployment Tests +# ============================================================================ + +class TestBoardConfig: + """Test BoardConfig.""" + + def test_to_dict_roundtrip(self): + board = BoardConfig( + rank=0, hostname="rfsoc-0", + ip_address="192.168.1.100", + mac_address="00:0a:35:00:00:00", + board_type=BoardType.ZCU216, + ) + d = board.to_dict() + assert d['rank'] == 0 + assert d['board_type'] == 'zcu216' + + restored = BoardConfig.from_dict(d) + assert restored.rank == 0 + assert restored.board_type == BoardType.ZCU216 + assert restored.hostname == "rfsoc-0" + + def test_default_values(self): + board = BoardConfig( + rank=1, hostname="test", + ip_address="10.0.0.1", + mac_address="aa:bb:cc:dd:ee:ff", + board_type=BoardType.RFSoC4x2, + ) + assert board.aurora_lanes == 4 + assert board.dac_channels == 8 + assert board.clock_source == "internal" + assert board.is_online is False + + +class TestDeploymentConfig: + """Test DeploymentConfig.""" + + def test_validate_valid(self): + config = create_default_deployment(num_boards=4) + errors = config.validate() + assert len(errors) == 0 + + def test_validate_too_few_boards(self): + config = DeploymentConfig(name="test", num_boards=1) + errors = config.validate() + assert any("Minimum 2" in e for e in errors) + + def test_validate_too_many_boards(self): + config = DeploymentConfig(name="test", num_boards=MAX_RANKS + 1) + errors = config.validate() + assert any("Maximum" in e for e in errors) + + def test_validate_master_rank_out_of_range(self): + config = DeploymentConfig(name="test", num_boards=4, master_rank=5) + errors = config.validate() + assert any("Master rank" in e for e in errors) + + def test_save_load_roundtrip(self): + config = create_default_deployment(num_boards=4, name="test-roundtrip") + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + path = Path(f.name) + + try: + config.save(path) + loaded = DeploymentConfig.load(path) + assert loaded.name == "test-roundtrip" + assert loaded.num_boards == 4 + assert len(loaded.boards) == 4 + assert len(loaded.links) > 0 + finally: + path.unlink(missing_ok=True) + + def test_min_links_topologies(self): + for topo, expected_min in [ + (NetworkTopology.STAR, 3), # n-1 = 3 for 4 boards + (NetworkTopology.RING, 4), # n = 4 + (NetworkTopology.TREE, 3), # n-1 = 3 + (NetworkTopology.FULL_MESH, 6), # 4*3/2 = 6 + ]: + config = DeploymentConfig(name="test", num_boards=4, topology=topo) + assert config._min_links_for_topology() == expected_min + + +class TestTopologyBuilder: + """Test TopologyBuilder.""" + + def _make_boards(self, n): + return [ + BoardConfig( + rank=i, hostname=f"board-{i}", + ip_address=f"10.0.0.{i}", + mac_address=f"00:00:00:00:00:{i:02x}", + board_type=BoardType.ZCU216, + ) + for i in range(n) + ] + + def test_build_star(self): + boards = self._make_boards(4) + links = TopologyBuilder.build_star(boards, center_rank=0) + # Each non-center board has bidirectional link = 2 * 3 = 6 + assert len(links) == 6 + + def test_build_ring(self): + boards = self._make_boards(4) + links = TopologyBuilder.build_ring(boards) + assert len(links) == 4 + + def test_build_tree(self): + boards = self._make_boards(4) + links = TopologyBuilder.build_tree(boards, root_rank=0, fanout=4) + # 3 child nodes * 2 (bidirectional) = 6 + assert len(links) == 6 + + def test_build_full_mesh(self): + boards = self._make_boards(4) + links = TopologyBuilder.build_full_mesh(boards) + # 4C2 * 2 = 12 bidirectional links + assert len(links) == 12 + + +class TestDeploymentManager: + """Test DeploymentManager basic functionality (no real network).""" + + def test_initial_state(self): + config = create_default_deployment(num_boards=4) + manager = DeploymentManager(config) + assert manager.state == DeploymentState.UNINITIALIZED + + def test_get_status(self): + config = create_default_deployment(num_boards=4) + manager = DeploymentManager(config) + status = manager.get_status() + assert status['state'] == 'uninitialized' + assert status['num_boards'] == 4 + assert status['online_boards'] == 0 + + def test_state_callbacks(self): + config = create_default_deployment(num_boards=4) + manager = DeploymentManager(config) + states = [] + manager.add_state_callback(lambda s: states.append(s)) + manager._set_state(DeploymentState.CONFIGURING) + assert states == [DeploymentState.CONFIGURING] + + def test_error_callbacks(self): + config = create_default_deployment(num_boards=4) + manager = DeploymentManager(config) + errors = [] + manager.add_error_callback(lambda msg: errors.append(msg)) + manager._report_error("test error") + assert "test error" in errors[0] + + def test_shutdown(self): + config = create_default_deployment(num_boards=4) + manager = DeploymentManager(config) + manager.shutdown() + assert manager.state == DeploymentState.SHUTDOWN + + +class TestCreateDefaultDeployment: + """Test create_default_deployment helper.""" + + def test_4_boards(self): + config = create_default_deployment(num_boards=4) + assert len(config.boards) == 4 + assert config.topology == NetworkTopology.TREE + assert len(config.links) > 0 + + def test_8_boards(self): + config = create_default_deployment(num_boards=8, name="big-test") + assert len(config.boards) == 8 + assert config.name == "big-test" + + def test_boards_have_correct_ips(self): + config = create_default_deployment(num_boards=4) + for rank in range(4): + assert config.boards[rank].ip_address == f"192.168.1.{100 + rank}" + + +class TestBoardType: + """Test BoardType enum.""" + + def test_all_types(self): + types = list(BoardType) + assert len(types) == 6 # ZCU111, ZCU216, RFSoC2x2, RFSoC4x2, HTGZRF16, CUSTOM + + +class TestNetworkTopology: + """Test NetworkTopology enum.""" + + def test_all_topologies(self): + topos = list(NetworkTopology) + assert len(topos) == 5 + + +class TestDeploymentState: + """Test DeploymentState enum.""" + + def test_state_values(self): + assert DeploymentState.UNINITIALIZED.value == "uninitialized" + assert DeploymentState.READY.value == "ready" + assert DeploymentState.ERROR.value == "error" diff --git a/test/quantum/test_ull_latency_validation.py b/test/quantum/test_ull_latency_validation.py new file mode 100644 index 0000000..0b5b806 --- /dev/null +++ b/test/quantum/test_ull_latency_validation.py @@ -0,0 +1,266 @@ +""" +ULL-specific latency validation tests. + +Validates that the ULL pipeline meets the 50ns budget target and that +component latencies are consistent with the HLS cycle counts. +""" + +import sys +import os + +import numpy as np +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'driver', 'python')) + +from accl_quantum.constants import ( + ULL_TARGET_MULTICAST_NS, ULL_TARGET_REDUCE_NS, + ULL_TARGET_DECODE_NS, ULL_TARGET_TRIGGER_NS, + ULL_TARGET_TOTAL_NS, ULL_MAX_JITTER_NS, + ULL_MAX_SYNDROME_BITS, ULL_LUT_DECODER_DEPTH, + CLOCK_PERIOD_NS, FIBER_DELAY_NS_PER_METER, + ULLPipelineConfig, LatencyBudget, ACCLMode, +) +from accl_quantum.hardware_accel import ( + HardwareAccelerator, DMABufferPool, LUTDecoder, FPGARegisterInterface, + ULLRegister, +) +from accl_quantum.feedback import HardwareFeedbackEngine, ULLFeedbackResult +from accl_quantum.driver import ACCLQuantum + + +class TestULLBudgetArithmetic: + """Validate that ULL budget arithmetic is internally consistent.""" + + def test_component_sum_within_budget(self): + """Sum of component targets must be <= total budget.""" + total = (ULL_TARGET_MULTICAST_NS + ULL_TARGET_REDUCE_NS + + ULL_TARGET_DECODE_NS + ULL_TARGET_TRIGGER_NS) + assert total <= ULL_TARGET_TOTAL_NS + + def test_cycle_count_consistency(self): + """Component targets must equal cycle_count * clock_period.""" + assert ULL_TARGET_MULTICAST_NS == 5 * CLOCK_PERIOD_NS # 5 cycles * 2ns + assert ULL_TARGET_REDUCE_NS == 2 * CLOCK_PERIOD_NS # 2 cycles * 2ns + assert ULL_TARGET_DECODE_NS == 4 * CLOCK_PERIOD_NS # 4 cycles * 2ns + assert ULL_TARGET_TRIGGER_NS == 1 * CLOCK_PERIOD_NS # 1 cycle * 2ns + + def test_total_budget_equals_25_cycles(self): + """50ns = 25 cycles at 500MHz.""" + assert ULL_TARGET_TOTAL_NS == 25 * CLOCK_PERIOD_NS + + def test_coherence_budget_derivation(self): + """Budget should be 0.1% of coherence time.""" + budget = LatencyBudget.for_ull_feedback(coherence_time_us=50.0) + assert budget.total_budget_ns == 50.0 + # 50ns = 0.1% of 50us (50000ns) + assert budget.total_budget_ns / 50000 == 0.001 + + +class TestULLHardwareAcceleratorLatency: + """Validate HardwareAccelerator latency estimates.""" + + def test_default_config_within_budget(self): + """Default ULL config should produce latency within 50ns.""" + accel = HardwareAccelerator() + estimated = accel.estimate_latency_ns() + assert estimated <= ULL_TARGET_TOTAL_NS + + def test_short_fiber_within_budget(self): + """1m fiber adds only 5ns, should be within budget.""" + config = ULLPipelineConfig(fiber_length_m=1.0) + accel = HardwareAccelerator(config) + estimated = accel.estimate_latency_ns() + fiber_component = 1.0 * FIBER_DELAY_NS_PER_METER + assert fiber_component == 5.0 + assert estimated <= ULL_TARGET_TOTAL_NS + + def test_long_fiber_exceeds_budget(self): + """10m fiber adds 50ns — should exceed budget.""" + config = ULLPipelineConfig(fiber_length_m=10.0) + accel = HardwareAccelerator(config) + estimated = accel.estimate_latency_ns() + assert estimated > ULL_TARGET_TOTAL_NS + + def test_hardware_multicast_vs_software(self): + """Hardware multicast should be faster than software fallback.""" + hw_config = ULLPipelineConfig(use_hardware_multicast=True) + sw_config = ULLPipelineConfig(use_hardware_multicast=False) + hw_accel = HardwareAccelerator(hw_config) + sw_accel = HardwareAccelerator(sw_config) + assert hw_accel.estimate_latency_ns() < sw_accel.estimate_latency_ns() + + def test_validation_warnings_long_fiber(self): + """Config with long fiber should produce warnings.""" + config = ULLPipelineConfig(fiber_length_m=5.0) + accel = HardwareAccelerator(config) + warnings = accel.validate_config() + fiber_warnings = [w for w in warnings if 'fiber' in w.lower() or 'Fiber' in w] + assert len(fiber_warnings) > 0 + + def test_validation_clean_default(self): + """Default config should produce no warnings.""" + accel = HardwareAccelerator() + warnings = accel.validate_config() + # Default has 1m fiber, should be clean + fiber_warnings = [w for w in warnings if 'fiber' in w.lower() or 'Fiber' in w] + assert len(fiber_warnings) == 0 + + +class TestULLFeedbackLatency: + """Validate HardwareFeedbackEngine produces results within budget.""" + + def _make_engine(self, **config_kwargs): + config = ULLPipelineConfig(**config_kwargs) + engine = HardwareFeedbackEngine(config) + engine.program_pipeline( + decoder_fn=lambda s: s, + syndrome_bits=16, + ) + return engine + + def test_single_cycle_within_budget(self): + engine = self._make_engine() + result = engine.run_autonomous_cycle() + assert result.success + assert result.within_budget + assert result.total_latency_ns <= ULL_TARGET_TOTAL_NS + + def test_phases_sum_to_total(self): + engine = self._make_engine() + result = engine.run_autonomous_cycle() + phase_sum = sum(result.phases.values()) + # Allow small float rounding + assert abs(phase_sum - result.total_latency_ns) < 1.0 + + def test_continuous_no_violations(self): + engine = self._make_engine() + results = engine.run_continuous(num_cycles=100) + violations = [r for r in results if not r.within_budget] + assert len(violations) == 0 + + def test_phase_names(self): + engine = self._make_engine() + result = engine.run_autonomous_cycle() + expected_phases = {'readout', 'multicast', 'reduce', 'decode', 'trigger'} + assert set(result.phases.keys()) == expected_phases + + def test_each_phase_positive(self): + engine = self._make_engine() + result = engine.run_autonomous_cycle() + for phase, ns in result.phases.items(): + assert ns > 0, f"Phase {phase} has non-positive latency: {ns}" + + def test_stats_tracking(self): + engine = self._make_engine() + for _ in range(10): + engine.run_autonomous_cycle() + stats = engine.get_stats() + assert stats['execution_count'] == 10 + assert stats['violations'] == 0 + + +class TestULLDriverLatency: + """Validate driver ULL mode returns correct latency values.""" + + def _make_ull_driver(self): + accl = ACCLQuantum(num_ranks=4, local_rank=0) + accl.configure(mode=ACCLMode.ULTRA_LOW_LATENCY) + return accl + + def test_broadcast_latency(self): + accl = self._make_ull_driver() + data = np.array([1, 0, 1, 0], dtype=np.uint8) + result = accl.broadcast(data, root=0) + assert result.latency_ns == ULL_TARGET_MULTICAST_NS + + def test_reduce_latency(self): + accl = self._make_ull_driver() + data = np.array([1, 0, 1, 0], dtype=np.uint8) + from accl_quantum.constants import ReduceOp + result = accl.reduce(data, op=ReduceOp.XOR, root=0) + assert result.latency_ns == ULL_TARGET_REDUCE_NS + + def test_allreduce_latency(self): + accl = self._make_ull_driver() + data = np.array([1, 0, 1, 0], dtype=np.uint8) + from accl_quantum.constants import ReduceOp + result = accl.allreduce(data, op=ReduceOp.XOR) + expected = ULL_TARGET_MULTICAST_NS + ULL_TARGET_REDUCE_NS + assert result.latency_ns == expected + + +class TestULLZeroCopyProof: + """Validate zero-copy semantics in ULL mode.""" + + def test_broadcast_zero_copy(self): + accl = ACCLQuantum(num_ranks=4, local_rank=0) + accl.configure(mode=ACCLMode.ULTRA_LOW_LATENCY) + data = np.array([42, 137, 255], dtype=np.uint8) + result = accl.broadcast(data, root=0) + assert result.data is data # identity check, not equality + + def test_reduce_zero_copy(self): + accl = ACCLQuantum(num_ranks=4, local_rank=0) + accl.configure(mode=ACCLMode.ULTRA_LOW_LATENCY) + data = np.array([1, 0, 1], dtype=np.uint8) + from accl_quantum.constants import ReduceOp + result = accl.reduce(data, op=ReduceOp.XOR, root=0) + assert result.data is data + + def test_allreduce_zero_copy(self): + accl = ACCLQuantum(num_ranks=4, local_rank=0) + accl.configure(mode=ACCLMode.ULTRA_LOW_LATENCY) + data = np.array([1, 0, 1], dtype=np.uint8) + from accl_quantum.constants import ReduceOp + result = accl.allreduce(data, op=ReduceOp.XOR) + assert result.data is data + + def test_standard_mode_copies(self): + """Standard mode should copy data, NOT zero-copy.""" + accl = ACCLQuantum(num_ranks=4, local_rank=0) + accl.configure(mode=ACCLMode.DETERMINISTIC) + data = np.array([42, 137, 255], dtype=np.uint8) + result = accl.broadcast(data, root=0) + # Standard mode copies + assert result.data is not data + assert np.array_equal(result.data, data) + + +class TestULLSyndromeLimits: + """Validate syndrome size enforcement in ULL mode.""" + + def test_max_syndrome_bits(self): + """LUT decoder should reject syndrome > max bits.""" + from accl_quantum.hardware_accel import LUTDecoder + with pytest.raises(ValueError, match="exceeds ULL max"): + LUTDecoder(num_syndrome_bits=ULL_MAX_SYNDROME_BITS + 1) + + def test_max_syndrome_accepted(self): + """Exactly max bits should be accepted.""" + from accl_quantum.hardware_accel import LUTDecoder + decoder = LUTDecoder(num_syndrome_bits=ULL_MAX_SYNDROME_BITS) + assert decoder._num_bits == ULL_MAX_SYNDROME_BITS + + +class TestULLModeIsolation: + """Validate that ULL mode doesn't affect other modes.""" + + def test_switch_to_ull_and_back(self): + accl = ACCLQuantum(num_ranks=4, local_rank=0) + + # Standard mode + accl.configure(mode=ACCLMode.DETERMINISTIC) + data = np.array([1, 2, 3], dtype=np.uint8) + r1 = accl.broadcast(data, root=0) + assert r1.data is not data # Standard mode copies + + # Switch to ULL + accl.configure(mode=ACCLMode.ULTRA_LOW_LATENCY) + r2 = accl.broadcast(data, root=0) + assert r2.data is data # ULL is zero-copy + + # Switch back + accl.configure(mode=ACCLMode.DETERMINISTIC) + r3 = accl.broadcast(data, root=0) + assert r3.data is not data # Back to standard