|
|
"""
|
|
|
Register file implementation for virtual GPU memory subsystem
|
|
|
"""
|
|
|
from typing import Dict, Any, List, Optional
|
|
|
import numpy as np
|
|
|
import threading
|
|
|
|
|
|
class RegisterFile:
|
|
|
def __init__(self, num_registers: int = 32, register_width: int = 32):
|
|
|
self.num_registers = num_registers
|
|
|
self.register_width = register_width
|
|
|
self.registers = np.zeros(num_registers, dtype=np.int32)
|
|
|
self.register_lock = threading.Lock()
|
|
|
self.usage_stats = {
|
|
|
"reads": 0,
|
|
|
"writes": 0,
|
|
|
"conflicts": 0
|
|
|
}
|
|
|
|
|
|
def read_register(self, reg_id: int) -> int:
|
|
|
"""Read value from specified register"""
|
|
|
if not 0 <= reg_id < self.num_registers:
|
|
|
raise ValueError(f"Invalid register ID: {reg_id}")
|
|
|
|
|
|
with self.register_lock:
|
|
|
self.usage_stats["reads"] += 1
|
|
|
return int(self.registers[reg_id])
|
|
|
|
|
|
def write_register(self, reg_id: int, value: int) -> None:
|
|
|
"""Write value to specified register"""
|
|
|
if not 0 <= reg_id < self.num_registers:
|
|
|
raise ValueError(f"Invalid register ID: {reg_id}")
|
|
|
|
|
|
|
|
|
max_value = (1 << self.register_width) - 1
|
|
|
value = value & max_value
|
|
|
|
|
|
with self.register_lock:
|
|
|
self.usage_stats["writes"] += 1
|
|
|
self.registers[reg_id] = value
|
|
|
|
|
|
def bulk_read(self, reg_ids: List[int]) -> List[int]:
|
|
|
"""Read multiple registers at once"""
|
|
|
if not all(0 <= rid < self.num_registers for rid in reg_ids):
|
|
|
raise ValueError("Invalid register ID in bulk read")
|
|
|
|
|
|
with self.register_lock:
|
|
|
self.usage_stats["reads"] += len(reg_ids)
|
|
|
return [int(self.registers[rid]) for rid in reg_ids]
|
|
|
|
|
|
def bulk_write(self, reg_data: Dict[int, int]) -> None:
|
|
|
"""Write to multiple registers at once"""
|
|
|
if not all(0 <= rid < self.num_registers for rid in reg_data.keys()):
|
|
|
raise ValueError("Invalid register ID in bulk write")
|
|
|
|
|
|
max_value = (1 << self.register_width) - 1
|
|
|
values = {rid: val & max_value for rid, val in reg_data.items()}
|
|
|
|
|
|
with self.register_lock:
|
|
|
self.usage_stats["writes"] += len(values)
|
|
|
for rid, val in values.items():
|
|
|
self.registers[rid] = val
|
|
|
|
|
|
def clear_registers(self) -> None:
|
|
|
"""Reset all registers to zero"""
|
|
|
with self.register_lock:
|
|
|
self.registers.fill(0)
|
|
|
|
|
|
def get_usage_stats(self) -> Dict[str, int]:
|
|
|
"""Get register file usage statistics"""
|
|
|
with self.register_lock:
|
|
|
return dict(self.usage_stats)
|
|
|
|
|
|
def check_conflicts(self, read_regs: List[int], write_regs: List[int]) -> bool:
|
|
|
"""Check for read/write conflicts between register sets"""
|
|
|
read_set = set(read_regs)
|
|
|
write_set = set(write_regs)
|
|
|
|
|
|
|
|
|
conflicts = read_set.intersection(write_set)
|
|
|
if conflicts:
|
|
|
self.usage_stats["conflicts"] += 1
|
|
|
return True
|
|
|
|
|
|
return False
|
|
|
|
