Utilities for OmniConnector configuration and validation.
module-attribute ¶
ConnectorPurpose = Literal[
"request_forwarding", "kv_transfer", "orchestrator"
]
build_stage_connectors(
stage_id: int,
connectors_config: dict[str, Any],
purpose: str = "request_forwarding",
) -> dict[tuple[str, str], Any] | None
Instantiate OmniConnectors for a stage based on config.
Deprecated: prefer get_stage_connector_config plus the unified connector factory. Kept as a thin shim so legacy callers keep working.
compute_connector_zmq_port(
base_port: int,
*,
purpose: ConnectorPurpose,
from_stage: int,
local_rank: int = 0,
replica_id: int = 0,
) -> int
Return a collision-free connector endpoint for an edge and worker.
create_connectors_from_config(
connectors_config: dict[tuple[str, str], ConnectorSpec],
purpose: str = "request_forwarding",
caller_stage_id: int | str | None = None,
is_sender: bool | None = None,
) -> dict[tuple[str, str], OmniConnectorBase]
Create connectors from config.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
connectors_config | dict[tuple[str, str], ConnectorSpec] | A dictionary of connector configurations. | required |
Returns:
| Type | Description |
|---|---|
dict[tuple[str, str], OmniConnectorBase] | A dictionary of connectors. |
get_connectors_config_for_stage(
transfer_config: OmniTransferConfig | None,
stage_id: str | int,
) -> dict[str, Any]
Extract connector configurations relevant for a specific stage worker.
Returns a dict compatible with worker initialization: { "from_stage_X": { "spec": { "name": "ConnectorName", "extra": {...} } }, ... }
get_stage_connector_config(
transfer_config: OmniTransferConfig | None,
stage_id: int,
) -> dict[str, Any]
Return the serialized connector config payload for a specific stage.
initialize_connectors_from_config(
config_path: str | Path | None = None,
purpose: str = "request_forwarding",
caller_stage_id: int | str | None = None,
is_sender: bool | None = None,
) -> tuple[
OmniTransferConfig | None,
dict[tuple[str, str], OmniConnectorBase],
]
Initialize connectors from configuration file.
Returns:
| Name | Type | Description |
|---|---|---|
tuple | tuple[OmniTransferConfig | None, dict[tuple[str, str], OmniConnectorBase]] | (OmniTransferConfig, dict of {(from, to): connector_instance}) |
initialize_orchestrator_connectors(
config_path: str | None,
) -> tuple[
OmniTransferConfig | None,
dict[tuple[str, str], OmniConnectorBase],
]
Initialize connectors shared at orchestrator level. Args: config_path: The path to the configuration file. Returns: A tuple containing the OmniTransferConfig and a dictionary of connectors.
load_omni_transfer_config(
config_path: str | Path | None = None,
config_dict: dict[str, Any] | None = None,
) -> OmniTransferConfig | None
Load OmniTransferConfig from file or dict.
resolve_connector_spec(
spec: ConnectorSpec,
*,
stage_id: int,
role: str | None,
purpose: ConnectorPurpose = "request_forwarding",
local_rank: int = 0,
replica_id: int = 0,
) -> ConnectorSpec
Copy and resolve a connector spec for one stage worker.
resolve_omni_kv_config_for_stage(
transfer_cfg: OmniTransferConfig | None,
stage_id: int | str,
) -> tuple[dict[str, Any] | None, str | None, str | None]
Resolve connector configuration for a specific stage (Sender/Receiver).
This determines the primary connector configuration to be injected into the engine arguments, prioritizing outgoing edges (Sender role).