Controller Plugins

Controller plugins extend the built-in functionality of the Quant-Net control plane. Currently, there are four types of plugins, each representing common classes of functions involved in quantum networking experiments.

Each plugin defines both client and server commands for RPC calls, along with the corresponding handlers that the controller must process. Additionally, plugins can define message commands to handle non-RPC messages. When the controller starts, it loads and initializes these plugins, registering the provided commands in its RPC server and client and message server.

The ControllerContextManager provides the running context of the controller to a plugin. The context object includes the current controller configuration, the RPC server/client, and the Message server/client.

An example of how to use a plugin to implement message handling can be found in the Tutorial.

Abstract Interfaces

class quantnet_controller.common.plugin.Plugin(name: str, ptype: str, context: ControllerContextManager = None)

Plugin abstract base class

Example:

p = Plugin("My Plugin", PluginType.ROUTING)

Parameters:

• str – Name of the plugin

• ptype – Plugin type

• context – Active context from the running Controller instance

abstractmethod initialize()

Abstract plugin initializtion method

abstractmethod destroy()

Abstract plugin destroy method

abstractmethod reset()

Abstract plugin reset method

get_client_commands() → list

Returns:

A list of registered RPC client protocol commands

Return type:

list

get_server_commands() → list

Returns:

A list of registered RPC server protocol commands

Return type:

list

get_msg_commands() → list

Returns:

A list of registered MsgServer protocol commands (topics)

Return type:

list

class quantnet_controller.common.plugin.RoutingPlugin(name: str, ptype: str, context: ControllerContextManager = None)

abstractmethod find_path(src, dst) → list

find a path through a quantum network topology

Example:

path = find_path("nodeA", "nodeB")

Parameters:

• src – Source node

• dst – Destination node

Returns:

A list of Node objects

Return type:

list

class quantnet_controller.common.plugin.SchedulingPlugin(name: str, ptype: str, context: ControllerContextManager = None)

abstractmethod start()

Abstract method to start the scheduling plugin

abstractmethod schedule()

Abstract method to invoke scheduling backend

class quantnet_controller.common.plugin.MonitoringPlugin(name: str, ptype: str, context: ControllerContextManager = None)

abstractmethod handle_resource_update()

Abstract method to handle resource updates on the monitoring MsgServer topic

class quantnet_controller.common.plugin.ProtocolPlugin(name: str, ptype: str, context: ControllerContextManager = None)

get_schema_paths() → list

Returns:

A list of registered schema metadata

Return type:

list

Request Manager

class quantnet_controller.common.request.RequestType(*values)

class quantnet_controller.common.request.RequestParameter(exp_name: str | None = None, path: ~typing.List[str] | None = <factory>, exp_params: ~typing.Dict[str, ~typing.Any] | None = <factory>)

Quant‑Net RequestParameter

class quantnet_controller.common.request.Request(request_type: ~quantnet_controller.common.request.RequestType, parameters: ~typing.Dict[str, ~typing.Any] = <factory>, rid: str | None = None)

First-class object to track all requests in the system.

add_result(key, value)

Add a result entry directly to the result dictionary.

Args:

key: Result key (e.g., agent_id, ‘error’, ‘metadata’) value: Result value

update_status(code: Code, error=None)

Update the request’s status and optional error information.

Parameters:

• status (str) – New status string (e.g., created, queued, executing, completed, failed)

• error (str | None) – Optional error message to record when status is failed

Returns:

None

to_dict()

Convert Request to dictionary using native dataclass serialization.

property func: Callable | None

Get the callable function.

property payload

Get the callable function.

class quantnet_controller.common.request.RequestManager(ctx, plugin_schema=None, request_type=RequestType.PROTOCOL, dbname=None, exp_def_path=None)

Manage Request objects using a singleton registry per plugin configuration.

This class provides global request tracking while allowing each plugin to maintain its own request type and optional experiment translator.

Variables:

• _instances – Mapping of plugin_key to RequestManager instances.

• _lock – Asyncio lock protecting singleton creation.

• _shared_db_handler – Shared database handler for all managers.

• _shared_active_requests – In‑memory store of active Request objects.

classmethod get_instance(plugin_schema=None, request_type=RequestType.PROTOCOL)

Retrieve a previously‑created RequestManager instance for a given plugin configuration.

Parameters:

• plugin_schema (type | None) – Optional plugin schema class used to differentiate managers.

• request_type (RequestType) – The request type this manager handles.

Returns:

The matching RequestManager instance, or None if it has not been instantiated yet.

Return type:

RequestManager | None

classmethod get_all_active_requests()

Return a mapping of all currently active Request objects across every plugin manager.

Returns:

Dictionary where the key is the request ID and the value is the corresponding Request instance.

Return type:

dict[str, Request]

new_request(payload, parameters=None, rid=None, func=None)

Create and register a new Request.

Parameters:

• payload (Any) – Plugin‑defined request schema (e.g., spgRequest instance) containing plugin‑specific data such as nodes, rate, duration, …

• parameters (dict | None) – Optional execution parameters for the request type (e.g., exp_name, file system path for experiments)

• rid (str | None) – Optional request identifier; a UUID will be generated if omitted

• func (Callable[[Request], Awaitable[Code]] | None) – Optional custom async callable for PROTOCOL requests. Signature should be async def func(request: Request) -> Code.

Returns:

The newly created Request instance, already stored in the in‑memory registry and persisted to the database.

Return type:

Request

async get_request(rid, include_result=False, raw=False)

Retrieve a Request instance by its identifier.

Parameters:

• rid (str) – Unique request identifier

• include_result (bool) – When True and the request has completed, also fetch the associated experiment result

• raw (json) – When True, return the raw database record instead of a Request object

Returns:

The matching Request object, or None if the identifier is unknown

Return type:

Request | None

async find_requests(raw=False, filter={}, **kwargs)

Locate requests that satisfy the provided filter criteria.

Parameters:

filters (Any) – Arbitrary keyword filters mapping field names to desired values (e.g., type="experiment", status="queued")

Returns:

List of matching Request instances

Return type:

list[Request]

del_request(rid)

Remove a request from both in‑memory tracking and persistent storage.

Parameters:

rid (str) – Unique request identifier

Returns:

True if the request was successfully deleted, False otherwise

Return type:

bool

async noSchedule(request, blocking=True)

Execute a request immediately without placing it on the queue.

Parameters:

• request (Request) – The Request to be executed

• blocking (bool) – If True the coroutine waits for the request to finish; if False a Future is returned instead

Returns:

Execution result code

Return type:

Code

async schedule(request, blocking=True)

Enqueue a request for later execution.

Parameters:

• request (Request) – The Request to be queued

• blocking (bool) – True to wait for the request to complete, False to obtain a Task that can be awaited later

Returns:

Code on blocking execution, otherwise an asyncio.Task instance

Return type:

Code | asyncio.Task

async process_queue()

Consume all pending requests from the internal queue and execute them sequentially.

Returns:

Code.OK when the queue has been fully processed

Return type:

Code

async get_experiment_result(id, timeout=60)

Retrieve the result of an experiment request from the translator.

Parameters:

• id (str) – Identifier of the experiment request

• timeout (int) – Maximum time (seconds) to wait for the result

Returns:

Experiment result object, or None if unavailable

Return type:

Any