Spaces:
Sleeping
Sleeping
| """ | |
| Cloud Resource Allocation RL Environment. | |
| This module implements the core environment class for simulating cloud infrastructure | |
| resource allocation. The environment maintains system state (CPU, memory, request rate, | |
| allocated resources) and processes actions to dynamically adjust resource allocation. | |
| """ | |
| import numpy as np | |
| import torch | |
| from typing import Tuple, Dict, List, Optional | |
| from env.config import EnvConfig | |
| from env.reward import RewardCalculator | |
| class CloudResourceEnv: | |
| """ | |
| Simulates a cloud infrastructure environment for RL-based resource allocation. | |
| The environment models a cloud system where an agent must dynamically allocate | |
| server instances based on observed system metrics. The state includes CPU utilization, | |
| memory utilization, incoming request rate, and currently allocated resources. | |
| State space: [cpu_util, memory_util, request_rate, allocated_resources] | |
| Action space: {0: decrease, 1: maintain, 2: increase} | |
| The environment incorporates stochastic dynamics to simulate real-world variability | |
| in request patterns and system behavior. | |
| """ | |
| # Action constants | |
| ACTION_DECREASE = 0 | |
| ACTION_MAINTAIN = 1 | |
| ACTION_INCREASE = 2 | |
| def __init__( | |
| self, | |
| config: Optional[EnvConfig] = None, | |
| reward_calculator: Optional[RewardCalculator] = None, | |
| task_name: str = "medium" | |
| ): | |
| """ | |
| Initialize the cloud resource allocation environment. | |
| Args: | |
| config: Environment configuration parameters. If None, uses default EnvConfig. | |
| reward_calculator: Reward calculator instance. If None, creates default RewardCalculator. | |
| task_name: Task difficulty level ("easy", "medium", or "hard"). Defaults to "medium". | |
| Raises: | |
| ValueError: If task_name is not in TASKS dictionary. | |
| """ | |
| # Import TASKS here to avoid circular imports | |
| from env.config import TASKS | |
| # Validate task_name | |
| if task_name not in TASKS: | |
| raise ValueError( | |
| f"Invalid task_name '{task_name}'. Must be one of: {list(TASKS.keys())}" | |
| ) | |
| # Use default config if not provided | |
| self.config = config if config is not None else EnvConfig() | |
| # Override config with task-specific parameters | |
| task_config = TASKS[task_name] | |
| self.config.base_request_rate = task_config["base_request_rate"] | |
| self.config.request_rate_std = task_config["request_rate_std"] | |
| self.task_name = task_name | |
| # Initialize reward calculator | |
| self.reward_calculator = reward_calculator if reward_calculator is not None else RewardCalculator() | |
| # Initialize state variables | |
| self.cpu_util: float = 0.0 | |
| self.memory_util: float = 0.0 | |
| self.request_rate: int = 0 | |
| self.allocated_resources: int = self.config.initial_resources | |
| self.latency: float = 0.0 # Latency in ms | |
| # Initialize episode tracking | |
| self.current_step: int = 0 | |
| self.episode_states: List[np.ndarray] = [] | |
| self.episode_actions: List[int] = [] | |
| self.episode_rewards: List[float] = [] | |
| self.cumulative_reward: float = 0.0 | |
| # Episode termination flag | |
| self.done: bool = False | |
| # Random number generator for reproducibility | |
| self.rng = np.random.RandomState() | |
| def reset(self) -> np.ndarray: | |
| """ | |
| Reset environment to initial state. | |
| This method reinitializes the environment for a new episode. It resets all | |
| state variables to their initial values, clears episode history, and introduces | |
| stochastic variation in the initial request rate to ensure diverse starting | |
| conditions across episodes. | |
| Returns: | |
| Initial observation as numpy array [cpu_util, memory_util, request_rate, allocated_resources] | |
| """ | |
| # Reset episode tracking | |
| self.current_step = 0 | |
| self.episode_states = [] | |
| self.episode_actions = [] | |
| self.episode_rewards = [] | |
| self.cumulative_reward = 0.0 | |
| self.done = False | |
| # Reset allocated resources to initial value | |
| self.allocated_resources = self.config.initial_resources | |
| # Initialize request rate with stochastic variation | |
| # This ensures different starting conditions across episodes | |
| self.request_rate = int( | |
| self.config.base_request_rate + | |
| self.rng.normal(0, self.config.request_rate_std) | |
| ) | |
| # Ensure request rate is non-negative | |
| self.request_rate = max(0, self.request_rate) | |
| # Calculate initial CPU and memory utilization | |
| # Utilization depends on request rate and allocated resources | |
| self.cpu_util = (self.request_rate * self.config.cpu_per_request) / ( | |
| self.allocated_resources * self.config.resource_capacity | |
| ) * 100.0 | |
| self.memory_util = (self.request_rate * self.config.memory_per_request) / ( | |
| self.allocated_resources * self.config.resource_capacity | |
| ) * 100.0 | |
| # Clamp utilization values to valid range [0, 100] | |
| self.cpu_util = max(0.0, min(100.0, self.cpu_util)) | |
| self.memory_util = max(0.0, min(100.0, self.memory_util)) | |
| # Calculate latency: higher request rate and fewer resources increase latency | |
| self.latency = self.request_rate / max(self.allocated_resources, 1) | |
| # Return initial observation | |
| return self._get_observation() | |
| def _get_observation(self) -> np.ndarray: | |
| """ | |
| Get current state as observation array. | |
| Returns: | |
| Numpy array [cpu_util, memory_util, request_rate, allocated_resources] | |
| """ | |
| return np.array([ | |
| self.cpu_util, | |
| self.memory_util, | |
| float(self.request_rate), | |
| float(self.allocated_resources) | |
| ], dtype=np.float32) | |
| def get_state_tensor(self) -> torch.Tensor: | |
| """ | |
| Get current state as PyTorch tensor. | |
| This method provides a PyTorch tensor representation of the current state, | |
| enabling seamless integration with PyTorch-based RL algorithms and neural networks. | |
| The tensor maintains the same structure as the NumPy observation array. | |
| Returns: | |
| PyTorch tensor [cpu_util, memory_util, request_rate, allocated_resources] | |
| """ | |
| observation = self._get_observation() | |
| return torch.from_numpy(observation) | |
| def state(self): | |
| """ | |
| Get current state as EnvState instance for OpenEnv compliance. | |
| This method provides typed, validated access to the environment state | |
| using the Pydantic EnvState model. It can be called after reset() or step() | |
| to retrieve the current state in a structured format. | |
| Returns: | |
| EnvState: Pydantic model with cpu, memory, request_rate, and resources fields | |
| """ | |
| from env.config import EnvState | |
| return EnvState( | |
| cpu=self.cpu_util, | |
| memory=self.memory_util, | |
| request_rate=float(self.request_rate), | |
| resources=self.allocated_resources | |
| ) | |
| def _update_state(self, action: int) -> None: | |
| """ | |
| Update system state based on action and stochastic dynamics. | |
| This method applies the selected action to modify allocated resources, | |
| updates the request rate with stochastic fluctuation, and recalculates | |
| CPU and memory utilization based on the new system configuration. | |
| Args: | |
| action: Action to execute (0: decrease, 1: maintain, 2: increase) | |
| """ | |
| # Apply action to allocated resources | |
| if action == self.ACTION_INCREASE: | |
| self.allocated_resources += 1 | |
| elif action == self.ACTION_DECREASE: | |
| # Maintain minimum of 1 resource | |
| self.allocated_resources = max(1, self.allocated_resources - 1) | |
| elif action == self.ACTION_MAINTAIN: | |
| # Keep resources unchanged | |
| pass | |
| else: | |
| raise ValueError(f"Invalid action: {action}. Must be 0, 1, or 2.") | |
| # Update request rate with stochastic fluctuation | |
| # This simulates real-world variability in incoming traffic | |
| fluctuation = self.rng.normal(0, self.config.request_rate_std) | |
| self.request_rate = int(self.config.base_request_rate + fluctuation) | |
| # Ensure request rate is non-negative | |
| self.request_rate = max(0, self.request_rate) | |
| # Calculate CPU utilization based on request rate and allocated resources | |
| # Higher request rate increases utilization | |
| # More allocated resources decrease utilization | |
| self.cpu_util = (self.request_rate * self.config.cpu_per_request) / ( | |
| self.allocated_resources * self.config.resource_capacity | |
| ) * 100.0 | |
| # Calculate memory utilization based on request rate and allocated resources | |
| self.memory_util = (self.request_rate * self.config.memory_per_request) / ( | |
| self.allocated_resources * self.config.resource_capacity | |
| ) * 100.0 | |
| # Clamp all values to valid ranges | |
| self.cpu_util = max(0.0, min(100.0, self.cpu_util)) | |
| self.memory_util = max(0.0, min(100.0, self.memory_util)) | |
| # Calculate latency: higher request rate and fewer resources increase latency | |
| self.latency = self.request_rate / max(self.allocated_resources, 1) | |
| def _check_termination(self) -> bool: | |
| """ | |
| Check if episode should terminate based on state or step count. | |
| Termination conditions: | |
| 1. Step count exceeds max_steps | |
| 2. CPU utilization exceeds termination threshold (default 95%) | |
| 3. Memory utilization exceeds termination threshold (default 95%) | |
| Returns: | |
| True if episode should terminate, False otherwise | |
| """ | |
| # Check if max steps exceeded | |
| if self.current_step >= self.config.max_steps: | |
| return True | |
| # Check if CPU utilization exceeds threshold | |
| if self.cpu_util > self.config.termination_threshold: | |
| return True | |
| # Check if memory utilization exceeds threshold | |
| if self.memory_util > self.config.termination_threshold: | |
| return True | |
| return False | |
| def step(self, action: int) -> Tuple[np.ndarray, float, bool, Dict]: | |
| """ | |
| Execute one time step with the given action. | |
| This method processes the action, updates the environment state, calculates | |
| the reward, checks for termination, and stores the transition in episode | |
| history. It returns the standard RL tuple (observation, reward, done, info). | |
| Args: | |
| action: Action to execute (0: decrease, 1: maintain, 2: increase) | |
| Returns: | |
| observation: Current state after action as numpy array | |
| reward: Reward signal from reward calculator | |
| done: Whether episode has terminated | |
| info: Additional diagnostic information dictionary | |
| Raises: | |
| ValueError: If action is not in valid range {0, 1, 2} | |
| RuntimeError: If step is called after episode termination | |
| """ | |
| # Check if episode already terminated | |
| if self.done: | |
| raise RuntimeError( | |
| "Cannot call step() after episode termination. Call reset() to start a new episode." | |
| ) | |
| # Validate action is in valid range | |
| if action not in {self.ACTION_DECREASE, self.ACTION_MAINTAIN, self.ACTION_INCREASE}: | |
| raise ValueError( | |
| f"Invalid action: {action}. Must be 0 (decrease), 1 (maintain), or 2 (increase)." | |
| ) | |
| # Update state based on action | |
| self._update_state(action) | |
| # Increment step counter | |
| self.current_step += 1 | |
| # Calculate reward | |
| reward = self.reward_calculator.calculate_reward( | |
| self.cpu_util, | |
| self.memory_util, | |
| self.allocated_resources | |
| ) | |
| # Check termination status | |
| self.done = self._check_termination() | |
| # Get current observation | |
| observation = self._get_observation() | |
| # Store transition in episode history | |
| self.episode_states.append(observation.copy()) | |
| self.episode_actions.append(action) | |
| self.episode_rewards.append(reward) | |
| self.cumulative_reward += reward | |
| # Prepare info dictionary | |
| info = { | |
| 'step': self.current_step, | |
| 'cumulative_reward': self.cumulative_reward, | |
| 'cpu_util': self.cpu_util, | |
| 'memory_util': self.memory_util, | |
| 'request_rate': self.request_rate, | |
| 'allocated_resources': self.allocated_resources, | |
| 'latency': self.latency | |
| } | |
| return observation, reward, self.done, info | |