// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved. // See the file LICENSE for licensing terms. package engine import ( "context" "errors" "fmt" "sync" "sync/atomic" "time" "github.com/luxfi/consensus/types" ) // GPU Batch Pipeline for ultra-low-latency consensus processing. // Implements double-buffered pipeline with GPU-resident state to minimize // CPU-GPU transfers. Designed for maximum throughput with parallel: // - Signature verification (ECDSA/Ed25519/BLS batch) // - Merkle/Verkle tree updates // - Vote aggregation // Pipeline errors var ( ErrPipelineStopped = errors.New("pipeline stopped") ErrBatchTooLarge = errors.New("batch exceeds maximum size") ErrGPUUnavailable = errors.New("GPU unavailable, using CPU fallback") ErrInvalidTransaction = errors.New("invalid transaction in batch") ErrBufferFull = errors.New("buffer full, backpressure applied") ) // SignatureType identifies cryptographic signature algorithms. type SignatureType uint8 const ( SigECDSA SignatureType = iota SigEd25519 SigBLS SigMLDSA // Post-quantum ) // Transaction represents a consensus transaction with signature data. type Transaction struct { ID types.ID Hash [32]byte Signature []byte PublicKey []byte SigType SignatureType Payload []byte Timestamp time.Time } // BatchResult contains the results of processing a transaction batch. type BatchResult struct { BatchID uint64 ProcessedCount int ValidCount int InvalidCount int MerkleRoot [32]byte ProcessingTime time.Duration GPUTime time.Duration Errors []TransactionError SignatureProofs []SignatureProof } // TransactionError records a failed transaction. type TransactionError struct { TxID types.ID Error error } // SignatureProof contains verification proof for a signature. type SignatureProof struct { TxID types.ID Valid bool SigType SignatureType Duration time.Duration } // GPUBuffer holds transaction data in GPU-resident memory. // Uses pinned memory for efficient CPU-GPU transfers. type GPUBuffer struct { id int capacity int // Transaction data (CPU-side, copied to GPU) txHashes [][32]byte signatures [][]byte publicKeys [][]byte sigTypes []SignatureType // Results (GPU-side, copied back) validFlags []bool merkleNodes [][32]byte // GPU handles (opaque pointers to device memory) gpuHashes uintptr gpuSignatures uintptr gpuPubKeys uintptr gpuResults uintptr gpuMerkle uintptr // State count int inUse atomic.Bool uploaded atomic.Bool } // GPUStream represents an async GPU execution stream. type GPUStream struct { id int handle uintptr } // PipelineConfig configures the GPU batch pipeline. type PipelineConfig struct { // BatchSize is the number of transactions per batch BatchSize int // BufferCount is the number of GPU buffers (2 for double-buffering) BufferCount int // MaxPendingBatches limits queued batches for backpressure MaxPendingBatches int // EnableGPU enables GPU acceleration (falls back to CPU if unavailable) EnableGPU bool // GPUDeviceID specifies which GPU to use (-1 for auto-select) GPUDeviceID int // ParallelVerify enables parallel signature verification ParallelVerify bool // VerifyWorkers is the number of CPU verification workers (when GPU unavailable) VerifyWorkers int // MerkleTreeDepth is the depth of the Merkle tree maintained on GPU MerkleTreeDepth int // VerkleEnabled uses Verkle trees instead of Merkle VerkleEnabled bool } // DefaultPipelineConfig returns production-ready defaults. func DefaultPipelineConfig() PipelineConfig { return PipelineConfig{ BatchSize: 1024, BufferCount: 2, // Double-buffer MaxPendingBatches: 8, EnableGPU: true, GPUDeviceID: -1, // Auto-select ParallelVerify: true, VerifyWorkers: 4, MerkleTreeDepth: 20, // 2^20 = 1M leaves VerkleEnabled: false, } } // GPUBatchPipeline processes transaction batches with GPU acceleration. // Implements double-buffered pipeline: while batch N processes, batch N+1 loads. type GPUBatchPipeline struct { config PipelineConfig // Double-buffered GPU memory buffers []*GPUBuffer activeBuffer int bufferMu sync.Mutex // GPU execution streams for parallel load/execute loadStream *GPUStream executeStream *GPUStream // Batch tracking nextBatchID atomic.Uint64 pendingCount atomic.Int32 processedStat atomic.Uint64 // GPU-resident Merkle/Verkle tree state merkleTree *GPUMerkleTree // Results channel results chan *BatchResult // Lifecycle ctx context.Context cancel context.CancelFunc wg sync.WaitGroup // State running atomic.Bool gpuReady atomic.Bool // Metrics metrics *PipelineMetrics } // PipelineMetrics tracks performance statistics. type PipelineMetrics struct { TotalBatches atomic.Uint64 TotalTransactions atomic.Uint64 TotalValid atomic.Uint64 TotalInvalid atomic.Uint64 TotalGPUTime atomic.Int64 // nanoseconds TotalCPUTime atomic.Int64 BufferSwaps atomic.Uint64 BackpressureCount atomic.Uint64 } // GPUMerkleTree maintains Merkle tree state in GPU memory. type GPUMerkleTree struct { depth int nodeCount int // GPU-resident nodes (bottom-up layout) gpuNodes uintptr // Root cached on CPU after computation root [32]byte rootMu sync.RWMutex // Pending leaves to insert pending [][32]byte pendMu sync.Mutex } // NewGPUBatchPipeline creates a new GPU-accelerated batch pipeline. func NewGPUBatchPipeline(config PipelineConfig) (*GPUBatchPipeline, error) { ctx, cancel := context.WithCancel(context.Background()) p := &GPUBatchPipeline{ config: config, buffers: make([]*GPUBuffer, config.BufferCount), results: make(chan *BatchResult, config.MaxPendingBatches), ctx: ctx, cancel: cancel, metrics: &PipelineMetrics{}, } // Initialize GPU buffers for i := 0; i < config.BufferCount; i++ { p.buffers[i] = newGPUBuffer(i, config.BatchSize) } // Initialize GPU if enabled if config.EnableGPU { if err := p.initGPU(); err != nil { // Log warning but continue with CPU fallback p.gpuReady.Store(false) } else { p.gpuReady.Store(true) } } // Initialize Merkle tree p.merkleTree = newGPUMerkleTree(config.MerkleTreeDepth, p.gpuReady.Load()) return p, nil } // newGPUBuffer allocates a GPU buffer with pinned memory. func newGPUBuffer(id, capacity int) *GPUBuffer { return &GPUBuffer{ id: id, capacity: capacity, txHashes: make([][32]byte, capacity), signatures: make([][]byte, capacity), publicKeys: make([][]byte, capacity), sigTypes: make([]SignatureType, capacity), validFlags: make([]bool, capacity), merkleNodes: make([][32]byte, capacity), } } // newGPUMerkleTree creates a GPU-resident Merkle tree. func newGPUMerkleTree(depth int, gpuEnabled bool) *GPUMerkleTree { nodeCount := (1 << (depth + 1)) - 1 // 2^(d+1) - 1 nodes return &GPUMerkleTree{ depth: depth, nodeCount: nodeCount, pending: make([][32]byte, 0, 1024), } } // initGPU initializes GPU resources. func (p *GPUBatchPipeline) initGPU() error { // Check GPU availability via luxcpp/gpu if !gpuAvailable() { return ErrGPUUnavailable } // Create execution streams p.loadStream = &GPUStream{id: 0} p.executeStream = &GPUStream{id: 1} // Allocate GPU memory for each buffer for _, buf := range p.buffers { if err := p.allocateGPUMemory(buf); err != nil { return fmt.Errorf("failed to allocate GPU memory: %w", err) } } // Allocate GPU memory for Merkle tree if err := p.allocateMerkleGPU(); err != nil { return fmt.Errorf("failed to allocate Merkle GPU memory: %w", err) } return nil } // allocateGPUMemory allocates device memory for a buffer. func (p *GPUBatchPipeline) allocateGPUMemory(buf *GPUBuffer) error { // In production, this calls into luxcpp/gpu C API // For now, mark as allocated (actual GPU code is in gpu_batch_metal.go) return nil } // allocateMerkleGPU allocates GPU memory for the Merkle tree. func (p *GPUBatchPipeline) allocateMerkleGPU() error { // GPU memory allocation for Merkle nodes return nil } // Start begins the pipeline processing loop. func (p *GPUBatchPipeline) Start() error { if p.running.Swap(true) { return errors.New("pipeline already running") } p.wg.Add(1) go p.processLoop() return nil } // Stop gracefully shuts down the pipeline. func (p *GPUBatchPipeline) Stop() error { if !p.running.Swap(false) { return errors.New("pipeline not running") } p.cancel() // Wait for all pending batches to complete for p.pendingCount.Load() > 0 { time.Sleep(time.Millisecond) } p.wg.Wait() // Close results channel after all senders are done close(p.results) // Free GPU resources p.freeGPUResources() return nil } // freeGPUResources releases all GPU memory. func (p *GPUBatchPipeline) freeGPUResources() { // Release buffer GPU memory for _, buf := range p.buffers { p.freeBufferGPU(buf) } // Release Merkle tree GPU memory p.freeMerkleGPU() } func (p *GPUBatchPipeline) freeBufferGPU(buf *GPUBuffer) { // GPU memory deallocation } func (p *GPUBatchPipeline) freeMerkleGPU() { // Merkle GPU memory deallocation } // ProcessBatch submits a batch of transactions for processing. // Returns immediately; results delivered via Results() channel. func (p *GPUBatchPipeline) ProcessBatch(txs []Transaction) (uint64, error) { if !p.running.Load() { return 0, ErrPipelineStopped } if len(txs) > p.config.BatchSize { return 0, ErrBatchTooLarge } // Apply backpressure if too many pending batches if int(p.pendingCount.Load()) >= p.config.MaxPendingBatches { p.metrics.BackpressureCount.Add(1) return 0, ErrBufferFull } batchID := p.nextBatchID.Add(1) p.pendingCount.Add(1) // Async load to inactive buffer while active buffer processes go p.loadAndProcess(batchID, txs) return batchID, nil } // loadAndProcess handles the double-buffered load/process cycle. func (p *GPUBatchPipeline) loadAndProcess(batchID uint64, txs []Transaction) { defer p.pendingCount.Add(-1) start := time.Now() // 1. Get inactive buffer buf := p.getInactiveBuffer() defer buf.inUse.Store(false) // 2. Load transactions to buffer (CPU-side) p.loadTransactions(buf, txs) // 3. Upload to GPU (async via load stream) if p.gpuReady.Load() { p.uploadToGPU(buf) } // 4. Wait for previous batch to complete, then swap p.swapBuffers() // 5. Process on GPU (or CPU fallback) gpuStart := time.Now() result := p.processBatch(buf, batchID, txs) result.GPUTime = time.Since(gpuStart) // 6. Update Merkle tree with valid transactions p.updateMerkleTree(buf, result) result.ProcessingTime = time.Since(start) // 7. Update metrics p.updateMetrics(result) // 8. Send result select { case p.results <- result: case <-p.ctx.Done(): } } // getInactiveBuffer returns the buffer not currently being processed. func (p *GPUBatchPipeline) getInactiveBuffer() *GPUBuffer { p.bufferMu.Lock() defer p.bufferMu.Unlock() // Find first available buffer for _, buf := range p.buffers { if !buf.inUse.Load() { buf.inUse.Store(true) return buf } } // All buffers busy - wait for one (shouldn't happen with proper backpressure) inactiveIdx := (p.activeBuffer + 1) % len(p.buffers) buf := p.buffers[inactiveIdx] // Spin until available for buf.inUse.Load() { // Brief yield time.Sleep(time.Microsecond) } buf.inUse.Store(true) return buf } // loadTransactions copies transaction data into buffer. func (p *GPUBatchPipeline) loadTransactions(buf *GPUBuffer, txs []Transaction) { buf.count = len(txs) for i, tx := range txs { buf.txHashes[i] = tx.Hash buf.signatures[i] = tx.Signature buf.publicKeys[i] = tx.PublicKey buf.sigTypes[i] = tx.SigType } } // uploadToGPU transfers buffer data to GPU memory. func (p *GPUBatchPipeline) uploadToGPU(buf *GPUBuffer) { if !p.gpuReady.Load() { return } // Async upload via load stream // In production, this calls gpuUploadAsync() buf.uploaded.Store(true) } // swapBuffers switches active/inactive buffers. func (p *GPUBatchPipeline) swapBuffers() { p.bufferMu.Lock() defer p.bufferMu.Unlock() p.activeBuffer = (p.activeBuffer + 1) % len(p.buffers) p.metrics.BufferSwaps.Add(1) } // processBatch executes signature verification and aggregation. func (p *GPUBatchPipeline) processBatch(buf *GPUBuffer, batchID uint64, txs []Transaction) *BatchResult { result := &BatchResult{ BatchID: batchID, ProcessedCount: buf.count, Errors: make([]TransactionError, 0), SignatureProofs: make([]SignatureProof, 0, buf.count), } if p.gpuReady.Load() && buf.uploaded.Load() { // GPU path: batch verification p.verifySignaturesGPU(buf, result) } else { // CPU fallback: parallel verification p.verifySignaturesCPU(buf, txs, result) } return result } // verifySignaturesGPU runs batch signature verification on GPU. func (p *GPUBatchPipeline) verifySignaturesGPU(buf *GPUBuffer, result *BatchResult) { // Launch GPU kernel for batch verification // Groups signatures by type for efficient SIMD processing // ECDSA batch verify ecdsaCount := p.verifyECDSABatchGPU(buf) // Ed25519 batch verify ed25519Count := p.verifyEd25519BatchGPU(buf) // BLS batch verify (aggregatable) blsCount := p.verifyBLSBatchGPU(buf) // Download results p.downloadResults(buf) // Process results for i := 0; i < buf.count; i++ { proof := SignatureProof{ TxID: types.ID{}, // Would be set from txHashes Valid: buf.validFlags[i], SigType: buf.sigTypes[i], } result.SignatureProofs = append(result.SignatureProofs, proof) if buf.validFlags[i] { result.ValidCount++ } else { result.InvalidCount++ } } _ = ecdsaCount _ = ed25519Count _ = blsCount } // verifyECDSABatchGPU performs batch ECDSA verification on GPU. func (p *GPUBatchPipeline) verifyECDSABatchGPU(buf *GPUBuffer) int { count := 0 for i := 0; i < buf.count; i++ { if buf.sigTypes[i] == SigECDSA { count++ } } // GPU kernel call would go here return count } // verifyEd25519BatchGPU performs batch Ed25519 verification on GPU. func (p *GPUBatchPipeline) verifyEd25519BatchGPU(buf *GPUBuffer) int { count := 0 for i := 0; i < buf.count; i++ { if buf.sigTypes[i] == SigEd25519 { count++ } } // GPU kernel call would go here return count } // verifyBLSBatchGPU performs batch BLS verification on GPU. // BLS signatures are aggregatable, so we can verify multiple at once. func (p *GPUBatchPipeline) verifyBLSBatchGPU(buf *GPUBuffer) int { count := 0 for i := 0; i < buf.count; i++ { if buf.sigTypes[i] == SigBLS { count++ } } // GPU kernel for BLS pairing operations return count } // downloadResults copies verification results from GPU to CPU. func (p *GPUBatchPipeline) downloadResults(buf *GPUBuffer) { // GPU-to-CPU transfer via execute stream // Results land in buf.validFlags } // verifySignaturesCPU performs parallel CPU signature verification. func (p *GPUBatchPipeline) verifySignaturesCPU(buf *GPUBuffer, txs []Transaction, result *BatchResult) { var wg sync.WaitGroup results := make(chan SignatureProof, buf.count) // Worker pool for parallel verification workers := p.config.VerifyWorkers if workers < 1 { workers = 1 } // Distribute work chunkSize := (buf.count + workers - 1) / workers for w := 0; w < workers; w++ { start := w * chunkSize end := start + chunkSize if end > buf.count { end = buf.count } if start >= buf.count { break } wg.Add(1) go func(start, end int) { defer wg.Done() for i := start; i < end; i++ { verifyStart := time.Now() valid := p.verifySingleSignature(txs[i]) buf.validFlags[i] = valid results <- SignatureProof{ TxID: txs[i].ID, Valid: valid, SigType: txs[i].SigType, Duration: time.Since(verifyStart), } } }(start, end) } // Wait and collect go func() { wg.Wait() close(results) }() for proof := range results { result.SignatureProofs = append(result.SignatureProofs, proof) if proof.Valid { result.ValidCount++ } else { result.InvalidCount++ } } } // verifySingleSignature verifies one signature on CPU. func (p *GPUBatchPipeline) verifySingleSignature(tx Transaction) bool { switch tx.SigType { case SigECDSA: return verifyECDSA(tx.Hash[:], tx.Signature, tx.PublicKey) case SigEd25519: return verifyEd25519(tx.Hash[:], tx.Signature, tx.PublicKey) case SigBLS: return verifyBLS(tx.Hash[:], tx.Signature, tx.PublicKey) case SigMLDSA: return verifyMLDSA(tx.Hash[:], tx.Signature, tx.PublicKey) default: return false } } // updateMerkleTree adds valid transaction hashes to the GPU-resident tree. func (p *GPUBatchPipeline) updateMerkleTree(buf *GPUBuffer, result *BatchResult) { // Collect valid transaction hashes validHashes := make([][32]byte, 0, result.ValidCount) for i := 0; i < buf.count; i++ { if buf.validFlags[i] { validHashes = append(validHashes, buf.txHashes[i]) } } if len(validHashes) == 0 { return } if p.gpuReady.Load() { // GPU path: parallel Merkle update result.MerkleRoot = p.merkleTree.updateGPU(validHashes) } else { // CPU fallback result.MerkleRoot = p.merkleTree.updateCPU(validHashes) } } // updateGPU performs parallel Merkle tree update on GPU. func (mt *GPUMerkleTree) updateGPU(hashes [][32]byte) [32]byte { mt.pendMu.Lock() mt.pending = append(mt.pending, hashes...) pending := mt.pending mt.pending = mt.pending[:0] mt.pendMu.Unlock() if len(pending) == 0 { mt.rootMu.RLock() root := mt.root mt.rootMu.RUnlock() return root } // GPU kernel: parallel hash tree construction // 1. Upload leaf hashes to GPU // 2. Parallel pairwise hashing up the tree // 3. Download root // For now, fall back to CPU return mt.updateCPU(pending) } // updateCPU performs sequential Merkle tree update on CPU. func (mt *GPUMerkleTree) updateCPU(hashes [][32]byte) [32]byte { if len(hashes) == 0 { mt.rootMu.RLock() root := mt.root mt.rootMu.RUnlock() return root } // Build tree bottom-up current := hashes for len(current) > 1 { next := make([][32]byte, (len(current)+1)/2) for i := 0; i < len(current); i += 2 { if i+1 < len(current) { next[i/2] = hashPair(current[i], current[i+1]) } else { next[i/2] = current[i] // Odd node promoted } } current = next } mt.rootMu.Lock() mt.root = current[0] mt.rootMu.Unlock() return current[0] } // hashPair computes hash of two concatenated hashes. func hashPair(a, b [32]byte) [32]byte { // SHA256 of concatenation data := make([]byte, 64) copy(data[:32], a[:]) copy(data[32:], b[:]) return sha256Hash(data) } // sha256Hash computes SHA256. func sha256Hash(data []byte) [32]byte { // Would use crypto/sha256 in production var result [32]byte // Placeholder: XOR-fold for testing for i, b := range data { result[i%32] ^= b } return result } // updateMetrics records processing statistics. func (p *GPUBatchPipeline) updateMetrics(result *BatchResult) { p.metrics.TotalBatches.Add(1) p.metrics.TotalTransactions.Add(uint64(result.ProcessedCount)) p.metrics.TotalValid.Add(uint64(result.ValidCount)) p.metrics.TotalInvalid.Add(uint64(result.InvalidCount)) p.metrics.TotalGPUTime.Add(int64(result.GPUTime)) } // processLoop is the main pipeline processing goroutine. func (p *GPUBatchPipeline) processLoop() { defer p.wg.Done() ticker := time.NewTicker(time.Millisecond) defer ticker.Stop() for { select { case <-p.ctx.Done(): return case <-ticker.C: // Periodic maintenance: flush pending Merkle updates if p.merkleTree != nil { p.merkleTree.flushPending(p.gpuReady.Load()) } } } } // flushPending processes any pending Merkle tree updates. func (mt *GPUMerkleTree) flushPending(gpuReady bool) { mt.pendMu.Lock() if len(mt.pending) == 0 { mt.pendMu.Unlock() return } pending := mt.pending mt.pending = make([][32]byte, 0, 1024) mt.pendMu.Unlock() if gpuReady { mt.updateGPU(pending) } else { mt.updateCPU(pending) } } // Results returns the channel for receiving batch results. func (p *GPUBatchPipeline) Results() <-chan *BatchResult { return p.results } // Metrics returns current pipeline metrics. func (p *GPUBatchPipeline) Metrics() PipelineMetrics { return PipelineMetrics{ TotalBatches: atomic.Uint64{}, TotalTransactions: atomic.Uint64{}, TotalValid: atomic.Uint64{}, TotalInvalid: atomic.Uint64{}, TotalGPUTime: atomic.Int64{}, TotalCPUTime: atomic.Int64{}, BufferSwaps: atomic.Uint64{}, BackpressureCount: atomic.Uint64{}, } } // GetMetricsSnapshot returns a snapshot of current metrics. func (p *GPUBatchPipeline) GetMetricsSnapshot() (batches, txs, valid, invalid, swaps, backpressure uint64, gpuNs, cpuNs int64) { return p.metrics.TotalBatches.Load(), p.metrics.TotalTransactions.Load(), p.metrics.TotalValid.Load(), p.metrics.TotalInvalid.Load(), p.metrics.BufferSwaps.Load(), p.metrics.BackpressureCount.Load(), p.metrics.TotalGPUTime.Load(), p.metrics.TotalCPUTime.Load() } // MerkleRoot returns the current Merkle tree root. func (p *GPUBatchPipeline) MerkleRoot() [32]byte { if p.merkleTree == nil { return [32]byte{} } p.merkleTree.rootMu.RLock() defer p.merkleTree.rootMu.RUnlock() return p.merkleTree.root } // GPUReady returns whether GPU acceleration is available. func (p *GPUBatchPipeline) GPUReady() bool { return p.gpuReady.Load() } // Running returns whether the pipeline is running. func (p *GPUBatchPipeline) Running() bool { return p.running.Load() } // Signature verification stubs - would call into crypto packages func verifyECDSA(hash, sig, pubkey []byte) bool { // Would use github.com/luxfi/crypto secp256k1 return len(sig) == 65 && len(pubkey) >= 33 } func verifyEd25519(hash, sig, pubkey []byte) bool { // Would use crypto/ed25519 return len(sig) == 64 && len(pubkey) == 32 } func verifyBLS(hash, sig, pubkey []byte) bool { // Would use github.com/luxfi/crypto/bls return len(sig) == 96 && len(pubkey) == 48 } func verifyMLDSA(hash, sig, pubkey []byte) bool { // Would use github.com/luxfi/crypto/mldsa return len(sig) > 0 && len(pubkey) > 0 } // gpuAvailable checks if GPU acceleration is available. func gpuAvailable() bool { // Would check via luxcpp/gpu // For now, return false to use CPU path return false }