nemo_gym.telemetry.resources_operations

A shared way for resources servers to report bounded work: how long it queued, how long it ran, and how it ended.

Many resources servers limit expensive work, such as a verifier subprocess or a judge call, with a semaphore. When that limit is too small, verification waits in a queue that no request metric shows, and the server looks slow without looking busy. Wrapping the limited work in resources_operation makes that visible:

async with resources_operation(self._semaphore, operation=“verify”, executor=“subprocess”) as op: result = await run_the_verifier() if it_timed_out: op.outcome = “timeout”

It reports, by operation and executor:

gym.resources.operation.queue_duration_ms (histogram, milliseconds): time waiting for the limit.
gym.resources.operation.duration_ms (histogram, milliseconds, also by outcome): time running inside the limit. The count of each outcome, such as timeout or crashed, comes from this histogram.
gym.resources.operation.queued and gym.resources.operation.active: how many operations wait and run right now.

operation is one of verify, tool, judge, or other, and executor is one of inline, thread, process, subprocess, or other, so the attributes stay bounded. Everything is recorded only while metrics are exporting; otherwise the wrapper only enters the limit.

Module Contents

Classes

Name	Description
`ResourcesOperation`	One limited operation. Set `outcome` inside the block when the work ended other than by returning normally.
`_OperationMetrics`	The instruments and live counts behind `resources_operation`, one set per meter.

Functions

Name	Description
`_metrics`	-
`resources_operation`	Enter `limit`, which is any async context manager such as an `asyncio.Semaphore`, and report the operation.

Data

ACTIVE_INSTRUMENT

DURATION_INSTRUMENT

EXECUTORS

EXECUTOR_ATTRIBUTE

OPERATIONS

OPERATION_ATTRIBUTE

OPERATION_BOUNDARIES_MS

OUTCOMES

OUTCOME_ATTRIBUTE

QUEUED_INSTRUMENT

QUEUE_DURATION_INSTRUMENT

_METRICS

API

class nemo_gym.telemetry.resources_operations.ResourcesOperation()

One limited operation. Set outcome inside the block when the work ended other than by returning normally.

outcome

Optional[str] = None

class nemo_gym.telemetry.resources_operations._OperationMetrics(
    meter: typing.Any
)

The instruments and live counts behind resources_operation, one set per meter.

active

dict[tuple[str, str], int] = {}

duration

queue_duration

queued

dict[tuple[str, str], int] = {}

nemo_gym.telemetry.resources_operations._metrics() -> typing.Optional[nemo_gym.telemetry.resources_operations._OperationMetrics]

nemo_gym.telemetry.resources_operations.resources_operation(
    limit: typing.Any,
    operation: str,
    executor: str
) -> typing.AsyncIterator[nemo_gym.telemetry.resources_operations.ResourcesOperation]

async

Enter limit, which is any async context manager such as an asyncio.Semaphore, and report the operation.

An exception sets the outcome to error, or cancelled for a cancellation, unless the block already set one.

nemo_gym.telemetry.resources_operations.ACTIVE_INSTRUMENT = 'gym.resources.operation.active'

nemo_gym.telemetry.resources_operations.DURATION_INSTRUMENT = 'gym.resources.operation.duration_ms'

nemo_gym.telemetry.resources_operations.EXECUTORS = frozenset({'inline', 'thread', 'process', 'subprocess', 'other'})

nemo_gym.telemetry.resources_operations.EXECUTOR_ATTRIBUTE = 'nemo.gym.resources.executor'

nemo_gym.telemetry.resources_operations.OPERATIONS = frozenset({'verify', 'tool', 'judge', 'other'})

nemo_gym.telemetry.resources_operations.OPERATION_ATTRIBUTE = 'nemo.gym.resources.operation'

nemo_gym.telemetry.resources_operations.OPERATION_BOUNDARIES_MS: tuple[float, ...] = (1, 5, 10, 50, 100, 500, 1000, 5000, 10000, 30000, 60000, 300000, 1800000)

nemo_gym.telemetry.resources_operations.OUTCOMES = frozenset({'ok', 'error', 'timeout', 'cancelled', 'crashed'})

nemo_gym.telemetry.resources_operations.OUTCOME_ATTRIBUTE = 'nemo.gym.resources.operation.outcome'

nemo_gym.telemetry.resources_operations.QUEUED_INSTRUMENT = 'gym.resources.operation.queued'

nemo_gym.telemetry.resources_operations.QUEUE_DURATION_INSTRUMENT = 'gym.resources.operation.queue_duration_ms'

nemo_gym.telemetry.resources_operations._METRICS: WeakKeyDictionary[Any, _OperationMetrics] = weakref.WeakKeyDictionary()