rpool#

class Outcome(ok: bool | None = None, reason: str | None = None)#

Bases: object

ok: bool#

reason: str | None#

class AllocationTransaction(inventory: Inventory)#

Bases: object

inventory#

deltas: list[tuple[str, str, int]]#

record(rtype: str, rid: str, slots: int) → None#

rollback() → None#

class Inventory(resources: dict[str, list[dict[str, Any]]])#

Bases: object

resources#

slots_available(rtype: str) → int#

class Allocator#

Bases: object

acquire(inventory: Inventory, rtype: str, slots: int) → dict[str, Any]#

class Node(id: str, resources: dict[str, list[dict[str, Any]]] | None = None, *, additional_properties: dict[str, Any] | None = None)#

Bases: object

Single-node resource inventory.

This is essentially the old flat ResourcePool behavior, scoped to one node. Resource IDs are node-local.

id#

resources: dict[str, list[dict[str, Any]]]#

slots_per_resource_type: Counter[str]#

additional_properties#

property types: list[str]#

empty() → bool#

slots_available(rtype: str) → int#

count(rtype: str) → int#

accommodates(request: list[dict[str, Any]]) → Outcome#: Determine if this node can accommodate a per-node resource request.

score(request: list[dict[str, Any]]) → float#

Score this node after satisfying request.

Higher score means more residual capacity. This borrows the idea from dpool.py, but computes the score without mutating the node.

checkout(request: list[dict[str, Any]]) → dict[str, list[dict]]#

Check resources out of this node.

Returned resource specs include the node ID.

checkin(resources: dict[str, list[dict]]) → None#

pop(rtype: str) → list[dict[str, Any]] | None#

getstate() → dict[str, Any]#

has_resource(rtype: str) → bool#

get_resource(rtype: str, default: list[dict[str, Any]] | None = None) → list[dict[str, Any]] | None#

set_resource(rtype: str, specs: list[dict[str, Any]]) → None#

set_resources(resources: dict[str, list[dict[str, Any]]]) → None#

set_resource_count(rtype: str, count: int) → None#

set_slots_per_resource(rtype: str, slots: int) → None#

multiply_slots_per_resource(rtype: str, factor: int) → None#

class ResourcePool(pool: dict[str, Any] | None = None, allow_multinode: bool = True)#

Bases: object

Topology-aware resource pool.

Canonical resource-pool layout:

resource_pool:
  allow_multinode: true
  additional_properties: {}
  nodes:
  - id: node-0
    resources:
      cpus:
      - id: "0"
        slots: 1
      gpus:
      - id: "0"
        slots: 1

Resource IDs are node-local. Checked-out resources include the owning node:

{
    "gpus": [
        {"node": "node-0", "id": "0", "slots": 1}
    ]
}

Multi-node semantics:

Without an explicit nodes request, checkout is single-node.
With {"type": "nodes", "slots": N}, the remaining resource request is interpreted as per-node.

additional_properties: dict[str, Any]#

nodes: list[Node]#

property types: list[str]#

property resources: dict[str, list[dict[str, Any]]]#

Aggregate view of resources across all nodes.

This is primarily a convenience view. Resource specs in this aggregate include the owning node ID.

property slots_per_resource_type: Counter[str]#

empty() → bool#

clear() → None#

dump(file: IO[Any]) → None#

getstate() → dict[str, Any]#

fill(pool: dict[str, Any]) → None#

populate(**kwds: int) → None#

node_ids() → list[str]#

get_node(node_id: str) → Node#

get_property(name: str) → Any#

count(rtype: str) → int#

count_by_node(rtype: str) → dict[str, int]#

slots_available(rtype: str) → int#

slots_by_node(rtype: str) → dict[str, int]#

count_per_node(rtype: str) → int#

pop(rtype: str) → list[dict[str, Any]] | None#

accommodates(request: list[dict[str, Any]]) → Outcome#: Determine if the resources for this test are available.

checkout(request: list[dict[str, Any]], **kwds: Any) → dict[str, dict]#

Returns resources available to the test.

Returned resources have the form:

{
    "metadata": {},
    "resources": {
        "<type>": [
            {"node": "<node-id>", "id": "<local-resource-id>", "slots": N},
            ...
        ],
        ...
    }
}

For multi-node checkout, nodes=N means the non-node resource request is applied independently to each selected node.

checkin(allocation: dict[str, dict]) → None#

first_node() → Node#

ensure_node(node_id: str | None = None) → Node#

add_node(node: Node) → None#

rebuild_node_index() → None#

property allow_multinode: bool#

set_resource_count(rtype: str, count: int, *, node_id: str | None = None) → None#

set_slots_per_resource(rtype: str, slots: int, *, node_id: str | None = None) → None#

multiply_slots_per_resource(rtype: str, factor: int, *, node_id: str | None = None) → None#

make_resource_pool(config: CanaryConfig) → ResourcePool#

exception ResourceUnavailable#: Bases: Exception

exception EmptyResourcePoolError#: Bases: Exception