fix: use group-level concurrency in fully-async rollout

[JayPritchet]

Problem

The fully-async worker receives a concurrency value derived from sglang_server_concurrency, which represents per-sample concurrency:

slime/slime/rollout/fully_async_rollout.py

Lines 58 to 60 in 4c193f1

	_global_worker = AsyncRolloutWorker(
	args, data_buffer, concurrency=args.sglang_server_concurrency * get_rollout_num_engines(args)
	)

However, the worker applies this value to the number of active generate_and_rm_group tasks and the number of completed groups in output_queue which are measured in groups rather than samples.

slime/slime/rollout/fully_async_rollout.py

Lines 148 to 154 in 4c193f1

	# Top up. The qsize gate is the queue's backpressure: once a
	# full pool of completed groups is waiting, stop pulling new
	# prompts until the training side drains some.
	while (
	len(active_tasks) < max_concurrent and self.output_queue.qsize() < max_concurrent and self.running
	):
	groups = self.data_buffer.get_samples(1)

When n_samples_per_prompt > 1, this makes the worker's scheduling and backpressure window larger than intended.

For example, with a sample concurrency of 512 and 8 samples per prompt, the worker previously allowed up to 512 active groups instead of roughly 64 groups.

The inner semaphore still limits actual SGLang sample concurrency, so this does not exceed the server request concurrency limit. However, it can cause excessive task creation, prefetching, and completed-group buffering before backpressure takes effect.

Change

Derive the worker's group concurrency from the total sample concurrency:

group_concurrency = ceil(sample_concurrency / n_samples_per_prompt)

Rename the worker fields and variables to make their group-level semantics explicit.

This change only affects the outer fully-async scheduling and backpressure window. The existing per-sample SGLang semaphore remains unchanged.

Testing

• Added a unit test covering sample-to-group concurrency conversion.
• Updated existing fully-async backpressure tests to use group-level terminology.
• tests/test_fully_async_rollout.py: 5 passed.