Q-Dreamer integration. #19
Conversation
Major Features: - QDREAMER integration with PuLP-based optimization - Multi-pilot architecture with intelligent load balancing - Refactored executor system with BaseExecutor interface - Quantum resource management with fidelity calculation - Real-time queue monitoring and task correlation New Components: - pilot/dreamer.py: Core QDREAMER engine - pilot/executors/: Refactored executor architecture - pilot/services/: Quantum execution and worker services - pilot/util/quantum_resource_generator.py: Resource discovery - examples/dreamer/: Comprehensive QDREAMER examples - tests/test_qdreamer_integration.py: Integration test suite Key Improvements: - Simplified PennyLane executor configuration - Enhanced task submission with quantum-specific features - Automatic circuit compatibility checking - Distributed resource selection at worker level - Comprehensive logging and monitoring Documentation: - Updated README.md with QDREAMER features - Added CHANGELOG.md with migration guide - Created examples/dreamer/README.md - Enhanced tests/README.md with new test categories Breaking Changes: - BaseQuantumExecutor → BaseExecutor - Simplified PennyLane configuration - Enhanced task submission API Closes: QDREAMER integration milestone
Removed outdated and redundant MD files: - MULTI_BACKEND_SUPPORT.md: Information now covered in main README and examples - REFACTORED_ARCHITECTURE.md: Architecture details now in CHANGELOG and code - QDREAMER_INITIALIZATION.md: Initialization info now in examples/dreamer/README Kept essential documentation: - README.md: Main project documentation - CHANGELOG.md: Version history and migration guide - tests/README.md: Test suite documentation - examples/dreamer/README.md: QDREAMER examples guide
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Summary of Changes
Hello @pradeepmantha, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request significantly enhances Pilot-Quantum by integrating Q-Dreamer, a Quantum Resource Allocation and Management Engine. This integration allows for intelligent and optimized selection of quantum computing resources based on task requirements and backend characteristics. The changes introduce a modular executor architecture supporting various quantum frameworks, improve task scheduling across heterogeneous resources, and provide new examples to showcase these advanced capabilities. This update aims to streamline the process of running quantum workloads by automating resource discovery and selection.
Highlights
- Q-Dreamer Integration: Pilot-Quantum now integrates QDREAMER (Quantum Resource Allocation and Management Engine) to intelligently select the optimal quantum backend for submitted tasks, considering factors like fidelity and queue status.
- Multi-Executor Support and Modular Architecture: The framework introduces support for multiple quantum executors including Qiskit, IBMQ, PennyLane, and AWS Braket, enabling flexible execution across diverse quantum platforms. This is supported by a new modular executor architecture.
- Comprehensive Examples: New examples have been added to demonstrate QDREAMER's capabilities, such as multi-pilot resource management, custom backend configurations, PennyLane integration, and intelligent load balancing.
- Improved Logging: The logging system has been enhanced to allow dynamic adjustment of log levels via environment variables, improving debugging and monitoring capabilities.
- Refactored Quantum Task Submission: Quantum task submission has been refactored, with a dedicated 'submit_quantum_task' method that leverages QDREAMER for resource selection and executes tasks remotely on worker nodes for scalability.
- Dynamic Quantum Resource Management: A new utility, 'QuantumResourceGenerator', has been introduced to dynamically discover and manage quantum resources, simplifying the process of integrating new quantum backends.
- Updated Dependencies: The project's dependencies have been updated to include necessary libraries for the new quantum frameworks and optimization tools, ensuring full functionality of the Q-Dreamer integration.
- New Comprehensive Test Suite: A comprehensive test suite has been added, covering basic functionality, executor behavior, Q-Dreamer integration, and performance, complete with a test runner for easy execution and reporting.
Using Gemini Code Assist
The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.
Invoking Gemini
You can request assistance from Gemini at any point in your pull request via creating an issue comment (i.e. comment on the pull request page) using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands.
| Feature | Command | Description |
|---|---|---|
| Code Review | /gemini review |
Performs a code review for the current pull request in its current state. |
| Pull Request Summary | /gemini summary |
Provides a summary of the current pull request in its current state. |
| Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in issue comments and review comments. |
| Help | /gemini help |
Displays a list of available commands. |
Customization
To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.
Limitations & Feedback
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.
You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.
Footnotes
-
Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩
There was a problem hiding this comment.
Code Review
This pull request introduces a major new feature, Q-Dreamer, for intelligent quantum resource management. The changes are extensive, including a new executor-based architecture for handling different quantum backends, a resource selection engine, and updates to the core pilot compute service. While this is a significant and well-structured addition, there are several critical issues that need to be addressed, primarily related to code duplication, incorrect method signatures in the new executor classes, and missing implementations of abstract methods. I have also pointed out some high-severity design concerns and medium-severity bugs and code style issues. Addressing these will greatly improve the robustness and maintainability of this new feature.
| self._device = qml.device(self.device_name, **device_kwargs) | ||
| self.logger.info(f"Created Pennylane device: {self.device_name} with {self.wires} wires") | ||
|
|
||
| return self._device |
There was a problem hiding this comment.
The class PennylaneExecutor does not implement the abstract method get_available_resources from the BaseExecutor base class. This will cause a TypeError at runtime. Please provide an implementation for this method. A simple implementation could return the result of get_backend_info().
| def is_simulator(self) -> bool: | ||
| """ | ||
| Check if this executor uses simulators. | ||
|
|
||
| Returns: | ||
| True if using simulators, False for real hardware | ||
| """ | ||
| # Check if backend name contains 'simulator' or if no token (fake backends) | ||
| return ("simulator" in self.backend_name.lower() or | ||
| "fake" in self.backend_name.lower() or | ||
| self.token is None) |
There was a problem hiding this comment.
The is_simulator method has a bug. It references self.backend_name and self.token, which are not attributes of BraketExecutor. This will raise an AttributeError. The logic seems to have been copied from another executor. You should implement the correct logic for Braket, which might involve checking if self.device_arn contains substrings like /simulator/.
def is_simulator(self) -> bool:
"""
Check if this executor uses simulators.
Returns:
True if using simulators, False for real hardware
"""
if not self.device_arn:
return True # Default to simulator if no ARN
return "simulator" in self.device_arn
pilot/dreamer.py
Outdated
| import csv | ||
| import threading | ||
| import time | ||
| import uuid | ||
| from copy import copy | ||
| from datetime import datetime | ||
| from threading import Lock | ||
| from typing import Dict, List, Optional, Any | ||
|
|
||
| try: | ||
| import pulp | ||
| PULP_AVAILABLE = True | ||
| except ImportError: | ||
| PULP_AVAILABLE = False | ||
| print("Warning: PuLP not available. Using fallback weighted approach.") | ||
|
|
||
| class Q_DREAMER: | ||
| def __init__(self, qdreamer_config_or_resources, quantum_resources_or_config=None, simulation=True, cache_ttl_seconds=30): | ||
| """ | ||
| QDREAMER framework provides intelligent resource selection for quantum tasks. | ||
|
|
||
| Supports both parameter orders for backward compatibility: | ||
| - Old: Q_DREAMER(qdreamer_config, quantum_resources, ...) | ||
| - New: Q_DREAMER(quantum_resources, qdreamer_config, ...) | ||
|
|
||
| :param qdreamer_config_or_resources: Either qdreamer_config dict or quantum_resources dict | ||
| :param quantum_resources_or_config: Either quantum_resources dict or qdreamer_config dict | ||
| :param simulation: Whether to run in simulation mode | ||
| :param cache_ttl_seconds: How long to cache queue information (default: 30 seconds) | ||
| """ | ||
| # Handle parameter order detection - now supports optimization_mode | ||
| if isinstance(qdreamer_config_or_resources, dict) and ('load_balancing' in qdreamer_config_or_resources or 'optimization_mode' in qdreamer_config_or_resources): | ||
| # First parameter is qdreamer_config | ||
| self.qdreamer_config = qdreamer_config_or_resources | ||
| self.quantum_resources = quantum_resources_or_config or {} | ||
| else: | ||
| # First parameter is quantum_resources | ||
| self.quantum_resources = qdreamer_config_or_resources | ||
| self.qdreamer_config = quantum_resources_or_config or {} | ||
|
|
||
| self.simulation = simulation | ||
| self.cache_ttl_seconds = cache_ttl_seconds | ||
|
|
||
| # Initialize intelligent optimizer | ||
| optimization_mode = self.qdreamer_config.get('optimization_mode', 'multi_objective') | ||
| self.optimizer = OptimizedResourceSelector(optimization_mode) | ||
|
|
||
|
|
||
|
|
||
| # Cache for queue information | ||
| self._queue_cache = {} | ||
| self._cache_timestamps = {} | ||
| self._cache_lock = Lock() | ||
|
|
||
| # Background monitoring | ||
| self._monitoring_active = False | ||
| self._monitor_thread = None | ||
| self._monitor_interval = 60 # seconds | ||
|
|
||
| self.logger = logging.getLogger(__name__) | ||
|
|
||
| # Initialize queue dynamics | ||
| self.queue_dynamics = self.qdreamer_config.get("queue_dynamics", {}) | ||
|
|
||
| # Start background monitoring | ||
| self._start_background_monitoring() | ||
|
|
||
| @property | ||
| def config(self): | ||
| """Get the QDREAMER configuration.""" | ||
| return { | ||
| 'qdreamer_config': self.qdreamer_config, | ||
| 'optimization_mode': self.optimizer.optimization_mode, | ||
| 'simulation': self.simulation, | ||
| 'cache_ttl_seconds': self.cache_ttl_seconds | ||
| } | ||
|
|
||
| def get_best_resource(self, quantum_task: QuantumTask, task_id: str = 'unknown') -> Optional[QuantumResource]: | ||
| """ | ||
| Get the best quantum resource for a given task using intelligent optimization. | ||
|
|
||
| Args: | ||
| quantum_task: The quantum task to find resources for | ||
| task_id: Task ID for correlation logging | ||
|
|
||
| Returns: | ||
| Best quantum resource or None if no suitable resource found | ||
| """ | ||
| self.logger.info(f"[TASK:{task_id}] 🔍 QDREAMER selecting best resource for task: {quantum_task.num_qubits} qubits, gates: {quantum_task.gate_set}") | ||
|
|
||
| # Get current queue dynamics | ||
| current_queue_dynamics = self._get_current_queue_dynamics() | ||
|
|
||
| # Use intelligent optimization | ||
| best_resource = self.optimizer.optimize_resource_selection( | ||
| quantum_task, self.quantum_resources, current_queue_dynamics, task_id | ||
| ) | ||
|
|
||
| if best_resource: | ||
| self.logger.info(f"[TASK:{task_id}] ✅ QDREAMER selected: {best_resource.name}") | ||
| return best_resource | ||
| else: | ||
| self.logger.warning(f"[TASK:{task_id}] ⚠️ No suitable quantum resource found!") | ||
| return None | ||
|
|
||
| def _get_current_queue_dynamics(self) -> Dict[str, float]: | ||
| """Get current queue utilization for all resources from executors.""" | ||
| current_time = time.time() | ||
|
|
||
| with self._cache_lock: | ||
| # Check if cache is still valid | ||
| if (current_time - self._cache_timestamps.get('queue_dynamics', 0)) < self.cache_ttl_seconds: | ||
| return self._queue_cache.get('queue_dynamics', {}) | ||
|
|
||
| # Update queue dynamics from executors | ||
| queue_dynamics = {} | ||
|
|
||
| # Group resources by executor type | ||
| executor_resources = {} | ||
| for resource_name, resource in self.quantum_resources.items(): | ||
| # Extract executor type from resource name (e.g., "pilot1_qiskit_aer" -> "qiskit") | ||
| if hasattr(resource, 'pilot_name'): | ||
| # Use pilot name to identify executor | ||
| executor_key = resource.pilot_name | ||
| else: | ||
| # Fallback: extract from resource name | ||
| parts = resource_name.split('_') | ||
| executor_key = parts[1] if len(parts) > 1 else 'unknown' | ||
|
|
||
| if executor_key not in executor_resources: | ||
| executor_resources[executor_key] = [] | ||
| executor_resources[executor_key].append((resource_name, resource)) | ||
|
|
||
| # Get queue information from each executor | ||
| for executor_key, resources in executor_resources.items(): | ||
| try: | ||
| # Get queue lengths from the executor | ||
| # This would need to be implemented in the quantum execution service | ||
| # For now, use the existing queue_dynamics as fallback | ||
| for resource_name, resource in resources: | ||
| original_name = getattr(resource, 'original_name', resource_name) | ||
| queue_util = self.queue_dynamics.get(original_name, 0.0) | ||
| queue_dynamics[resource_name] = queue_util | ||
|
|
||
| except Exception as e: | ||
| self.logger.warning(f"Failed to get queue dynamics for executor {executor_key}: {e}") | ||
| # Fallback to existing queue_dynamics | ||
| for resource_name, resource in resources: | ||
| original_name = getattr(resource, 'original_name', resource_name) | ||
| queue_util = self.queue_dynamics.get(original_name, 0.0) | ||
| queue_dynamics[resource_name] = queue_util | ||
|
|
||
| # Update cache | ||
| self._queue_cache['queue_dynamics'] = queue_dynamics | ||
| self._cache_timestamps['queue_dynamics'] = current_time | ||
|
|
||
| return queue_dynamics | ||
|
|
||
| def _start_background_monitoring(self): | ||
| """Start background monitoring of queue dynamics.""" | ||
| if self._monitoring_active: | ||
| return | ||
|
|
||
| self._monitoring_active = True | ||
| self._monitor_thread = threading.Thread(target=self._monitor_queues, daemon=True) | ||
| self._monitor_thread.start() | ||
| self.logger.info("Background queue monitoring started") | ||
|
|
||
| def _monitor_queues(self): | ||
| """Background thread to monitor queue dynamics.""" | ||
| while self._monitoring_active: | ||
| try: | ||
| # Update queue dynamics (in a real implementation, this would query actual queue status) | ||
| if self.simulation: | ||
| # Simulate queue changes | ||
| for resource_name in self.quantum_resources.keys(): | ||
| original_name = getattr(self.quantum_resources[resource_name], 'original_name', resource_name) | ||
| if original_name in self.queue_dynamics: | ||
| # Simulate some queue variation | ||
| current = self.queue_dynamics[original_name] | ||
| variation = (hash(f"{original_name}_{int(time.time() / 60)}") % 100) / 1000.0 | ||
| self.queue_dynamics[original_name] = max(0.0, min(1.0, current + variation)) | ||
|
|
||
| time.sleep(self._monitor_interval) | ||
|
|
||
| except Exception as e: | ||
| self.logger.error(f"Error in queue monitoring: {e}") | ||
| time.sleep(self._monitor_interval) | ||
|
|
||
| def stop_monitoring(self): | ||
| """Stop background monitoring.""" | ||
| self._monitoring_active = False | ||
| if self._monitor_thread: | ||
| self._monitor_thread.join(timeout=5) | ||
| self.logger.info("Stopped background queue monitoring") | ||
|
|
||
|
|
||
|
No newline at end of file |
There was a problem hiding this comment.
The entire content of this file from line 1 to 207 is duplicated starting at line 210. This is a significant maintenance issue and should be removed to avoid inconsistencies and bugs in the future.
| class QuantumResource: | ||
| def __init__(self, name, qubit_count, gateset, error_rate=None, noise_level=None, pending_jobs=0, quantum_config=None): | ||
| """ | ||
| Initializes a QuantumResource instance. | ||
|
|
||
| :param name: Backend name | ||
| :param qubit_count: Number of qubits available | ||
| :param gateset: List of supported gates | ||
| :param error_rate: Estimated error rate (default: None) | ||
| :param noise_level: Noise level of the backend (default: None) | ||
| :param pending_jobs: Number of pending jobs in the queue (default: 0) | ||
| :param quantum_config: Original quantum configuration from pilot description | ||
| """ | ||
| self.name = name | ||
| self.qubit_count = qubit_count | ||
| self.gateset = gateset or [] | ||
| self.error_rate = error_rate if error_rate is not None else float("inf") | ||
| self.noise_level = noise_level if noise_level is not None else float("inf") | ||
| self.available_qubits = qubit_count | ||
| self.quantum_config = quantum_config or {} | ||
|
|
||
| def to_dict(self): | ||
| """Returns a dictionary representation of the QuantumResource.""" | ||
| return { | ||
| "name": self.name, | ||
| "qubit_count": self.qubit_count, | ||
| "gateset": self.gateset, | ||
| "error_rate": self.error_rate, | ||
| "noise_level": self.noise_level, | ||
| "quantum_config": self.quantum_config | ||
| } | ||
|
|
||
| def __repr__(self): | ||
| """Returns a detailed string representation (useful for debugging).""" | ||
| return f"QuantumResource(name={self.name}, qubit_count={self.qubit_count}, gateset={self.gateset}, " \ | ||
| f"error_rate={self.error_rate}, noise_level={self.noise_level}, quantum_config={self.quantum_config})" | ||
|
|
||
| def __str__(self): | ||
| """Returns a human-readable string representation.""" | ||
| return f"{self.name}: {self.qubit_count} qubits, Gateset: {self.gateset}, " \ | ||
| f"Error Rate: {self.error_rate:.4f}, Noise Level: {self.noise_level:.4f}, " \ | ||
| f"Config: {self.quantum_config}" | ||
|
|
||
| @property | ||
| def fidelity(self) -> float: | ||
| """Calculate fidelity as 1 - error_rate.""" | ||
| return 1.0 - self.error_rate if self.error_rate is not None and self.error_rate != float("inf") else 1.0 | ||
|
|
There was a problem hiding this comment.
The QuantumResource class is also defined in pilot/dreamer.py. Having duplicate class definitions is a major source of bugs and maintenance overhead. Please consolidate this class into a single file, preferably pilot/dreamer.py where other data model classes like QuantumTask reside. Then, import it here and in other places where it's needed.
pilot/executors/qiskit_executor.py
Outdated
| self._backend = None | ||
| self._is_custom_backend = False | ||
|
|
||
| def execute_circuit(self, circuit, *args, **kwargs): |
There was a problem hiding this comment.
The signature of execute_circuit does not match the one defined in the abstract base class BaseExecutor. It's missing the resource_name parameter. The signature must match the ABC to ensure proper polymorphism.
| def execute_circuit(self, circuit, *args, **kwargs): | |
| def execute_circuit(self, circuit, resource_name: str, *args, **kwargs): |
| from .ibmq_executor import IBMQExecutor | ||
|
|
||
| __all__ = [ | ||
| 'BaseQuantumExecutor', |
There was a problem hiding this comment.
The __all__ list contains BaseQuantumExecutor, but the class defined in base_executor.py is named BaseExecutor. This will raise an ImportError for users who do from pilot.executors import *.
| 'BaseQuantumExecutor', | |
| 'BaseExecutor', |
| def _get_executor_type_from_resource(self, resource) -> str: | ||
| """ | ||
| Determine executor type from resource information. | ||
|
|
||
| Args: | ||
| resource: Quantum resource information | ||
|
|
||
| Returns: | ||
| Executor type string | ||
| """ | ||
| resource_name = resource.name.lower() | ||
|
|
||
| if 'qiskit' in resource_name: | ||
| return 'qiskit' | ||
| elif 'pennylane' in resource_name: | ||
| return 'pennylane' | ||
| elif 'braket' in resource_name: | ||
| return 'braket' | ||
| elif 'ibmq' in resource_name: | ||
| return 'ibmq' | ||
| else: | ||
| # Default to qiskit for unknown resources | ||
| return 'qiskit' |
There was a problem hiding this comment.
The _get_executor_type_from_resource method relies on string matching on the resource name to determine the executor type. This is brittle and can lead to incorrect executor selection if resource naming conventions are not strictly followed. A more robust approach would be to store the executor type as an attribute on the QuantumResource object when it's created in QuantumResourceGenerator. This would make the lookup explicit and reliable.
| import math | ||
| import os | ||
| import time | ||
| import random | ||
| from typing import Dict, List | ||
|
|
||
| import pennylane as qml | ||
| import ray | ||
| from pilot.dreamer import Coupling, QuantumTask, TaskType | ||
| from pilot.pilot_compute_service import ExecutionEngine, PilotComputeService | ||
| from pilot.util.quantum_resource_generator import QuantumResourceGenerator, QuantumResource | ||
| from time import sleep |
There was a problem hiding this comment.
There are several unused imports in this file: math, random, pennylane, Coupling, TaskType, QuantumResourceGenerator, QuantumResource, and sleep. Please remove them to improve code clarity and reduce clutter.
| self.logger.info(f"Pilot submitting with resource: {pilot_compute_description['resource']}, \ | ||
| cores_per_node: {pilot_compute_description['cores_per_node']}, \ | ||
| number_of_nodes: {pilot_compute_description['number_of_nodes']}") |
There was a problem hiding this comment.
This log message is very long and has inconsistent indentation, which makes it hard to read. Consider breaking it into multiple lines or using a more structured format for better readability.
| self.logger.info(f"Pilot submitting with resource: {pilot_compute_description['resource']}, \ | |
| cores_per_node: {pilot_compute_description['cores_per_node']}, \ | |
| number_of_nodes: {pilot_compute_description['number_of_nodes']}") | |
| self.logger.info(f"Pilot submitting with resource: {pilot_compute_description['resource']}, " | |
| f"cores_per_node: {pilot_compute_description['cores_per_node']}, " | |
| f"number_of_nodes: {pilot_compute_description['number_of_nodes']}") |
pilot/dreamer.py
Outdated
| def __init__(self, type: TaskType, resource_config: dict): | ||
| self.type = TaskType | ||
| self.resource_config = None | ||
| self.resource_config = resource_config |
There was a problem hiding this comment.
There are a couple of issues in the Task class constructor:
self.type = TaskTypeassigns theTaskTypeEnum class itself to the instance variable, not the value passed in thetypeparameter. It should beself.type = type.self.resource_config = Noneis immediately overwritten by the next line, making it redundant.
| def __init__(self, type: TaskType, resource_config: dict): | |
| self.type = TaskType | |
| self.resource_config = None | |
| self.resource_config = resource_config | |
| def __init__(self, type: TaskType, resource_config: dict): | |
| self.type = type | |
| self.resource_config = resource_config |
Fixed several issues in QDREAMER integration tests: - Fixed queue_dynamics parameter handling in dreamer.py to handle None values - Added missing get_available_resources method to PennylaneExecutor - Fixed test method signatures to match actual OptimizedResourceSelector API - Corrected test logic for circuit compatibility and error handling - Fixed task correlation logging test to expect correct UUID length All QDREAMER integration tests now pass successfully.
Fixed executor constructor calls in test_executor_compatibility.py: - Added required 'name' parameter to QiskitExecutor constructor calls - Added required 'name' parameter to PennylaneExecutor constructor calls - All executor compatibility tests now pass successfully The executors require both a name and config parameter, but tests were only passing config.
Fixed several issues in test_all_executors.py: - Fixed initialize_dreamer() return value expectations (method returns None, not QDREAMER instance) - Updated tests to check pcs.qdreamer and pcs.dreamer_enabled attributes instead - Fixed error handling test to reflect actual graceful fallback behavior for invalid executors - Fixed circuit validation test to expect None result for invalid circuits - All executor tests now pass successfully The tests now correctly reflect the actual behavior of the QDREAMER system.
No description provided.