From c1da072744bf64547efba5fa5f65edfeece33fcc Mon Sep 17 00:00:00 2001 From: hsliu Date: Tue, 16 Sep 2025 16:54:27 +0800 Subject: [PATCH] del api md --- docs/api/core_module_api.md | 580 ------------------------------------ requirements.txt | 4 +- 2 files changed, 3 insertions(+), 581 deletions(-) delete mode 100644 docs/api/core_module_api.md diff --git a/docs/api/core_module_api.md b/docs/api/core_module_api.md deleted file mode 100644 index b32bcd3a9..000000000 --- a/docs/api/core_module_api.md +++ /dev/null @@ -1,580 +0,0 @@ -# Core Module API Design - -## 1. Module Overview - -**Purpose**: The core module provides fundamental scheduling, caching, and resource management functionality for vLLM-omni. - -**Responsibilities**: -- Request scheduling and prioritization across different model types -- DiT cache management for diffusion models -- Resource allocation and coordination between workers -- Inter-module communication and coordination -- Queue management and load balancing - -**Dependencies**: -- `vllm_omni.request` - Request handling and types -- `vllm_omni.config` - Configuration management -- `vllm_omni.utils` - Utility functions and helpers -- `vllm` - vLLM core components for integration - -**Integration Points**: -- Receives requests from entrypoints -- Coordinates with engine modules for model execution -- Manages worker allocation and task distribution -- Provides status and metrics to monitoring systems -- Handles cache coordination between different model types - -## 2. Core Classes/Interfaces - -```python -from abc import ABC, abstractmethod -from typing import List, Optional, Dict, Any, Union -from dataclasses import dataclass, field -from enum import Enum -import asyncio -from vllm_omni.request import OmniRequest, RequestType, ProcessingStage - -class SchedulerType(Enum): - """Types of schedulers available.""" - FIFO = "fifo" - PRIORITY = "priority" - MULTIMODAL = "multimodal" - DIFFUSION = "diffusion" - HYBRID = "hybrid" - -class CacheStrategy(Enum): - """Cache strategies for different model types.""" - LRU = "lru" - LFU = "lfu" - TTL = "ttl" - ADAPTIVE = "adaptive" - -@dataclass -class SchedulerConfig: - """Configuration for scheduler.""" - scheduler_type: SchedulerType = SchedulerType.FIFO - max_queue_size: int = 1000 - priority_weights: Dict[str, float] = field(default_factory=lambda: { - "high": 2.0, "normal": 1.0, "low": 0.5 - }) - timeout_seconds: int = 300 - batch_size: int = 32 - enable_preemption: bool = True - -@dataclass -class CacheConfig: - """Configuration for cache management.""" - max_memory_gb: float = 8.0 - strategy: CacheStrategy = CacheStrategy.LRU - ttl_seconds: int = 3600 - enable_compression: bool = True - compression_ratio: float = 0.7 - -class BaseScheduler(ABC): - """Abstract base class for all schedulers.""" - - @abstractmethod - async def schedule_request(self, request: OmniRequest) -> bool: - """Schedule a request for processing.""" - pass - - @abstractmethod - async def get_next_request(self) -> Optional[OmniRequest]: - """Get the next request to process.""" - pass - - @abstractmethod - async def remove_request(self, request_id: str) -> bool: - """Remove a request from the queue.""" - pass - - @abstractmethod - def get_queue_size(self) -> int: - """Get current queue size.""" - pass - - @abstractmethod - def get_processed_count(self) -> int: - """Get number of processed requests.""" - pass - -class DiTCacheManager: - """Manages DiT cache for diffusion models.""" - - def __init__(self, config: CacheConfig): - self.config = config - self.cache = {} - self.cache_stats = {"hits": 0, "misses": 0} - self.memory_usage = 0.0 - - async def get_cache(self, cache_key: str) -> Optional[Any]: - """Retrieve cached data.""" - pass - - async def set_cache(self, cache_key: str, data: Any) -> None: - """Store data in cache.""" - pass - - async def invalidate_cache(self, cache_key: str) -> None: - """Invalidate cached data.""" - pass - - def get_hit_rate(self) -> float: - """Get cache hit rate.""" - pass - - def get_memory_usage(self) -> float: - """Get current memory usage in GB.""" - pass - -class ResourceManager: - """Manages system resources and allocation.""" - - def __init__(self, config: Dict[str, Any]): - self.config = config - self.allocated_resources = {} - self.available_resources = {} - - async def allocate_resources(self, request: OmniRequest) -> Dict[str, Any]: - """Allocate resources for a request.""" - pass - - async def deallocate_resources(self, request_id: str) -> None: - """Deallocate resources for a request.""" - pass - - def get_resource_status(self) -> Dict[str, Any]: - """Get current resource allocation status.""" - pass -``` - -## 3. Public API Methods - -### CoreScheduler Class - -```python -class OmniScheduler(SchedulerInterface): - """Main scheduler for vLLM-omni core module.""" - - def __init__(self, config: SchedulerConfig, cache_config: CacheConfig): - """ - Initialize the core scheduler. - - Args: - config: Scheduler configuration - cache_config: Cache configuration - """ - self.config = config - self.cache_manager = DiTCacheManager(cache_config) - self.resource_manager = ResourceManager({}) - self.scheduler = self._create_scheduler() - self._running = False - self._stats = { - "total_requests": 0, - "processed_requests": 0, - "failed_requests": 0, - "average_processing_time": 0.0 - } - - def _create_scheduler(self) -> BaseScheduler: - """Factory method to create appropriate scheduler.""" - if self.config.scheduler_type == SchedulerType.FIFO: - return FIFOScheduler(self.config) - elif self.config.scheduler_type == SchedulerType.PRIORITY: - return PriorityScheduler(self.config) - elif self.config.scheduler_type == SchedulerType.MULTIMODAL: - return MultimodalScheduler(self.config) - elif self.config.scheduler_type == SchedulerType.DIFFUSION: - return DiffusionScheduler(self.config) - elif self.config.scheduler_type == SchedulerType.HYBRID: - return HybridScheduler(self.config) - else: - raise ValueError(f"Unknown scheduler type: {self.config.scheduler_type}") -``` - -#### Core Operations - -```python - async def schedule_request(self, request: OmniRequest) -> bool: - """ - Schedule a request for processing. - - Args: - request: The request to schedule - - Returns: - bool: True if successfully scheduled, False otherwise - - Raises: - SchedulerError: If scheduling fails - QueueFullError: If queue is at capacity - ResourceError: If insufficient resources - """ - try: - # Validate request - if not self._validate_request(request): - return False - - # Check resource availability - if not await self.resource_manager.can_allocate(request): - raise ResourceError("Insufficient resources") - - # Schedule the request - success = await self.scheduler.schedule_request(request) - if success: - self._stats["total_requests"] += 1 - await self.resource_manager.allocate_resources(request) - - return success - - except Exception as e: - self._stats["failed_requests"] += 1 - raise SchedulerError(f"Failed to schedule request: {e}") - - async def get_next_request(self) -> Optional[OmniRequest]: - """ - Get the next request to process. - - Returns: - OmniRequest or None: Next request to process - - Raises: - SchedulerError: If retrieval fails - """ - try: - request = await self.scheduler.get_next_request() - if request: - request.update_processing_stage(ProcessingStage.PREPROCESSING) - return request - except Exception as e: - raise SchedulerError(f"Failed to get next request: {e}") - - async def process_request(self, request: OmniRequest) -> Any: - """ - Process a request through the appropriate pipeline. - - Args: - request: The request to process - - Returns: - Any: Processing result - - Raises: - ProcessingError: If processing fails - """ - start_time = time.time() - try: - # Update processing stage - request.update_processing_stage(ProcessingStage.AR_GENERATION) - - # Check cache first - cache_key = request.generate_cache_key() - cached_result = await self.cache_manager.get_cache(cache_key) - if cached_result: - return cached_result - - # Process based on request type - if request.request_type == RequestType.TEXT: - result = await self._process_text_request(request) - elif request.request_type == RequestType.IMAGE: - result = await self._process_image_request(request) - elif request.request_type == RequestType.MULTIMODAL: - result = await self._process_multimodal_request(request) - elif request.request_type == RequestType.DIFFUSION: - result = await self._process_diffusion_request(request) - else: - raise ProcessingError(f"Unknown request type: {request.request_type}") - - # Cache the result - await self.cache_manager.set_cache(cache_key, result) - - # Update statistics - processing_time = time.time() - start_time - self._update_stats(processing_time) - - request.update_processing_stage(ProcessingStage.COMPLETED) - return result - - except Exception as e: - request.add_error(str(e)) - self._stats["failed_requests"] += 1 - raise ProcessingError(f"Failed to process request: {e}") - finally: - # Cleanup resources - await self.resource_manager.deallocate_resources(request.request_id) -``` - -#### Configuration Methods - -```python - def update_config(self, new_config: SchedulerConfig) -> None: - """ - Update scheduler configuration. - - Args: - new_config: New configuration to apply - - Raises: - ConfigError: If configuration is invalid - """ - try: - new_config.validate() - self.config = new_config - # Recreate scheduler with new config - self.scheduler = self._create_scheduler() - except Exception as e: - raise ConfigError(f"Failed to update config: {e}") - - def get_config(self) -> SchedulerConfig: - """Get current configuration.""" - return self.config - - def update_cache_config(self, new_config: CacheConfig) -> None: - """Update cache configuration.""" - self.cache_manager.config = new_config -``` - -#### Lifecycle Management - -```python - async def start(self) -> None: - """Start the scheduler.""" - if self._running: - raise SchedulerError("Scheduler is already running") - - self._running = True - # Start background tasks - asyncio.create_task(self._background_cleanup()) - asyncio.create_task(self._monitor_resources()) - - async def stop(self) -> None: - """Stop the scheduler gracefully.""" - if not self._running: - return - - self._running = False - # Wait for current requests to complete - await self._wait_for_completion() - # Cleanup resources - await self._cleanup_resources() - - async def shutdown(self) -> None: - """Force shutdown the scheduler.""" - self._running = False - # Immediate cleanup - await self._force_cleanup() -``` - -#### Monitoring Methods - -```python - def get_status(self) -> Dict[str, Any]: - """Get current scheduler status.""" - return { - "running": self._running, - "queue_size": self.scheduler.get_queue_size(), - "processed_requests": self.scheduler.get_processed_count(), - "cache_hit_rate": self.cache_manager.get_hit_rate(), - "memory_usage": self.cache_manager.get_memory_usage(), - "resource_status": self.resource_manager.get_resource_status(), - "stats": self._stats.copy() - } - - def get_metrics(self) -> Dict[str, float]: - """Get performance metrics.""" - return { - "throughput": self._calculate_throughput(), - "average_latency": self._stats["average_processing_time"], - "cache_hit_rate": self.cache_manager.get_hit_rate(), - "memory_utilization": self.cache_manager.get_memory_usage() / self.cache_manager.config.max_memory_gb, - "queue_utilization": self.scheduler.get_queue_size() / self.config.max_queue_size - } - - def get_health_status(self) -> Dict[str, Any]: - """Get health status of the scheduler.""" - return { - "healthy": self._is_healthy(), - "issues": self._get_health_issues(), - "recommendations": self._get_recommendations() - } -``` - -## 4. Configuration - -```python -@dataclass -class CoreModuleConfig: - """Complete configuration for the core module.""" - - # Scheduler configuration - scheduler: SchedulerConfig = field(default_factory=SchedulerConfig) - - # Cache configuration - cache: CacheConfig = field(default_factory=CacheConfig) - - # Resource limits - max_memory_gb: float = 16.0 - max_gpu_utilization: float = 0.8 - max_cpu_utilization: float = 0.9 - - # Timeouts - request_timeout: int = 300 - worker_timeout: int = 60 - cleanup_interval: int = 30 - - # Monitoring - enable_metrics: bool = True - metrics_interval: int = 10 - enable_health_checks: bool = True - - def validate(self) -> None: - """Validate configuration parameters.""" - if self.max_memory_gb <= 0: - raise ValueError("max_memory_gb must be positive") - if not 0 < self.max_gpu_utilization <= 1: - raise ValueError("max_gpu_utilization must be between 0 and 1") - if not 0 < self.max_cpu_utilization <= 1: - raise ValueError("max_cpu_utilization must be between 0 and 1") - if self.request_timeout <= 0: - raise ValueError("request_timeout must be positive") -``` - -## 5. Error Handling - -```python -class CoreModuleError(Exception): - """Base exception for core module errors.""" - pass - -class SchedulerError(CoreModuleError): - """Scheduler-related errors.""" - pass - -class QueueFullError(SchedulerError): - """Queue is at capacity.""" - pass - -class CacheError(CoreModuleError): - """Cache-related errors.""" - pass - -class ResourceError(CoreModuleError): - """Resource allocation errors.""" - pass - -class ProcessingError(CoreModuleError): - """Request processing errors.""" - pass - -class ConfigError(CoreModuleError): - """Configuration errors.""" - pass -``` - -## 6. Examples - -### Basic Usage - -```python -from vllm_omni.core import CoreScheduler, CoreModuleConfig -from vllm_omni.request import create_text_request - -# Create configuration -config = CoreModuleConfig( - scheduler=SchedulerConfig(scheduler_type=SchedulerType.FIFO), - cache=CacheConfig(max_memory_gb=8.0) -) - -# Initialize scheduler -scheduler = CoreScheduler(config.scheduler, config.cache) -await scheduler.start() - -# Create and schedule a request -request = create_text_request( - request_id="req_001", - prompt="Hello, world!", - sampling_params=sampling_params -) - -success = await scheduler.schedule_request(request) -if success: - result = await scheduler.process_request(request) - print(f"Result: {result}") - -# Get status -status = scheduler.get_status() -print(f"Queue size: {status['queue_size']}") -print(f"Cache hit rate: {status['cache_hit_rate']:.2%}") - -await scheduler.stop() -``` - -### Advanced Usage - -```python -# Custom scheduler with priority handling and monitoring -config = CoreModuleConfig( - scheduler=SchedulerConfig( - scheduler_type=SchedulerType.PRIORITY, - priority_weights={"high": 3.0, "normal": 1.0, "low": 0.3}, - enable_preemption=True - ), - cache=CacheConfig( - strategy=CacheStrategy.ADAPTIVE, - max_memory_gb=16.0, - enable_compression=True - ), - enable_metrics=True, - enable_health_checks=True -) - -scheduler = CoreScheduler(config.scheduler, config.cache) -await scheduler.start() - -# Monitor scheduler -async def monitor_scheduler(): - while True: - status = scheduler.get_status() - metrics = scheduler.get_metrics() - health = scheduler.get_health_status() - - print(f"Status: {status}") - print(f"Metrics: {metrics}") - print(f"Health: {health}") - - await asyncio.sleep(10) - -# Start monitoring -asyncio.create_task(monitor_scheduler()) -``` - -### Integration Example - -```python -# Integration with other modules -from vllm_omni.engine import EngineManager -from vllm_omni.executor import ExecutorManager - -class OmniSystem: - def __init__(self, config: CoreModuleConfig): - self.scheduler = CoreScheduler(config.scheduler, config.cache) - self.engine_manager = EngineManager(config.engine) - self.executor_manager = ExecutorManager(config.executor) - - async def start(self): - await self.scheduler.start() - await self.engine_manager.start() - await self.executor_manager.start() - - async def process_request(self, request: OmniRequest): - # Schedule request - await self.scheduler.schedule_request(request) - - # Get next request - next_request = await self.scheduler.get_next_request() - if next_request: - # Route to appropriate executor - executor = self.executor_manager.get_executor(next_request) - result = await executor.execute(next_request) - return result -``` diff --git a/requirements.txt b/requirements.txt index 45339bd42..84e553ccf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,6 @@ -# Core dependencies +from vllm_omni.entrypoints.api_server import run_server + +# Start the API server vllm>=0.9.0 torch>=2.0.0 transformers>=4.30.0