Update qemu/web_interface.py

qemu/web_interface.py

@@ -1,130 +1,746 @@
 """
-Web Interface for QEMU VM
-Provides browser-based access to VM display
+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
 """
 
-from fastapi import FastAPI, WebSocket, WebSocketDisconnect
-from fastapi.staticfiles import StaticFiles
-from fastapi.responses import HTMLResponse
+import os
+import sys
 import asyncio
+import subprocess
+import time
+import socket
+import json
 import logging
 from pathlib import Path
-from typing import Optional, Dict
-import json
-import os
-import sys
-
-# Add the parent directory to the Python path
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from qemu.qemu_manager import QEMUManager
+from typing import Dict, Optional, List, Union, Any
+from datetime import datetime
 
 # Configure logging
 logging.basicConfig(level=logging.DEBUG)
 logger = logging.getLogger(__name__)
 
-app = FastAPI()
-
-# Mount static files
-static_path = Path(__file__).parent / "static"
-app.mount("/static", StaticFiles(directory=str(static_path)), name="static")
-
-# Global QEMU manager instance
-qemu: Optional[QEMUManager] = None
-
-class VMConnection:
-
-    def __init__(self):
-        self.active_connections: Dict[int, WebSocket] = {}
+class QMPClient:
+    """Simple QMP (QEMU Machine Protocol) client implementation"""
+    
+    def __init__(self, address: str = 'localhost', port: int = 4444):
+        self.address = address
+        self.port = port
+        self.sock: Optional[socket.socket] = None
         
-    async def connect(self, websocket: WebSocket):
-        await websocket.accept()
-        client_id = id(websocket)
-        self.active_connections[client_id] = websocket
-        return client_id
+    async def connect(self):
+        """Connect to QEMU's QMP interface"""
+        try:
+            self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            self.sock.connect((self.address, self.port))
+            greeting = await self._receive()
+            await self._execute("qmp_capabilities")
+            return greeting
+        except Exception as e:
+            logger.error(f"QMP connection failed: {e}")
+            if self.sock:
+                self.sock.close()
+            raise
+            
+    async def _execute(self, cmd: str, args: Dict = None) -> Dict[str, Any]:
+        """Execute a QMP command"""
+        if not self.sock:
+            raise RuntimeError("Not connected")
+            
+        request = {"execute": cmd}
+        if args:
+            request["arguments"] = args
+            
+        self.sock.send(json.dumps(request).encode() + b'\n')
+        return await self._receive()
         
-    def disconnect(self, client_id: int):
-        self.active_connections.pop(client_id, None)
+    async def _receive(self) -> Dict[str, Any]:
+        """Receive and parse QMP response"""
+        if not self.sock:
+            raise RuntimeError("Not connected")
+            
+        data = self.sock.recv(4096).decode()
+        return json.loads(data)
         
-    async def broadcast(self, message: str):
-        for ws in self.active_connections.values():
-            try:
-                await ws.send_text(message)
-            except:
-                continue
+    def close(self):
+        """Close QMP connection"""
+        if self.sock:
+            self.sock.close()
+            self.sock = None
 
-vm_connection = VMConnection()
+# Import local modules using absolute imports
+from FServe.virtual_gpu_driver.src.driver_api import GPUError, VirtualGPUDriver
+from FServe.virtual_gpu_driver.src.memory.duckdb_memory_manager import DuckDBMemoryManager
+from FServe.virtual_gpu_driver.src.hal.hal import HardwareAbstractionLayer
+from FServe.vram.remote_storage import RemoteStorageManager
+from FServe.config import get_db_url, get_hf_token_cached
+from FServe.cpu.enhanced_cpu import EnhancedCPU, CPUGroupType, VirtualCPU, CPUInstruction, InstructionType
 
-@app.get("/")
-async def get_index():
-    """Serve main HTML interface"""
-    html_path = static_path / "index.html"
-    return HTMLResponse(content=html_path.read_text())
+# Configure logging
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
 
-@app.websocket("/ws/vm")
-async def vm_websocket(websocket: WebSocket):
-    """Handle VM display and control WebSocket"""
-    client_id = await vm_connection.connect(websocket)
+class QEMUManager:
     
-    try:
+    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:
-            message = await websocket.receive_json()
-            
-            if message["type"] == "install":
-                # Start fresh OS installation
-                if qemu:
-                    vnc_port = await qemu.install_os(message["iso_url"])
-                    await websocket.send_json({
-                        "type": "vnc_info",
-                        "port": vnc_port
-                    })
+            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")
                     
-            elif message["type"] == "boot":
-                # Boot existing OS
-                if qemu:
-                    try:
-                        vnc_port = await qemu.boot_os()
-                        await websocket.send_json({
-                            "type": "vnc_info",
-                            "port": vnc_port
-                        })
-                    except FileNotFoundError:
-                        await websocket.send_json({
-                            "type": "error",
-                            "message": "No OS installation found"
-                        })
-                        
-            elif message["type"] == "shutdown":
-                # Shutdown VM
-                if qemu:
-                    await qemu.shutdown()
-                    await websocket.send_json({
-                        "type": "status",
-                        "message": "VM shut down"
-                    })
+                # 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 WebSocketDisconnect:
-        vm_connection.disconnect(client_id)
-        
-@app.on_event("startup")
-async def startup_event():
-    """Initialize QEMU manager on startup"""
-    global qemu
-    qemu = QEMUManager()
-    logger.info("QEMU manager initialized")
-    
-@app.on_event("shutdown")
-async def shutdown_event():
-    """Cleanup on shutdown"""
-    if qemu:
-        await qemu.shutdown()
-        logger.info("QEMU manager shut down")
-
+            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()
 
-def main():
-    """Entry point for the FServe application"""
-    import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8080)
+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__":
-    main()
+    asyncio.run(main())