qpanda3_runtime.task.qtask_manager#

Classes#

QTaskManager

Quantum Computing Task Manager for QPanda3 Runtime Service.

Module Contents#

class qpanda3_runtime.task.qtask_manager.QTaskManager(runtime_service, task_data, used_remote_backend=True, used_fake_backend=False, used_program_set=False)#

Quantum Computing Task Manager for QPanda3 Runtime Service.

This class provides local management for quantum computing tasks, including task submission, result retrieval, and state management. It supports synchronous, asynchronous, and background execution modes.

Parameters:

runtime_service (RuntimeService) – Runtime service instance for task execution and communication.
task_data (Dict) – Task metadata containing task ID, session information, and other configuration parameters.
used_remote_backend (bool, default True) – Whether the task uses a remote computational backend.
used_fake_backend (bool, default False) – Whether the task uses a FakeBackend simulator.
used_program_set (bool, default False) – Whether the task uses ProgramSet for parameterized circuits.

Variables:

_runtime_service (RuntimeService) – Reference to the runtime service.
_id (Union[str, List[str]]) – Task ID or list of sub-task IDs.
_session_id (str) – Session identifier for variational algorithms.
_data (Dict) – Original task metadata.
_used_remote_backend (bool) – Remote backend usage flag.
_used_program_set (bool) – ProgramSet usage flag.
_client_task_id (Optional[str]) – Client-generated task identifier.
_sub_task_id__result_id (Dict[str, int]) – Mapping from sub-task ID to result index.
_result (List[Any]) – Task execution results.
_is_transpile_successed (bool) – Circuit transpilation success flag.
_processing_sub_task_ids (List[str]) – List of sub-task IDs still being processed.
_used_fake_backend (bool) – FakeBackend usage flag.
_data_lock (threading.Lock) – Thread lock for concurrent access.
_result_fetching_thread (Optional[threading.Thread]) – Background result fetching thread.

Examples

>>> from qpanda3_runtime import RuntimeService
>>> from qpanda3_runtime.task import QTaskManager
>>> # Create task manager for submitted task
>>> runtime_service = RuntimeService()
>>> task_data = {'task_id': 'task_123', 'channel': 'qcloud'}
>>> task_manager = QTaskManager(runtime_service, task_data)
>>> # Retrieve results
>>> results, finished, progress = task_manager.try_get_result()

Notes

Thread-safe design for concurrent access
Supports checkpoint saving and recovery
Provides multiple result retrieval strategies
Integrates with tracking system for monitoring

try_get_result(additional_info: List[str] | None = None, timeout: float = 600)#

Query task execution results with incremental strategy.

This method queries the current status of computational tasks and returns available results. It only queries unfinished subtasks but returns all currently obtained result data. Subsequent calls use incremental strategy, storing fetched results internally.

Parameters:

additional_info (Optional[List[str]], default None) – Additional query fields for task execution information. Refer to backend documentation for supported fields.
timeout (float, default 600) – Timeout in seconds for each query request.

Returns:

Three-element tuple containing: - List of task results (unavailable results marked as “Status:Processing”) - Boolean indicating if all results are complete - String showing completion percentage (e.g., “finished 75%”)

Return type:

Tuple[List[Any], bool, str]

Raises:

RuntimeError – If task query fails or returns error response.

Notes

Uses incremental fetching: only queries unfinished subtasks
Results are cached internally for subsequent calls
For batch tasks, queries all sub-task results
Integrates with tracking system for monitoring

Examples

>>> results, finished, progress = task_manager.try_get_result()
>>> while not finished:
...     print(f"Progress: {progress}")
...     time.sleep(5)
...     results, finished, progress = task_manager.try_get_result()
>>> print(f"Final results: {results}")

async get_result_async(additional_info: List[str] | None = None, timeout=1800)#

A coroutine function wrapped for the asyncio asynchronous library.

Parameters:

additional_info (Optional[List[str]]) – Query fields related to compute task execution information. For specific details about these fields, please refer to the support documentation of the respective compute backend.
timeout (int) – The maximum duration for querying the task process. Once this time length (in seconds) is exceeded, the system will start canceling the query requests for the task.

Returns:

The execution results of the computational task

Return type:

list

get_result_sync(additional_info: List[str] | None = None, timeout=1000 * 1800, using_http=True)#

Synchronously query task execution results with blocking behavior.

This method blocks the client program until all result data is returned or timeout is reached. It continuously polls for results until completion.

Parameters:

additional_info (Optional[List[str]], default None) – Additional query fields for task execution information.
timeout (int, default 1,800,000) – Maximum query duration in seconds before raising timeout error.
using_http (bool, default True) – Protocol to use for query: - True: HTTP protocol - False: TCP protocol

Returns:

Complete task execution results.

Return type:

List[Any]

Raises:

RuntimeError – If query timeout is exceeded or task execution fails.

Notes

Blocks until all results are available or timeout
Uses polling strategy with adaptive wait times
Supports both HTTP and TCP protocols
Timeout value should be sufficiently large for quantum tasks

Examples

>>> # Block until results are available (up to 30 minutes)
>>> results = task_manager.get_result_sync(timeout=1800)
>>> print(f"Synchronous results: {results}")

get_result_background(call_back: Callable[[Any], None], additional_info: List[str] | None = None, timeout=1800)#

Use a sub-thread to pull the execution results of the computational task in the background.

Termination of the sub‑thread: The sub‑thread ends after returning all result data, or is terminated when the main thread ends.
Monitoring the status of the sub‑thread: QTaskManager.get_result_background returns a monitoring function. The return value of the monitoring function is a Boolean: True indicates that the background sub‑thread has ended; otherwise, it returns False.

Parameters:

call_back (Callable[[Any], None]) – A callback function can be passed to QTaskManager.get_result_background.
additional_info (Optional[List[str]]) – Query fields related to compute task execution information. For specific details about these fields, please refer to the support documentation of the respective compute backend.
timeout (int) – The maximum duration for querying the task process. Once this time length (in seconds) is exceeded, the system will start canceling the query requests for the task.

Returns:

The execution results of the computational task

Return type:

list

id()#

Get the ID of the computational task (represented by the set of subtask IDs).

Returns:

list – The ID of the computational task, a set of subtask IDs.
str – The ID of the computational task,a subtask’s id.

progs()#

Retrieve all quantum circuits from the metadata of the quantum computing task.

Returns:: All quantum circuits from the metadata of the quantum computing task
Return type:: list

used_remote_backend()#

Determine whether the quantum computing task uses a remote computing backend.

Returns:

True: Yes

False: No

Return type:

bool

used_fake_backend()#

Determine whether the quantum computing task uses a FakeBackend.

Returns:

True: Yes

False: No

Return type:

bool

run_on_real_device(shots=1000)#

Execute pre-verified task on real quantum hardware.

Submit a quantum computing task that was previously verified on a FakeBackend to real quantum hardware for execution. This method is only available when the current task uses a FakeBackend and transpilation was successful.

Parameters:: shots (int, default 1000) – Number of measurement shots for the experiment.
Returns:: New task manager for the real hardware execution.
Return type:: QTaskManager
Raises:: RuntimeError – If task doesn’t use FakeBackend or transpilation failed.

Notes

Requires successful transpilation on FakeBackend
Creates new task on real quantum hardware
Preserves all task parameters except backend
Useful for simulation-to-hardware workflow

Examples

>>> # First run on FakeBackend for verification
>>> fake_task = runtime_service.sample(circuit, fake_backend, shots=100)
>>> # If successful, run on real hardware
>>> if fake_task.used_fake_backend():
...     real_task = fake_task.run_on_real_device(shots=1000)
...     results = real_task.get_result_sync()

channel()#

For getting the backend access channel where the computation task resides

Returns:: The backend access channel where the computation task resides
Return type:: str

get_task_state() → Dict#

Get serializable state of quantum computing task.

Returns a dictionary containing serializable members of QTaskManager instance, used for checkpoint saving and recovery.

Returns:: Dictionary containing the following serializable members: - qtask_manager_init_time: QTaskManager initialization time (timestamp) - session_id: Session ID - data: Task metadata - used_remote_backend: Whether remote backend is used - used_program_set: Whether ProgramSet is used - sub_task_id__result_id: Mapping from sub-task ID to result ID - result: Task execution result - is_transpile_successed: Whether transpilation succeeded - processing_sub_task_ids: List of sub-task IDs being processed - used_fake_backend: Whether FakeBackend is used
Return type:: dict

check_point(filepath: str | None = None, user_data: Dict | None = None) → str#

Save checkpoint data for quantum computing task.

Saves the current task state to a checkpoint file using the CheckPoint.save method, enabling task state recovery later.

Parameters:

filepath (str, optional) – Checkpoint file save path. If None, system generates a default filename and writes to current working directory.
user_data (Dict, optional) – User-defined data dictionary with JSON-serializable values.

Returns:

Absolute path to the written checkpoint file.

Return type:

str

Raises:

TypeError – If task_state is not a dict and lacks get_task_state() method.
Exception – Original exception from get_task_state() is propagated.

Notes

System does not manage checkpoint identifiers or guarantee filename uniqueness
Default filename format: checkpoint_YYYYMMDD_HHMMSS.json
Default filename does not guarantee global uniqueness
When filepath exists, it is overwritten by default
Not thread-safe/process-safe for concurrent writes
File deletion is user’s responsibility

Warning

Concurrent writes to same file may cause data corruption
User must handle filename uniqueness in concurrent scenarios
user_data must be JSON-serializable

Examples

>>> # Save checkpoint with default filename
>>> checkpoint_path = task_manager.check_point()
>>> print(f"Checkpoint saved to: {checkpoint_path}")
>>>
>>> # Save checkpoint with custom filename and user data
>>> user_data = {'experiment_id': 'exp_123', 'note': 'VQE iteration 50'}
>>> checkpoint_path = task_manager.check_point(
...     filepath='vqe_iteration_50.json',
...     user_data=user_data
... )

static from_task_state(runtime_service, task_state, recover_completely=True)#

Create QTaskManager object from task state and RuntimeService instance.

Parameters:

runtime_service (RuntimeService) – RuntimeService instance used for executing quantum computing tasks.
task_state (dict) – Task state dictionary returned by get_task_state method.
recover_completely (bool, optional) – Whether to completely recover task state. Default is True. - True: Completely recover original task state, task_state must contain all information - False: Create a new task using state data, allow default values

Returns:

QTaskManager instance created from task state.

Return type:

QTaskManager

Raises:

ValueError – If recover_completely is True and task_state is missing required fields.