Update qemu/qemu_manager.py

qemu/qemu_manager.py

@@ -1,702 +1,457 @@
-"""
-QEMU Virtual Machine Manager with Direct DuckDB Storage and Virtual Hardware Integration
-Manages VM state, virtual GPU, VRAM, and CPU states through remote DuckDB backend
-"""
-
-import os
-import asyncio
-import subprocess
-import time
-from pathlib import Path
-from typing import Dict, Optional, List
-import json
-import logging
-import duckdb
-from datetime import datetime
-import qemu.qmp as qmp
-from qemu import qmp_shell
-from libvirt import libvirt
-
-# Local imports
-from ..virtual_gpu_driver.src.driver_api import GPUError, VirtualGPUDriver
-from ..virtual_gpu_driver.src.memory.duckdb_memory_manager import DuckDBMemoryManager
-from ..virtual_gpu_driver.src.hal.hal import HardwareAbstractionLayer
-from ..vram.remote_storage import RemoteStorageManager
-from ..config import get_db_url, get_hf_token_cached
-from ..cpu.enhanced_cpu import EnhancedCPU, CPUGroupType, VirtualCPU, CPUInstruction, InstructionType
-
-# Import QEMU Python bindings
-try:
-    import pyqemu
-except ImportError:
-    logger.warning("pyqemu not found, some features may be limited")
-
-# Configure logging
-logging.basicConfig(level=logging.DEBUG)
-logger = logging.getLogger(__name__)
-
-class QEMUManager:
-    
-    def __init__(self, config_path: str = "config.json"):
-        self.config = self.load_config(config_path)
-        self.vm_process: Optional[subprocess.Popen] = None
-        self.monitor_socket = None
-        self.vnc_port = 5900
-        self.memory_service = None
-        
-        # Initialize storage and hardware components
-        self.db_url = "hf://datasets/Fred808/helium/storage.json"
-        self.con = self._init_db_connection()
-        self.memory_manager = DuckDBMemoryManager(self.db_url)
-        self.storage_manager = RemoteStorageManager(self.db_url)
-        
-        # Memory service socket path
-        self.memory_socket = "/tmp/memory-backend.sock"
-        
-    def _init_db_connection(self) -> duckdb.DuckDBPyConnection:
-        """Initialize database connection with HuggingFace configuration"""
-        con = duckdb.connect(self.db_url)
-        
-        # Configure HuggingFace access
-        con.execute("INSTALL httpfs;")
-        con.execute("LOAD httpfs;")
-        con.execute("SET s3_endpoint='hf.co';")
-        con.execute("SET s3_use_ssl=true;")
-        con.execute("SET s3_url_style='path';")
-        con.execute(f"SET s3_access_key_id='{self.HF_TOKEN}';")
-        con.execute(f"SET s3_secret_access_key='{self.HF_TOKEN}';")
-        
-        return con
-        
-        # Initialize virtual GPU driver with our architecture
-        self.gpu_driver = VirtualGPUDriver(
-            num_gpus=8,  # 8 virtual GPUs
-            num_sms_per_gpu=1500,  # 1500 SMs per GPU
-            cores_per_sm=128  # 128 cores per SM
-        )
-        
-        self.init_virtual_hardware()
-        
-        # MMIO regions for device communication
-        self.mmio_regions = {
-            'gpu_cmd': 0xF0000000,
-            'gpu_fb': 0xF1000000,
-            'gpu_status': 0xF3000000
-        }
-        
-    def init_virtual_hardware(self):
-        """Initialize virtual hardware components"""
-        self._init_cpu_state()
-        self._init_virtual_gpu()
-        self._init_virtual_vram()
-        
-    def _init_cpu_state(self):
-        """Initialize CPU state tracking with EnhancedCPU and GPU integration"""
-        self.cpu_groups = {
-            CPUGroupType.UI_DISPLAY: (0, 499),
-            CPUGroupType.COMPUTATION: (500, 999),
-            CPUGroupType.IO_STORAGE: (1000, 1499),
-            CPUGroupType.SYSTEM_TASKS: (1500, 1999)
-        }
-        
-        # Initialize HAL for GPU integration
-        self.hal = HardwareAbstractionLayer()
-        
-        # Initialize virtual CPUs for each group
-        self.virtual_cpus = {}
-        for group_type, (start, end) in self.cpu_groups.items():
-            for cpu_id in range(start, end + 1):
-                # Create CPU with shared GPU driver instance
-                cpu = EnhancedCPU(
-                    cpu_id=cpu_id,
-                    group_type=group_type,
-                    gpu_driver=self.gpu_driver
-                )
-                self.virtual_cpus[cpu_id] = cpu
-                
-        # Initialize state tracking in DuckDB
-        self.con.execute("""
-            CREATE TABLE IF NOT EXISTS cpu_states (
-                cpu_id INTEGER,
-                core_id INTEGER,
-                thread_id INTEGER,
-                state JSON,
-                registers JSON,
-                last_instruction INTEGER,
-                group_type VARCHAR,
-                timestamp TIMESTAMP,
-                PRIMARY KEY (cpu_id, core_id, thread_id)
-            )
-        """)
-        
-    def _init_virtual_gpu(self):
-        """Initialize virtual GPU state"""
-        self.con.execute("""
-            CREATE TABLE IF NOT EXISTS gpu_state (
-                device_id INTEGER PRIMARY KEY,
-                command_buffer JSON,
-                framebuffer BLOB,
-                status JSON,
-                timestamp TIMESTAMP
-            )
-        """)
-        
-    def _init_virtual_vram(self):
-        """Initialize virtual VRAM mapping"""
-        self.con.execute("""
-            CREATE TABLE IF NOT EXISTS vram_mapping (
-                address BIGINT PRIMARY KEY,
-                size INTEGER,
-                content BLOB,
-                flags INTEGER,
-                last_access TIMESTAMP
-            )
-        """)
-        
-    def load_config(self, config_path: str) -> Dict:
-        """Load VM configuration"""
-        with open(config_path) as f:
-            config = json.load(f)
-            
-        # Hardware-based configuration limits for 2025 server hardware
-        max_cpus = 8  # Maximum physical CPU sockets (high-end server)
-        max_cores = 128  # Maximum cores per CPU (e.g., future EPYC)
-        max_threads = 2  # Standard hyperthreading (2 threads per core)
-        max_memory = '8192G'  # 8TB RAM maximum for 2025 server
-        max_disk = '16384G'  # 16TB maximum storage
-        max_vram = '192G'  # Maximum VRAM (future datacenter GPU)
-        max_fps = 144  # Maximum refresh rate supported
-        
-        # Validate and set defaults with hardware limits
-        cpu_count = min(int(config.get('cpus', max_cpus)), max_cpus)
-        core_count = min(int(config.get('cores_per_cpu', max_cores)), max_cores)
-        thread_count = min(int(config.get('threads_per_core', max_threads)), max_threads)
-        fps = min(int(config.get('fps', 60)), max_fps)  # Default to 60 FPS, cap at max_fps
-        
-        # Convert memory sizes to GB for validation
-        def parse_size(size_str):
-            if isinstance(size_str, (int, float)):
-                return size_str
-            unit = size_str[-1].upper()
-            value = float(size_str[:-1])
-            if unit == 'T': value *= 1024
-            elif unit == 'G': value = value
-            elif unit == 'M': value /= 1024
-            return value
-            
-        mem_size = min(parse_size(config.get('memory', max_memory)), parse_size(max_memory))
-        disk_size = min(parse_size(config.get('disk_size', max_disk)), parse_size(max_disk))
-        vram_size = min(parse_size(config.get('vgpu_memory', max_vram)), parse_size(max_vram))
-        
-        # Set validated configuration
-        config.update({
-            'cpus': cpu_count,  # Number of CPU sockets
-            'cores_per_cpu': core_count,  # Cores per CPU
-            'threads_per_core': thread_count,  # Threads per core
-            'memory': f'{int(mem_size)}G',  # System memory
-            'disk_size': f'{int(disk_size)}G',  # Virtual disk size
-            'vgpu_memory': f'{int(vram_size)}G',  # VRAM size
-            'fps': fps,  # Display refresh rate
-        })
-        
-        # Log actual configuration
-        total_vcpus = cpu_count * core_count * thread_count
-        logger.info(f"VM Configuration:")
-        logger.info(f" - Total vCPUs: {total_vcpus} ({cpu_count} sockets × {core_count} cores × {thread_count} threads)")
-        logger.info(f" - Memory: {config['memory']}")
-        logger.info(f" - Disk Size: {config['disk_size']}")
-        logger.info(f" - VGPU Memory: {config['vgpu_memory']}")
-        
-        return config
-        
-    def create_virtual_disk(self) -> str:
-        """Create virtual disk using DuckDB backend"""
-        logger.info("Initializing DuckDB-backed virtual disk")
-        
-        # Initialize disk structure in DuckDB
-        self.con.execute("""
-            CREATE TABLE IF NOT EXISTS virtual_disk (
-                sector_id BIGINT PRIMARY KEY,
-                data BLOB,
-                flags INTEGER,
-                last_access TIMESTAMP
-            )
-        """)
-        
-        # Create disk mapping in memory
-        self.memory_manager.conn.execute("""
-            CREATE TABLE IF NOT EXISTS disk_mapping (
-                virtual_address BIGINT PRIMARY KEY,
-                sector_id BIGINT,
-                permissions INTEGER,
-                mapped_time TIMESTAMP
-            )
-        """)
-        
-        # Return special URL for QEMU to access our virtual disk
-        return f"duckdb://{self.db_url}?table=virtual_disk"
-        
-    def get_qemu_command(self, disk_path: str, iso_path: Optional[str] = None) -> str:
-        """Generate QEMU command with all hardware configuration"""
-        # Calculate maximum CPUs for our virtual hardware
-        total_cpus = (
-            int(self.config.get('max_cpus', 8)) *
-            int(self.config.get('max_cores', 128)) *
-            int(self.config.get('max_threads', 2))
-        )
-        
-        # Start our virtual memory management service
-        self._start_memory_service()
-        
-        cmd = [
-            "qemu-system-x86_64",
-            
-            # Machine configuration - using our virtual hardware
-            "-machine microvm",  # Lightweight VM without emulated BIOS/hardware
-            "-nodefaults",  # Don't create default devices
-            "-no-acpi",    # Using our own ACPI implementation
-            
-            # Enhanced Virtual CPU configuration
-            "-cpu", "custom,vendor=Virtual,family=2,model=1,stepping=1",
-            "-cpu-cores", f"{self.config['cores_per_cpu']}",
-            "-cpu-threads", f"{self.config['threads_per_core']}",
-            "-cpu-features", "virtual-insns=on,virtual-mmu=on,virtual-timer=on",  # Virtual CPU features
-            "-cpu-grid", "enhanced-grid=on,cpu-groups=4",  # Enhanced CPU features
-            "-cpu-group-sizes", f"{','.join(str(end-start+1) for start, end in self.cpu_groups.values())}",
-            "-no-hpet", "-no-tsc"  # Using pure virtual timing
-            
-            # CPU topology using our virtual cores
-            "-smp", f"cpus={self.config['cpus']},"
-            f"cores={self.config['cores_per_cpu']},"
-            f"threads={self.config['threads_per_core']},"
-            f"sockets=1,maxcpus={total_cpus}"
-            
-            # Memory configuration using our virtual memory manager
-            "-object memory-backend-ram,"
-            f"size={self.config['memory']},id=ram0",
-            "-object rng-random,filename=/dev/urandom,id=rng0",
-            "-device virtio-rng-pci,rng=rng0,bus=pcie.0",
-            "-numa node,memdev=ram0,nodeid=0",
-            
-            # Enable our memory management daemon
-            "-chardev socket,id=memory-backend,"
-            "path=/tmp/memory-backend.sock,server=on,wait=off",
-            "-object memory-backend-proxy,id=mem1,"
-            "chardev=memory-backend,size=8G",
-            
-            # Pure virtual GPU configuration
-            "-device virtio-gpu-pci,"  # Use virtio for virtual GPU
-            "id=gpu0,max_outputs=1,"  # Single display output
-            "virtual-gpu=on,"  # Enable virtual GPU mode
-            f"bus=pcie.0,addr=0x2,"
-            "virtual-render-api=custom,"  # Use our custom render API
-            f"virtual-gpu-path={self.hal.get_gpu_path()},"  # Virtual device path
-            
-            # VRAM configuration using IVSHMEM
-            "-object memory-backend-file,"
-            f"size={self.config['vgpu_memory']},"
-            "id=vram0,share=on",
-            "-device ivshmem-plain,"
-            "memdev=vram0,"
-            f"bus=pcie.0,addr=0x3",
-            
-            # MMIO region mapping
-            "-device ivshmem-doorbell,"
-            f"vectors={len(self.mmio_regions)},"
-            "id=shmem0",
-            
-            # Storage configuration
-            f"-drive file={disk_path},"
-            "format=raw,"  # Use raw format for DuckDB backend
-            "if=none,id=drive0,"
-            "aio=native,cache.direct=on",  # Enable direct I/O
-            "-device virtio-blk-pci,"
-            "drive=drive0,bootindex=1",
-            
-            # Network configuration
-            "-netdev user,id=net0",
-            "-device virtio-net-pci,netdev=net0",
-            
-            # Display and monitoring
-            "-display none",  # Headless mode
-            f"-vnc :{self.vnc_port - 5900}",  # VNC for remote access
-            "-monitor unix:qemu-monitor-socket,server,nowait",
-            
-            # Debug and performance options
-            "-no-user-config",  # Don't load user config
-            "-nodefaults",  # Don't add default devices
-            "-global kvm-pit.lost_tick_policy=delay",  # Better timing
-            "-rtc base=localtime,clock=host",  # Use host clock
-            "-boot strict=off"  # Flexible boot options
-        ]
-        
-        # Add ISO if installing
-        if iso_path:
-            cmd.extend([
-                "-cdrom", iso_path,
-                "-boot", "d"
-            ])
-            
-        return " ".join(cmd)
-        
-    async def install_os(self, iso_url: str):
-        """Install OS from ISO with virtual hardware support"""
-        import aiohttp
-        import aiofiles
-        import hashlib
-        
-        # Initialize virtual hardware first
-        await self._init_virtual_hardware()
-        
-        # Download ISO if needed
-        iso_path = os.path.join(os.path.dirname(__file__), 'iso', 'os.iso')
-        os.makedirs(os.path.dirname(iso_path), exist_ok=True)
-        
-        if not os.path.exists(iso_path):
-            logger.info(f"Downloading ISO from {iso_url}")
-            try:
-                async with aiohttp.ClientSession() as session:
-                    async with session.get(iso_url) as response:
-                        if response.status != 200:
-                            raise RuntimeError(f"Failed to download ISO: {response.status}")
-                            
-                        # Stream download with progress
-                        total_size = int(response.headers.get('content-length', 0))
-                        chunk_size = 8192
-                        downloaded = 0
-                        
-                        async with aiofiles.open(iso_path, 'wb') as f:
-                            async for chunk in response.content.iter_chunked(chunk_size):
-                                await f.write(chunk)
-                                downloaded += len(chunk)
-                                progress = (downloaded / total_size) * 100
-                                logger.info(f"Download progress: {progress:.1f}%")
-                                
-                # Verify download
-                async with aiofiles.open(iso_path, 'rb') as f:
-                    content = await f.read()
-                    checksum = hashlib.sha256(content).hexdigest()
-                    logger.info(f"ISO downloaded. SHA256: {checksum}")
-                    
-            except Exception as e:
-                if os.path.exists(iso_path):
-                    os.remove(iso_path)
-                raise RuntimeError(f"Failed to download ISO: {e}")
-                
-        # Create virtual disk and map hardware
-        disk_url = self.create_virtual_disk()
-        self._map_virtual_hardware()
-        
-        # Prepare for installation
-        logger.info("Preparing virtual hardware for OS installation...")
-        
-        # Configure GPU for installation
-        await self.hal.set_gpu_mode_async('install')
-        
-        # Start hardware monitoring
-        monitor_task = asyncio.create_task(self._monitor_hardware())
-        
-        # Start QEMU with ISO and virtual hardware
-        cmd = self.get_qemu_command(disk_url, iso_path)
-        logger.info("Starting OS installation with virtual hardware...")
-        
-        # Set environment for virtual devices during install
-        env = os.environ.copy()
-        env.update({
-            'VGPU_DB_URL': self.db_url,
-            'VGPU_MMIO_CMD': hex(self.mmio_regions['gpu_cmd']),
-            'VGPU_MMIO_FB': hex(self.mmio_regions['gpu_fb']),
-            'VGPU_MMIO_STATUS': hex(self.mmio_regions['gpu_status']),
-            'VRAM_SIZE': self.config['vgpu_memory'],
-            'CPU_TOPOLOGY': json.dumps({
-                'cpus': self.config['cpus'],
-                'cores': self.config['cores_per_cpu'],
-                'threads': self.config['threads_per_core']
-            }),
-            'INSTALL_MODE': '1'  # Signal we're in installation mode
-        })
-        
-        self.vm_process = subprocess.Popen(
-            cmd.split(),
-            env=env,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE
-        )
-        
-        # Wait for hardware initialization
-        try:
-            await asyncio.wait_for(self._wait_for_hardware_init(), timeout=30.0)
-        except asyncio.TimeoutError:
-            logger.error("Hardware initialization timeout during installation")
-            await self.shutdown()
-            raise RuntimeError("Failed to initialize virtual hardware for installation")
-            
-        logger.info("Virtual hardware initialized for installation")
-        
-        # Start monitoring installation progress
-        asyncio.create_task(self._monitor_installation())
-        
-        return self.vnc_port
-        
-    async def _monitor_installation(self):
-        """Monitor OS installation progress"""
-        while self.vm_process and self.vm_process.poll() is None:
-            try:
-                # Check installation progress through QEMU QMP
-                async with self.hal.get_qmp_connection() as qmp:
-                    info = await qmp.execute("query-block")
-                    for device in info:
-                        if device.get('device') == 'drive0':
-                            progress = device.get('inserted', {}).get('progress', 0)
-                            if progress > 0:
-                                logger.info(f"Installation progress: {progress}%")
-                                
-                # Check virtual hardware status
-                gpu_state = await self.hal.get_gpu_state_async()
-                if gpu_state['status'].get('installation_complete'):
-                    logger.info("OS installation completed successfully")
-                    return
-                    
-            except Exception as e:
-                logger.warning(f"Installation monitoring error: {e}")
-                
-            await asyncio.sleep(5)  # Check every 5 seconds
-        
-    async def boot_os(self):
-        """Boot existing OS installation with virtual hardware support"""
-        # Initialize virtual hardware
-        await self._init_virtual_hardware()
-        
-        # Get virtual disk URL from DuckDB
-        disk_url = self.create_virtual_disk()
-        
-        # Map MMIO regions
-        self._map_virtual_hardware()
-        
-        # Start hardware monitoring
-        monitor_task = asyncio.create_task(self._monitor_hardware())
-        
-        # Start QEMU with virtual hardware configuration
-        cmd = self.get_qemu_command(disk_url)
-        logger.info("Booting OS with virtual hardware...")
-        
-        # Set environment for virtual devices
-        env = os.environ.copy()
-        env.update({
-            'VGPU_DB_URL': self.db_url,
-            'VGPU_MMIO_CMD': hex(self.mmio_regions['gpu_cmd']),
-            'VGPU_MMIO_FB': hex(self.mmio_regions['gpu_fb']),
-            'VGPU_MMIO_STATUS': hex(self.mmio_regions['gpu_status']),
-            'VRAM_SIZE': self.config['vgpu_memory'],
-            'CPU_TOPOLOGY': json.dumps({
-                'cpus': self.config['cpus'],
-                'cores': self.config['cores_per_cpu'],
-                'threads': self.config['threads_per_core']
-            })
-        })
-        
-        self.vm_process = subprocess.Popen(
-            cmd.split(),
-            env=env,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE
-        )
-        
-        # Wait for hardware initialization
-        try:
-            await asyncio.wait_for(self._wait_for_hardware_init(), timeout=30.0)
-        except asyncio.TimeoutError:
-            logger.error("Hardware initialization timeout")
-            await self.shutdown()
-            raise RuntimeError("Failed to initialize virtual hardware")
-            
-        logger.info("Virtual hardware initialized successfully")
-        return self.vnc_port
-        
-    async def _wait_for_hardware_init(self):
-        """Wait for virtual hardware to initialize"""
-        while True:
-            try:
-                # Check GPU status
-                gpu_state = await self.hal.get_gpu_state_async()
-                if gpu_state['status'].get('initialized'):
-                    # Check VRAM mapping
-                    vram_status = await self.hal.get_vram_status_async()
-                    if vram_status['mapped']:
-                        # Check CPU topology
-                        cpu_status = await self.hal.get_cpu_status_async()
-                        if cpu_status['topology_set']:
-                            return
-            except Exception as e:
-                logger.warning(f"Hardware check failed: {e}")
-            await asyncio.sleep(0.1)
-            
-    async def _monitor_hardware(self):
-        """Monitor virtual hardware state"""
-        while self.vm_process and self.vm_process.poll() is None:
-            try:
-                # Update Enhanced CPU states
-                for cpu_id, cpu in self.virtual_cpus.items():
-                    cpu_state = cpu.get_state()
-                    for core_id in range(cpu.virtual_cpu.core_count):
-                        for thread_id in range(cpu.virtual_cpu.thread_count):
-                            thread_state = cpu_state['cores'][core_id]['threads'][thread_id]
-                            await self.memory_manager.conn.execute_async("""
-                                INSERT OR REPLACE INTO cpu_states 
-                                VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
-                            """, (cpu_id, core_id, thread_id,
-                                 json.dumps(thread_state['state']),
-                                 json.dumps(thread_state['registers']),
-                                 thread_state['last_instruction'],
-                                 cpu.virtual_cpu.group_type.name))
-                                 
-                # Update GPU state
-                gpu_state = await self.hal.get_gpu_state_async()
-                await self.memory_manager.conn.execute_async("""
-                    INSERT OR REPLACE INTO gpu_state
-                    VALUES (0, ?, ?, ?, CURRENT_TIMESTAMP)
-                """, (json.dumps(gpu_state['cmd_buffer']),
-                      gpu_state['framebuffer'],
-                      json.dumps(gpu_state['status'])))
-                      
-                # Update VRAM mappings
-                dirty_pages = await self.hal.get_dirty_vram_pages_async()
-                for addr, data in dirty_pages.items():
-                    await self.memory_manager.conn.execute_async("""
-                        INSERT OR REPLACE INTO vram_mapping
-                        VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
-                    """, (addr, len(data), data,
-                          await self.hal.get_vram_flags_async(addr)))
-                          
-            except Exception as e:
-                logger.error(f"Hardware monitoring error: {e}")
-                
-            await asyncio.sleep(0.01)  # 10ms polling interval
-            
-    def _start_memory_service(self):
-        """Start the virtual memory management service"""
-        if os.path.exists(self.memory_socket):
-            os.remove(self.memory_socket)
-            
-        # Start memory management daemon
-        self.memory_service = subprocess.Popen([
-            sys.executable,
-            "-m", "virtual_gpu_driver.src.memory.memory_daemon",
-            "--socket", self.memory_socket,
-            "--db-url", self.db_url,
-            "--size", self.config['memory']
-        ])
-        
-        # Wait for socket to be created
-        timeout = 30
-        while timeout > 0 and not os.path.exists(self.memory_socket):
-            time.sleep(0.1)
-            timeout -= 0.1
-            
-        if not os.path.exists(self.memory_socket):
-            raise RuntimeError("Memory management service failed to start")
-            
-        logger.info("Virtual memory management service started")
-        
-    def _stop_memory_service(self):
-        """Stop the virtual memory management service"""
-        if self.memory_service:
-            self.memory_service.terminate()
-            self.memory_service.wait()
-            self.memory_service = None
-            
-        if os.path.exists(self.memory_socket):
-            os.remove(self.memory_socket)
-            
-    async def shutdown(self):
-        """Gracefully shutdown VM and save hardware state"""
-        if self.vm_process:
-            logger.info("Initiating graceful shutdown...")
-            
-            try:
-                # Save final hardware state
-                await self._save_hardware_state()
-                
-                # Send ACPI shutdown through QEMU monitor
-                async with self.hal.get_qmp_connection() as qmp:
-                    await qmp.execute("system_powerdown")
-                    
-                # Wait for VM to shutdown gracefully
-                try:
-                    await asyncio.wait_for(
-                        self.vm_process.wait(),
-                        timeout=30.0
-                    )
-                except asyncio.TimeoutError:
-                    logger.warning("Graceful shutdown timed out, forcing...")
-                    self.vm_process.kill()
-                    
-            except Exception as e:
-                logger.error(f"Error during shutdown: {e}")
-                self.vm_process.kill()
-                
-            finally:
-                self.vm_process = None
-                # Stop memory management service
-                self._stop_memory_service()
-                logger.info("VM and services shutdown complete")
-                
-    async def _save_hardware_state(self):
-        """Save final hardware state before shutdown"""
-        try:
-            # Get final CPU states
-            cpu_states = await self.hal.get_cpu_states_async()
-            for cpu_id, cores in cpu_states.items():
-                for core_id, threads in cores.items():
-                    for thread_id, state in threads.items():
-                        await self.memory_manager.conn.execute_async("""
-                            INSERT OR REPLACE INTO cpu_states 
-                            VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
-                        """, (cpu_id, core_id, thread_id,
-                             json.dumps(state['state']),
-                             json.dumps(state['registers']),
-                             state['last_instruction']))
-                             
-            # Get final GPU state
-            gpu_state = await self.hal.get_gpu_state_async()
-            await self.memory_manager.conn.execute_async("""
-                INSERT OR REPLACE INTO gpu_state
-                VALUES (0, ?, ?, ?, CURRENT_TIMESTAMP)
-            """, (json.dumps(gpu_state['cmd_buffer']),
-                  gpu_state['framebuffer'],
-                  json.dumps(gpu_state['status'])))
-                  
-            # Commit any pending changes
-            await self.memory_manager.conn.commit_async()
-            
-        except Exception as e:
-            logger.error(f"Error saving hardware state: {e}")
-            raise
-            
-    async def __aenter__(self):
-        return self
-        
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        await self.shutdown()
-
-async def main():
-    """Test QEMU manager"""
-    async with QEMUManager() as qemu:
-        # For testing, try to boot existing OS or install new one
-        try:
-            vnc_port = await qemu.boot_os()
-            logger.info(f"OS booted. VNC available on port {vnc_port}")
-        except FileNotFoundError:
-            # No OS installed, do fresh install
-            iso_url = "http://releases.ubuntu.com/22.04/ubuntu-22.04-desktop-amd64.iso"
-            vnc_port = await qemu.install_os(iso_url)
-            logger.info(f"Installing OS. Monitor installation on VNC port {vnc_port}")
-            
-        # Keep running until interrupted
-        try:
-            await asyncio.Event().wait()
-        except KeyboardInterrupt:
-            logger.info("Shutting down...")
-
-if __name__ == "__main__":
-    asyncio.run(main())
+"""
+QEMU Virtual Machine Manager with Direct DuckDB Storage and Virtual Hardware Integration
+Manages VM state, virtual GPU, VRAM, and CPU states through remote DuckDB backend
+"""
+
+import os
+import sys
+import asyncio
+import subprocess
+import time
+from pathlib import Path
+from typing import Dict, Optional, List
+import json
+import logging
+import duckdb
+from datetime import datetime
+
+# Add parent directory to path for imports
+sys.path.append(str(Path(__file__).parent.parent))
+
+from virtual_gpu_driver.src.driver_api import GPUError, VirtualGPUDriver
+from virtual_gpu_driver.src.memory.duckdb_memory_manager import DuckDBMemoryManager
+from virtual_gpu_driver.src.hal.hal import HardwareAbstractionLayer
+from vram.remote_storage import RemoteStorageManager
+from cpu.enhanced_cpu import EnhancedCPU, CPUGroupType, VirtualCPU, CPUInstruction, InstructionType
+
+# Configure logging
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+class QEMUManager:
+    
+    def __init__(self, config_path: str = "config.json"):
+        self.config = self.load_config(str(Path(__file__).parent / config_path))
+        self.vm_process: Optional[subprocess.Popen] = None
+        self.monitor_socket = None
+        self.vnc_port = 5900
+        self.memory_service = None
+        
+        # Initialize storage and hardware components
+        self.db_url = "hf://datasets/Fred808/helium/storage.json"
+        self.con = self._init_db_connection()
+        self.memory_manager = DuckDBMemoryManager(self.db_url)
+        self.storage_manager = RemoteStorageManager(self.db_url)
+        
+        # Memory service socket path
+        self.memory_socket = "/tmp/memory-backend.sock"
+        
+    def _init_db_connection(self) -> duckdb.DuckDBPyConnection:
+        """Initialize database connection with HuggingFace configuration"""
+        con = duckdb.connect(self.db_url)
+        
+        # Configure HuggingFace access
+        con.execute("INSTALL httpfs;")
+        con.execute("LOAD httpfs;")
+        con.execute("SET s3_endpoint='hf.co';")
+        con.execute("SET s3_use_ssl=true;")
+        con.execute("SET s3_url_style='path';")
+        con.execute(f"SET s3_access_key_id='{self.HF_TOKEN}';")
+        con.execute(f"SET s3_secret_access_key='{self.HF_TOKEN}';")
+
+        return ConnectionRefusedError
+        
+        # Initialize virtual GPU driver with our architecture
+        self.gpu_driver = VirtualGPUDriver(
+            num_gpus=8,  # 8 virtual GPUs
+            num_sms_per_gpu=1500,  # 1500 SMs per GPU
+            cores_per_sm=128  # 128 cores per SM
+        )
+        
+        self.init_virtual_hardware()
+        
+        # MMIO regions for device communication
+        self.mmio_regions = {
+            'gpu_cmd': 0xF0000000,
+            'gpu_fb': 0xF1000000,
+            'gpu_status': 0xF3000000
+        }
+        
+    def init_virtual_hardware(self):
+        """Initialize virtual hardware components"""
+        self._init_cpu_state()
+        self._init_virtual_gpu()
+        self._init_virtual_vram()
+        
+    def _init_cpu_state(self):
+        """Initialize CPU state tracking with EnhancedCPU and GPU integration"""
+        self.cpu_groups = {
+            CPUGroupType.UI_DISPLAY: (0, 499),
+            CPUGroupType.COMPUTATION: (500, 999),
+            CPUGroupType.IO_STORAGE: (1000, 1499),
+            CPUGroupType.SYSTEM_TASKS: (1500, 1999)
+        }
+        
+        # Initialize HAL for GPU integration
+        self.hal = HardwareAbstractionLayer()
+        
+        # Initialize virtual CPUs for each group
+        self.virtual_cpus = {}
+        for group_type, (start, end) in self.cpu_groups.items():
+            for cpu_id in range(start, end + 1):
+                # Create CPU with shared GPU driver instance
+                cpu = EnhancedCPU(
+                    cpu_id=cpu_id,
+                    group_type=group_type,
+                    gpu_driver=self.gpu_driver
+                )
+                self.virtual_cpus[cpu_id] = cpu
+                
+        # Initialize state tracking in DuckDB
+        self.con.execute("""
+            CREATE TABLE IF NOT EXISTS cpu_states (
+                cpu_id INTEGER,
+                core_id INTEGER,
+                thread_id INTEGER,
+                state JSON,
+                registers JSON,
+                last_instruction INTEGER,
+                group_type VARCHAR,
+                timestamp TIMESTAMP,
+                PRIMARY KEY (cpu_id, core_id, thread_id)
+            )
+        """)
+        
+    def _init_virtual_gpu(self):
+        """Initialize virtual GPU state"""
+        self.con.execute("""
+            CREATE TABLE IF NOT EXISTS gpu_state (
+                device_id INTEGER PRIMARY KEY,
+                command_buffer JSON,
+                framebuffer BLOB,
+                status JSON,
+                timestamp TIMESTAMP
+            )
+        """)
+        
+    def _init_virtual_vram(self):
+        """Initialize virtual VRAM mapping"""
+        self.con.execute("""
+            CREATE TABLE IF NOT EXISTS vram_mapping (
+                address BIGINT PRIMARY KEY,
+                size INTEGER,
+                content BLOB,
+                flags INTEGER,
+                last_access TIMESTAMP
+            )
+        """)
+        
+    def load_config(self, config_path: str) -> Dict:
+        """Load VM configuration"""
+        with open(config_path) as f:
+            config = json.load(f)
+            
+        # Hardware-based configuration limits for 2025 server hardware
+        max_cpus = 8  # Maximum physical CPU sockets (high-end server)
+        max_cores = 128  # Maximum cores per CPU (e.g., future EPYC)
+        max_threads = 2  # Standard hyperthreading (2 threads per core)
+        max_memory = '8192G'  # 8TB RAM maximum for 2025 server
+        max_disk = '16384G'  # 16TB maximum storage
+        max_vram = '192G'  # Maximum VRAM (future datacenter GPU)
+        max_fps = 144  # Maximum refresh rate supported
+        
+        # Validate and set defaults with hardware limits
+        cpu_count = min(int(config.get('cpus', max_cpus)), max_cpus)
+        core_count = min(int(config.get('cores_per_cpu', max_cores)), max_cores)
+        thread_count = min(int(config.get('threads_per_core', max_threads)), max_threads)
+        fps = min(int(config.get('fps', 60)), max_fps)  # Default to 60 FPS, cap at max_fps
+        
+        # Convert memory sizes to GB for validation
+        def parse_size(size_str):
+            if isinstance(size_str, (int, float)):
+                return size_str
+            unit = size_str[-1].upper()
+            value = float(size_str[:-1])
+            if unit == 'T': value *= 1024
+            elif unit == 'G': value = value
+            elif unit == 'M': value /= 1024
+            return value
+            
+        mem_size = min(parse_size(config.get('memory', max_memory)), parse_size(max_memory))
+        disk_size = min(parse_size(config.get('disk_size', max_disk)), parse_size(max_disk))
+        vram_size = min(parse_size(config.get('vgpu_memory', max_vram)), parse_size(max_vram))
+        
+        # Set validated configuration
+        config.update({
+            'cpus': cpu_count,  # Number of CPU sockets
+            'cores_per_cpu': core_count,  # Cores per CPU
+            'threads_per_core': thread_count,  # Threads per core
+            'memory': f'{int(mem_size)}G',  # System memory
+            'disk_size': f'{int(disk_size)}G',  # Virtual disk size
+            'vgpu_memory': f'{int(vram_size)}G',  # VRAM size
+            'fps': fps,  # Display refresh rate
+        })
+        
+        # Log actual configuration
+        total_vcpus = cpu_count * core_count * thread_count
+        logger.info(f"VM Configuration:")
+        logger.info(f" - Total vCPUs: {total_vcpus} ({cpu_count} sockets × {core_count} cores × {thread_count} threads)")
+        logger.info(f" - Memory: {config['memory']}")
+        logger.info(f" - Disk Size: {config['disk_size']}")
+        logger.info(f" - VGPU Memory: {config['vgpu_memory']}")
+        
+        return config
+        
+    def create_virtual_disk(self) -> str:
+        """Create virtual disk using DuckDB backend"""
+        logger.info("Initializing DuckDB-backed virtual disk")
+        
+        # Initialize disk structure in DuckDB
+        self.con.execute("""
+            CREATE TABLE IF NOT EXISTS virtual_disk (
+                sector_id BIGINT PRIMARY KEY,
+                data BLOB,
+                flags INTEGER,
+                last_access TIMESTAMP
+            )
+        """)
+        
+        # Create disk mapping in memory
+        self.memory_manager.conn.execute("""
+            CREATE TABLE IF NOT EXISTS disk_mapping (
+                virtual_address BIGINT PRIMARY KEY,
+                sector_id BIGINT,
+                permissions INTEGER,
+                mapped_time TIMESTAMP
+            )
+        """)
+        
+        # Return special URL for QEMU to access our virtual disk
+        return f"duckdb://{self.db_url}?table=virtual_disk"
+        
+    def get_qemu_command(self, disk_path: str, iso_path: Optional[str] = None) -> str:
+        """Generate QEMU command with all hardware configuration"""
+        # Calculate maximum CPUs for our virtual hardware
+        total_cpus = (
+            int(self.config.get('max_cpus', 8)) *
+            int(self.config.get('max_cores', 128)) *
+            int(self.config.get('max_threads', 2))
+        )
+        
+        # Start our virtual memory management service
+        self._start_memory_service()
+        
+        cmd = [
+            "qemu-system-x86_64",
+            
+            # Machine configuration - using our virtual hardware
+            "-machine microvm",  # Lightweight VM without emulated BIOS/hardware
+            "-nodefaults",  # Don't create default devices
+            "-no-acpi",    # Using our own ACPI implementation
+            
+            # Enhanced Virtual CPU configuration
+            "-cpu custom,vendor=Virtual,family=2,model=1,stepping=1",
+            f"cores={self.config['cores_per_cpu']}",
+            f"threads={self.config['threads_per_core']}",
+            "virtual-insns=on,virtual-mmu=on,virtual-timer=on,",  # Virtual CPU features
+            "enhanced-grid=on,cpu-groups=4,",  # Enhanced CPU features
+            f"group-sizes={','.join(str(end-start+1) for start, end in self.cpu_groups.values())},",
+            "-no-hpet,-no-tsc",  # Using pure virtual timing
+            
+            # CPU topology using our virtual cores
+            f"-smp {self.config['cpus']}",
+            f"cores={self.config['cores_per_cpu']}",
+            f"threads={self.config['threads_per_core']}",
+            f"sockets=1,maxcpus={total_cpus}",
+            
+            # Memory configuration using our virtual memory manager
+            "-object memory-backend-ram,",
+            f"size={self.config['memory']},id=ram0",
+            "-object rng-random,filename=/dev/urandom,id=rng0",
+            "-device virtio-rng-pci,rng=rng0,bus=pcie.0",
+            "-numa node,memdev=ram0,nodeid=0",
+            
+            # Enable our memory management daemon
+            "-chardev socket,id=memory-backend,",
+            "path=/tmp/memory-backend.sock,server=on,wait=off",
+            "-object memory-backend-proxy,id=mem1,",
+            "chardev=memory-backend,size=8G",
+            
+            # Pure virtual GPU configuration
+            "-device virtio-gpu-pci,",  # Use virtio for virtual GPU
+            "id=gpu0,max_outputs=1,",  # Single display output
+            "virtual-gpu=on,",  # Enable virtual GPU mode
+            f"bus=pcie.0,addr=0x2,",
+            "virtual-render-api=custom,",  # Use our custom render API
+            f"virtual-gpu-path={self.hal.get_gpu_path()},",  # Virtual device path
+            
+            # VRAM configuration using IVSHMEM
+            "-object memory-backend-file,",
+            f"size={self.config['vgpu_memory']}",
+            "id=vram0,share=on",
+            "-device ivshmem-plain,",
+            "memdev=vram0,",
+            f"bus=pcie.0,addr=0x3",
+            
+            # MMIO region mapping
+            "-device ivshmem-doorbell,",
+            f"vectors={len(self.mmio_regions)},",
+            "id=shmem0",
+            
+            # Storage configuration
+            f"-drive file={disk_path},",
+            "format=raw,",  # Use raw format for DuckDB backend
+            "if=none,id=drive0,",
+            "aio=native,cache.direct=on",  # Enable direct I/O
+            "-device virtio-blk-pci,",
+            "drive=drive0,bootindex=1",
+            
+            # Network configuration
+            "-netdev user,id=net0",
+            "-device virtio-net-pci,netdev=net0",
+            
+            # Display and monitoring
+            "-display none",  # Headless mode
+            f"-vnc :{self.vnc_port - 5900}",  # VNC for remote access
+            "-monitor unix:qemu-monitor-socket,server,nowait",
+            
+            # Debug and performance options
+            "-no-user-config",  # Don't load user config
+            "-nodefaults",  # Don't add default devices
+            "-global kvm-pit.lost_tick_policy=delay",  # Better timing
+            "-rtc base=localtime,clock=host",  # Use host clock
+            "-boot strict=off"  # Flexible boot options
+        ]
+        
+        # Add ISO if installing
+        if iso_path:
+            cmd.extend([
+                "-cdrom", iso_path,
+                "-boot", "d"
+            ])
+            
+        return " ".join(cmd)
+        
+    async def install_os(self, iso_url: str):
+        """Install OS from ISO with virtual hardware support"""
+        import aiohttp
+        import aiofiles
+        import hashlib
+        
+        # Initialize virtual hardware first
+        await self._init_virtual_hardware()
+        
+        # Download ISO if needed
+        iso_path = os.path.join(os.path.dirname(__file__), 'iso', 'os.iso')
+        os.makedirs(os.path.dirname(iso_path), exist_ok=True)
+        
+        if not os.path.exists(iso_path):
+            logger.info(f"Downloading ISO from {iso_url}")
+            try:
+                async with aiohttp.ClientSession() as session:
+                    async with session.get(iso_url) as response:
+                        if response.status != 200:
+                            raise RuntimeError(f"Failed to download ISO: {response.status}")
+                            
+                        # Stream download with progress
+                        total_size = int(response.headers.get('content-length', 0))
+                        chunk_size = 8192
+                        downloaded = 0
+                        
+                        async with aiofiles.open(iso_path, 'wb') as f:
+                            async for chunk in response.content.iter_chunked(chunk_size):
+                                await f.write(chunk)
+                                downloaded += len(chunk)
+                                progress = (downloaded / total_size) * 100
+                                logger.info(f"Download progress: {progress:.1f}%")
+                                
+                # Verify download
+                async with aiofiles.open(iso_path, 'rb') as f:
+                    content = await f.read()
+                    checksum = hashlib.sha256(content).hexdigest()
+                    logger.info(f"ISO downloaded. SHA256: {checksum}")
+                    
+            except Exception as e:
+                if os.path.exists(iso_path):
+                    os.remove(iso_path)
+                raise RuntimeError(f"Failed to download ISO: {e}")
+                
+        # Create virtual disk and map hardware
+        disk_url = self.create_virtual_disk()
+        self._map_virtual_hardware()
+        
+        # Prepare for installation
+        logger.info("Preparing virtual hardware for OS installation...")
+        
+        # Configure GPU for installation
+        await self.hal.set_gpu_mode_async('install')
+        
+        # Start hardware monitoring
+        monitor_task = asyncio.create_task(self._monitor_hardware())
+        
+        # Start QEMU with ISO and virtual hardware
+        cmd = self.get_qemu_command(disk_url, iso_path)
+        logger.info("Starting OS installation with virtual hardware...")
+        
+        # Set environment for virtual devices during install
+        env = os.environ.copy()
+        env.update({
+            'VGPU_DB_URL': self.db_url,
+            'VGPU_MMIO_CMD': hex(self.mmio_regions['gpu_cmd']),
+            'VGPU_MMIO_FB': hex(self.mmio_regions['gpu_fb']),
+            'VGPU_MMIO_STATUS': hex(self.mmio_regions['gpu_status'])
+        })
+        
+        self.vm_process = subprocess.Popen(cmd, shell=True, env=env)
+        
+        # Wait for installation to complete
+        await self.vm_process.wait()
+        
+        # Stop monitoring
+        monitor_task.cancel()
+        
+        logger.info("OS installation finished.")
+        return self.vnc_port
+        
+    async def boot_os(self):
+        """Boot existing OS with virtual hardware"""
+        disk_path = os.path.join(os.path.dirname(__file__), 'disk', 'os.qcow2')
+        if not os.path.exists(disk_path):
+            raise FileNotFoundError("No OS installation found.")
+            
+        # Initialize and map virtual hardware
+        await self._init_virtual_hardware()
+        self._map_virtual_hardware()
+        
+        # Configure GPU for boot
+        await self.hal.set_gpu_mode_async('boot')
+        
+        # Start hardware monitoring
+        monitor_task = asyncio.create_task(self._monitor_hardware())
+        
+        # Start QEMU
+        cmd = self.get_qemu_command(disk_path)
+        logger.info("Booting OS with virtual hardware...")
+        
+        env = os.environ.copy()
+        env.update({
+            'VGPU_DB_URL': self.db_url,
+            'VGPU_MMIO_CMD': hex(self.mmio_regions['gpu_cmd']),
+            'VGPU_MMIO_FB': hex(self.mmio_regions['gpu_fb']),
+            'VGPU_MMIO_STATUS': hex(self.mmio_regions['gpu_status'])
+        })
+        
+        self.vm_process = subprocess.Popen(cmd, shell=True, env=env)
+        
+        return self.vnc_port
+        
+    async def shutdown(self):
+        """Shutdown running VM"""
+        if self.vm_process:
+            self.vm_process.terminate()
+            await self.vm_process.wait()
+            self.vm_process = None
+            logger.info("VM shut down.")
+
+
+