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, str | int]]])#

Bases: object

resources#

slots_available(rtype: str) → int#

class Allocator#

Bases: object

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

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

Bases: object

Class representing resources available on the computer

Parameters:: pool – The resource pool specification

Resource specification#

The specification for the resource pool is adopted from the ctest schema:

local:
  <resource name>:
  - id: str
    slots: int

For example, a machine with 4 GPUs may have

local:
  gpus:
  - id: "01"
    slots: 1
  - id: "02"
    slots: 1
  - id: "03"
    slots: 1
  - id: "04"
    slots: 1

canary adopts a similar layout:

resource_pool:
  additional_properties: {}
  resources:
    <resource name>:
    - id: str
      slots: int
    ...

For example, a machine having 8 CPUs and 4 GPUs may have

resource_pool:
  additional_properties: {}
  resources:
    gpus:
    - id: "01"
      slots: 1
    - id: "02"
      slots: 1
    - id: "03"
      slots: 1
    - id: "04"
      slots: 1
    cpus:
    - id: "01"
      slots: 1
    - id: "02"
      slots: 1
    - id: "03"
      slots: 1
    - id: "04"
      slots: 1
    - id: "05"
      slots: 1
    - id: "06"
      slots: 1
    - id: "07"
      slots: 1
    - id: "08"
      slots: 1

additional_properties: dict[str, Any]#

resources: dict[str, list[dict[str, str | int]]]#

slots_per_resource_type: Counter[str]#

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

property types: list[str]#

empty() → bool#

count(type: str) → int#

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

pop(type: str) → list[dict[str, str | int]] | None#

getstate() → dict[str, Any]#

clear() → None#

populate(**kwds: int) → 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, list[dict]]#

Returns resources available to the test

local[i] = {<type>: [{‘id’: <id>, ‘slots’: <slots>}, …], … }

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

make_resource_pool(config: CanaryConfig)#

exception ResourceUnavailable#: Bases: Exception

exception EmptyResourcePoolError#: Bases: Exception