> This page is for version Main (default).
> For other versions, use one of these documentation indexes:
> - Main (default): https://docs.nvidia.com/nemo/gym/main/llms.txt
> - 0.7.0: https://docs.nvidia.com/nemo/gym/v0.7.0/llms.txt
> - 0.6.0: https://docs.nvidia.com/nemo/gym/v0.6.0/llms.txt
> - 0.5.1: https://docs.nvidia.com/nemo/gym/v0.5.1/llms.txt
> - 0.5.0: https://docs.nvidia.com/nemo/gym/v0.5.0/llms.txt
> - 0.4.0: https://docs.nvidia.com/nemo/gym/v0.4.0/llms.txt
> - 0.3.0: https://docs.nvidia.com/nemo/gym/v0.3.0/llms.txt
> - 0.2.1: https://docs.nvidia.com/nemo/gym/v0.2.1/llms.txt

> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.nvidia.com/nemo/gym/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/gym/_mcp/server.

# 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`](#nemo_gym-telemetry-concurrency-MeteredSemaphore) | An `asyncio.BoundedSemaphore` that counts its slots in use and its waiters for export. |
| [`_LimitKey`](#nemo_gym-telemetry-concurrency-_LimitKey)               | -                                                                                      |
| [`_LimitState`](#nemo_gym-telemetry-concurrency-_LimitState)           | -                                                                                      |

### Functions

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

### Data

[`ACTIVE_INSTRUMENT`](#nemo_gym-telemetry-concurrency-ACTIVE_INSTRUMENT)

[`ADMISSION_CANCELLED_INSTRUMENT`](#nemo_gym-telemetry-concurrency-ADMISSION_CANCELLED_INSTRUMENT)

[`ADMISSION_TIMEOUTS_INSTRUMENT`](#nemo_gym-telemetry-concurrency-ADMISSION_TIMEOUTS_INSTRUMENT)

[`AVAILABLE_INSTRUMENT`](#nemo_gym-telemetry-concurrency-AVAILABLE_INSTRUMENT)

[`COMPONENT_ATTRIBUTE`](#nemo_gym-telemetry-concurrency-COMPONENT_ATTRIBUTE)

[`ConcurrencyComponent`](#nemo_gym-telemetry-concurrency-ConcurrencyComponent)

[`LIMIT_INSTRUMENT`](#nemo_gym-telemetry-concurrency-LIMIT_INSTRUMENT)

[`LIMIT_NAME_ATTRIBUTE`](#nemo_gym-telemetry-concurrency-LIMIT_NAME_ATTRIBUTE)

[`QUEUE_DEPTH_INSTRUMENT`](#nemo_gym-telemetry-concurrency-QUEUE_DEPTH_INSTRUMENT)

[`_ADMISSION_CANCELLED`](#nemo_gym-telemetry-concurrency-_ADMISSION_CANCELLED)

[`_ADMISSION_TIMEOUTS`](#nemo_gym-telemetry-concurrency-_ADMISSION_TIMEOUTS)

[`_LIMITS`](#nemo_gym-telemetry-concurrency-_LIMITS)

[`_LOCK`](#nemo_gym-telemetry-concurrency-_LOCK)

[`logger`](#nemo_gym-telemetry-concurrency-logger)

### API

```python
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`](#nemo_gym-telemetry-concurrency-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]` — default: None

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

---

**`admission_timeout_s`** `Optional[float]` — default: 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.

---

```python
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`.

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

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

```python
class nemo_gym.telemetry.concurrency._LimitKey()
```

**Bases:** `NamedTuple`

**`component`** `str`

---

**`limit_name`** `str`

---

**`server_name`** `Optional[str]`

---

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

```python
class nemo_gym.telemetry.concurrency._LimitState()
```

**Bases:** `NamedTuple`

**`active`** `int`

---

**`limit`** `int`

---

**`queued`** `int`

---

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

```python
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.

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

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

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

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

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

```python
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.

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

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

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

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

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

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

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

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

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

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

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

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

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

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