NVIDIA-NeMo · andreatgretel · Jul 8, 2026 · Jul 7, 2026 · Jul 7, 2026 · Jul 8, 2026
@@ -107,7 +107,7 @@ DatasetBuilder.build()
   → collect TaskTraces, emit telemetry
 ```
 
-Row-group admission is fixed by default in the dataset-builder path: the configured row-group concurrency is the hard in-flight cap. The scheduler also has an internal adaptive row-group mode for direct use that only raises a soft target up to that cap; it is additive ramp-up, not AIMD shrink/recovery behavior.
+Dataset-builder row-group admission is fixed: `RunConfig.max_concurrent_row_groups` (default `3`) is the hard in-flight cap. `buffer_size` controls the records per group, so raising the cap can increase active memory. The scheduler's adaptive row-group mode remains internal and is not wired through the public interface.
 
 When request admission is available, async scheduling may use request-pressure snapshots as a read-only advisory during fair-queue selection. A request-pressured task can be skipped for an eligible peer without mutating request-admission state; provider/model/domain request limits remain owned by request admission.
 

@@ -103,17 +103,17 @@ Within each column, cells are processed **in parallel** up to the configured lim
 
 ### Concurrency Formula
 
-At any moment, the number of concurrent LLM requests is:
+For a given model, async request concurrency is bounded by:
 
 ```python
 concurrent_requests = min(
-    buffer_size,                # Records in an active row group
-    current_admission_limit,    # AIMD-managed limit (≤ max_parallel_requests)
-    remaining_cells_in_column   # Cells left to generate
+    active_ready_model_cells,  # Ready cells across active row groups
+    current_admission_limit,   # AIMD-managed limit (≤ max_parallel_requests)
+    max_in_flight_tasks,       # Global scheduler task-lease ceiling
 )
 ```
 
-`max_parallel_requests` sets the **ceiling**. The actual limit (`current_admission_limit`) is managed at runtime by an AIMD (Additive Increase / Multiplicative Decrease) request-admission controller that reacts to rate-limit signals from the inference server:
+`active_ready_model_cells` depends on the column DAG and the active row-group horizon. Each active row group contributes up to `buffer_size` rows per ready model column. Other work may consume part of the global `max_in_flight_tasks` budget. `max_parallel_requests` sets the **per-model ceiling**. The actual limit (`current_admission_limit`) is managed at runtime by an AIMD (Additive Increase / Multiplicative Decrease) request-admission controller that reacts to rate-limit signals from the inference server:
 
 - **During optional startup ramp**: when `startup_ramp_seconds` is greater than 0, a new request domain starts at one concurrent request and increases linearly toward `max_parallel_requests` over that duration.
 - **On the first 429 in a burst**: the limit is reduced by a configurable factor (default: 25% reduction) and a cooldown is applied. Further 429s from already in-flight requests in the same burst do not reduce the limit again — they release their permits and hold the limit steady.
@@ -192,6 +192,22 @@ designer.set_run_config(run_config)
 
 ---
 
+### `max_concurrent_row_groups` (RunConfig)
+
+Sets the fixed hard cap on async row groups that may be active at once. The default is `3`; each row group contains up to `buffer_size` records.
+
+```python
+run_config = dd.RunConfig(
+    buffer_size=1_000,
+    max_concurrent_row_groups=8,
+)
+designer.set_run_config(run_config)
+```
+
+A wider horizon can expose more ready work to high-capacity endpoints, but it retains more row data in memory and may delay checkpoints. It does not adapt during a run or replace `max_parallel_requests`, which remains the per-model request ceiling. Treat this as an expert override: benchmark before and after changing it.
+
+---
+
 ## Resuming Interrupted Runs
 
 Long generation jobs can be resumed from checkpoints by passing `resume` to `DataDesigner.create()` or `data-designer create --resume`.

@@ -135,6 +135,8 @@ class RunConfig(ConfigBase):
             monitoring begins. Must be >= 1. Default is 10.
         buffer_size: Number of records in each row group during dataset generation.
             Must be > 0. Default is 1000.
+        max_concurrent_row_groups: Maximum number of row groups the async scheduler may
+            keep active at once. Must be >= 1. Default is 3.
         max_in_flight_tasks: Maximum number of async scheduler tasks that may hold task
             leases at once. Tasks may be executing, awaiting I/O, or waiting on model
             request admission. Model API request concurrency is controlled separately by
@@ -178,6 +180,11 @@ class RunConfig(ConfigBase):
     shutdown_error_rate: float = Field(default=0.5, ge=0.0, le=1.0)
     shutdown_error_window: int = Field(default=10, ge=1)
     buffer_size: int = Field(default=1000, gt=0)
+    max_concurrent_row_groups: int = Field(
+        default=3,
+        ge=1,
+        description="Maximum number of row groups the async scheduler may keep active at once.",
+    )
     max_in_flight_tasks: int = Field(
         default=1024,
         ge=1,

@@ -91,6 +91,19 @@ def test_run_config_accepts_disabled_dropped_column_preservation() -> None:
     assert run_config.preserve_dropped_columns is False
 
 
+def test_run_config_defaults_max_concurrent_row_groups_to_three() -> None:
+    assert RunConfig().max_concurrent_row_groups == 3
+
+
+def test_run_config_accepts_custom_max_concurrent_row_groups() -> None:
+    assert RunConfig(max_concurrent_row_groups=8).max_concurrent_row_groups == 8
+
+
+def test_run_config_rejects_invalid_max_concurrent_row_groups() -> None:
+    with pytest.raises(ValidationError, match="max_concurrent_row_groups"):
+        RunConfig(max_concurrent_row_groups=0)
+
+
 def test_run_config_defaults_max_in_flight_tasks_to_1024() -> None:
     assert RunConfig().max_in_flight_tasks == 1024
 

@@ -963,6 +963,7 @@ def on_before_checkpoint(rg_id: int, rg_size: int) -> None:
             tracker=tracker,
             row_groups=row_groups,
             buffer_manager=buffer_manager,
+            max_concurrent_row_groups=self._resource_provider.run_config.max_concurrent_row_groups,
             max_in_flight_tasks=max_in_flight_tasks,
             max_model_task_admission=max_model_task_admission,
             on_finalize_row_group=on_finalize_row_group,

@@ -403,7 +403,12 @@ def __init__(self, **kwargs: object) -> None:
     model_registry.request_admission = request_admission
     provider = SimpleNamespace(
         model_registry=model_registry,
-        run_config=SimpleNamespace(max_in_flight_tasks=64, progress_interval=5.0, display_tui=False),
+        run_config=SimpleNamespace(
+            max_concurrent_row_groups=8,
+            max_in_flight_tasks=64,
+            progress_interval=5.0,
+            display_tui=False,
+        ),
     )
     processor_runner = MagicMock()
     processor_runner.has_processors_for.return_value = False
@@ -420,6 +425,7 @@ def __init__(self, **kwargs: object) -> None:
 
     assert captured_kwargs["request_pressure_provider"] is request_admission
     assert captured_kwargs["request_pressure_advisory"] is True
+    assert captured_kwargs["max_concurrent_row_groups"] == 8
     assert captured_kwargs["max_in_flight_tasks"] == 64
     assert captured_kwargs["max_model_task_admission"] == 64
 
@@ -436,7 +442,12 @@ def __init__(self, **kwargs: object) -> None:
     model_registry.request_admission = None
     provider = SimpleNamespace(
         model_registry=model_registry,
-        run_config=SimpleNamespace(max_in_flight_tasks=64, progress_interval=5.0, display_tui=False),
+        run_config=SimpleNamespace(
+            max_concurrent_row_groups=3,
+            max_in_flight_tasks=64,
+            progress_interval=5.0,
+            display_tui=False,
+        ),
     )
     processor_runner = MagicMock()
     processor_runner.has_processors_for.return_value = False