[data][llm] Promote max_tasks_in_flight_per_actor to a first-class config field and adjust defaults

[jeffreywang-anyscale]

Why

Ray Data LLM hardcoded DEFAULT_MAX_TASKS_IN_FLIGHT = 16 instead of using Ray Data's actor-pool fallback, which (a) didn't track max_concurrent_batches when users tuned it and (b) bypassed both DataContext.max_tasks_in_flight_per_actor and the env-var override of the factor.

What changes?

• New top-level field OfflineProcessorConfig.max_tasks_in_flight_per_actor: Optional[int] = None.
• Removed the DEFAULT_MAX_TASKS_IN_FLIGHT = 16 constant; engine processors pass config.max_tasks_in_flight_per_actor straight through to ActorPoolStrategy (including None).
• Default in-flight cap: hardcoded 16 → max_concurrent_batches × FACTOR, resolved by Ray Data's actor pool.
• DataContext.max_tasks_in_flight_per_actor and RAY_DATA_ACTOR_DEFAULT_MAX_TASKS_IN_FLIGHT_TO_MAX_CONCURRENCY_FACTOR are now honored (previously bypassed by the explicit 16).
• experimental["max_tasks_in_flight_per_actor"] is deprecated: migrated to the new field at construction with a logger.warning. Top-level field wins if both are set.

Original API

OfflineProcessorConfig(
    ...,
    experimental={"max_tasks_in_flight_per_actor": 32},  # only knob
)

New API

OfflineProcessorConfig(
    ...,
    max_concurrent_batches=8,           # unchanged
    max_tasks_in_flight_per_actor=32,   # new top-level field, Optional[int]
)

Behavior changes

• Users who set RAY_DATA_ACTOR_DEFAULT_MAX_TASKS_IN_FLIGHT_TO_MAX_CONCURRENCY_FACTOR now have their override honored by Ray Data LLM (previously ignored).
• Setting via experimental still works but logs a deprecation warning. The top-level field overrides experimental if both are set.

max_concurrent_batches	max_tasks_in_flight_per_actor	Ray actor max_concurrency	Effective in-flight cap
unset (default 8)	unset (None)	8	16
16	unset (None)	16	32
unset (default 8)	50	8	50
16	50	16	50

Related issues

Link related issues: "Fixes #1234", "Closes #1234", or "Related to #1234".

Additional information

Optional: Add implementation details, API changes, usage examples, screenshots, etc.

[gemini-code-assist]

Code Review

This pull request promotes max_tasks_in_flight_per_actor from an experimental configuration to a top-level field in OfflineProcessorConfig and its subclasses. It introduces deprecation warnings for the experimental key and implements a resolution strategy that defaults to a calculated value based on max_concurrent_batches. Feedback identifies a potential type mismatch where a float could be assigned to an integer field when bypassing Pydantic validation, suggesting an explicit integer cast to ensure compatibility with Ray Data's actor pool.

[gemini-code-assist]

The resolved value for max_tasks_in_flight_per_actor can be a float, particularly when calculated using DEFAULT_ACTOR_MAX_TASKS_IN_FLIGHT_TO_MAX_CONCURRENCY_FACTOR, which is defined as a float. The max_tasks_in_flight_per_actor field is typed as Optional[int], but using object.__setattr__ bypasses Pydantic's type coercion. This could result in a float value being passed to ray.data.ActorPoolStrategy, which expects an integer and may lead to unexpected behavior or a runtime error.

To ensure an integer is always assigned, the resolved value should be explicitly cast to int. This would also align with the original logic in Ray Data's actor pool, which performs this integer conversion.

Suggested change

	object.__setattr__(self, "max_tasks_in_flight_per_actor", resolved)
	object.__setattr__(self, "max_tasks_in_flight_per_actor", int(resolved))

[Aydin-ab]

making a bit more explicit in the doc that it's 2 * max_concurrent_batches

[kouroshHakha]

The approach is sound — using a Pydantic mode="after" validator to eagerly resolve the None sentinel is clean, and the resolution order (explicit > experimental > formula) is implemented correctly. The behavioral no-op for default users (8×2=16) is a good property.

Note

This review was co-written with AI assistance (Claude Code).