nemo_gym.telemetry.concurrency

The live state of Gym’s concurrency limits: the configured limit, the slots in use, and who is waiting.

A queue-wait duration shows that work waited, not why. A dashboard that shows “limit 100, active 100, queued 75” for each limit answers why: the limit is full and work is piling up behind it. MeteredSemaphore is a drop-in asyncio.BoundedSemaphore that keeps those numbers, and the process observer exports them, by limit:

gym.concurrency.limit: the configured number of slots.
gym.concurrency.active: slots held right now.
gym.concurrency.available: slots free right now, limit - active.
gym.concurrency.queue_depth: callers waiting for a slot right now.
gym.concurrency.admission_timeouts_total: waits that ran out of the limit’s admission timeout. Reported only for a limit that has an admission timeout, because no other limit can time out.
gym.concurrency.admission_cancelled_total: waits cancelled before a slot was free, for example because the caller’s own request was cancelled.

Every point carries nemo.gym.concurrency.component (ConcurrencyComponent), nemo.gym.concurrency.limit_name (a name the code chose for what the limit bounds, such as upstream_requests), and nemo.gym.server.name when the limit belongs to a server. All three are bounded: no rollout, request, or sandbox ID is ever an attribute. Limits with the same attributes in one process are summed.

A component without a configured limit creates no MeteredSemaphore, so it reports no series at all rather than a limit of zero. Acquiring and releasing cost a few integer updates and no OpenTelemetry call. The exporter’s background thread reads the totals through observable instruments, which exist only while the process observer runs, that is, while metrics are exported.

Module Contents

Classes

Name	Description
`MeteredSemaphore`	An `asyncio.BoundedSemaphore` that counts its slots in use and its waiters for export.
`_LimitKey`	-
`_LimitState`	-

Functions

Name	Description
`_count_admission_failure`	-
`_limit_states`	Sum the live limits that share attributes. A garbage-collected limit no longer reports.
`_observe_admission_failures`	-
`_observe_limits`	-
`_observer_running`	-
`_register`	-
`_reset_for_testing`	-
`register_concurrency_instruments`	Create the `gym.concurrency.*` instruments on `meter`. The process observer calls this once per meter.

Data

ACTIVE_INSTRUMENT

ADMISSION_CANCELLED_INSTRUMENT

ADMISSION_TIMEOUTS_INSTRUMENT

AVAILABLE_INSTRUMENT

COMPONENT_ATTRIBUTE

ConcurrencyComponent

LIMIT_INSTRUMENT

LIMIT_NAME_ATTRIBUTE

QUEUE_DEPTH_INSTRUMENT

_ADMISSION_CANCELLED

_ADMISSION_TIMEOUTS

_LIMITS

_LOCK

logger

API

class nemo_gym.telemetry.concurrency.MeteredSemaphore(
    limit: int,
    component: nemo_gym.telemetry.concurrency.ConcurrencyComponent,
    limit_name: str,
    server_name: typing.Optional[str] = None,
    admission_timeout_s: typing.Optional[float] = None
)

Bases: BoundedSemaphore

An asyncio.BoundedSemaphore that counts its slots in use and its waiters for export.

Use it wherever a component has a configured concurrency limit; async with, acquire, and release behave as on the base class.

Parameters:

limit

int

The configured number of slots, at least one.

component

ConcurrencyComponent

ConcurrencyComponent — The kind of component that owns the limit.

limit_name

str

A fixed name for what the limit bounds, such as rollouts or upstream_requests.

server_name

Optional[str]Defaults to None

The owning server’s configured name, or None outside a server.

admission_timeout_s

Optional[float]Defaults to None

How long one caller may wait for a slot before acquire raises TimeoutError, or None to wait indefinitely. Pass the timeout here rather than wrapping acquire in asyncio.wait_for, which this class would count as a cancellation.

_active

= 0

_key

= _LimitKey(component, limit_name, server_name)

_queued

= 0

active

int

Slots held right now.

admission_timeout_s

Optional[float]

How long one caller may wait for a slot, or None to wait indefinitely.

attributes

dict[str, str]

The metric attributes this limit reports under.

limit

int

The configured number of slots.

queued

int

Callers waiting for a slot right now.

nemo_gym.telemetry.concurrency.MeteredSemaphore.acquire() -> typing.Literal[True]

async

Take a slot, waiting for one if all are held.

Raises:

TimeoutError: No slot became free within admission_timeout_s.

nemo_gym.telemetry.concurrency.MeteredSemaphore.release() -> None

Return a slot; raises ValueError when no slot is held, as asyncio.BoundedSemaphore does.

class nemo_gym.telemetry.concurrency._LimitKey()

Bases: NamedTuple

component

str

limit_name

str

server_name

Optional[str]

nemo_gym.telemetry.concurrency._LimitKey.attributes() -> dict[str, str]

class nemo_gym.telemetry.concurrency._LimitState()

Bases: NamedTuple

active

int

limit

int

queued

int

nemo_gym.telemetry.concurrency._count_admission_failure(
    totals: dict[nemo_gym.telemetry.concurrency._LimitKey, int],
    key: nemo_gym.telemetry.concurrency._LimitKey
) -> None

nemo_gym.telemetry.concurrency._limit_states() -> dict[nemo_gym.telemetry.concurrency._LimitKey, nemo_gym.telemetry.concurrency._LimitState]

Sum the live limits that share attributes. A garbage-collected limit no longer reports.

nemo_gym.telemetry.concurrency._observe_admission_failures(
    totals: dict[nemo_gym.telemetry.concurrency._LimitKey, int]
) -> typing.Callable[[Any], list]

nemo_gym.telemetry.concurrency._observe_limits(
    read: typing.Callable[[_LimitState], int]
) -> typing.Callable[[Any], list]

nemo_gym.telemetry.concurrency._observer_running() -> bool

nemo_gym.telemetry.concurrency._register(
    semaphore: nemo_gym.telemetry.concurrency.MeteredSemaphore
) -> None

nemo_gym.telemetry.concurrency._reset_for_testing() -> None

nemo_gym.telemetry.concurrency.register_concurrency_instruments(
    meter: typing.Any
) -> None

Create the gym.concurrency.* instruments on meter. The process observer calls this once per meter.

nemo_gym.telemetry.concurrency.ACTIVE_INSTRUMENT = 'gym.concurrency.active'

nemo_gym.telemetry.concurrency.ADMISSION_CANCELLED_INSTRUMENT = 'gym.concurrency.admission_cancelled_total'

nemo_gym.telemetry.concurrency.ADMISSION_TIMEOUTS_INSTRUMENT = 'gym.concurrency.admission_timeouts_total'

nemo_gym.telemetry.concurrency.AVAILABLE_INSTRUMENT = 'gym.concurrency.available'

nemo_gym.telemetry.concurrency.COMPONENT_ATTRIBUTE = 'nemo.gym.concurrency.component'

nemo_gym.telemetry.concurrency.ConcurrencyComponent = Literal['rollout_driver', 'agent_server', 'model_server', 'resources_server', 'e...

nemo_gym.telemetry.concurrency.LIMIT_INSTRUMENT = 'gym.concurrency.limit'

nemo_gym.telemetry.concurrency.LIMIT_NAME_ATTRIBUTE = 'nemo.gym.concurrency.limit_name'

nemo_gym.telemetry.concurrency.QUEUE_DEPTH_INSTRUMENT = 'gym.concurrency.queue_depth'

nemo_gym.telemetry.concurrency._ADMISSION_CANCELLED: dict[_LimitKey, int] = {}

nemo_gym.telemetry.concurrency._ADMISSION_TIMEOUTS: dict[_LimitKey, int] = {}

nemo_gym.telemetry.concurrency._LIMITS: WeakSet[MeteredSemaphore] = weakref.WeakSet()

nemo_gym.telemetry.concurrency._LOCK = threading.Lock()

nemo_gym.telemetry.concurrency.logger = logging.getLogger(__name__)