[Optimization] Support logprob async copy

[Sunny-bot1]

Motivation

异步调度支持开启logprob

Modifications

LogprobsTensors支持异步拷贝

# 同步拷贝
indices = indices.cpu()
top_logprobs = top_logprobs.cpu()
token_ranks = token_ranks.cpu()

# 修改为异步拷贝
if current_platform.is_cuda():
    indices_cpu = paddle.empty_like(indices, device="cpu").pin_memory()
    top_logprobs_cpu = paddle.empty_like(top_logprobs, device="cpu").pin_memory()
    token_ranks_cpu = paddle.empty_like(token_ranks, device="cpu").pin_memory()
    indices_cpu.copy_(indices, False)
    top_logprobs_cpu.copy_(top_logprobs, False)
    token_ranks_cpu.copy_(token_ranks, False)
else:
    indices_cpu = indices.cpu()
    top_logprobs_cpu = top_logprobs.cpu()
    token_ranks_cpu = token_ranks.cpu()

Usage or Command

python -m fastdeploy.entrypoints.openai.api_server --model ${model_path} \
    --max-model-len 32768 \
    --max-num-seqs 128 \
    --port 8908 \
    --tensor-parallel-size 2 \
    --enable-logprob \
    --enable-overlap-schedule \

Accuracy Tests

Checklist

• Add at least a tag in the PR title.
  • Tag list: [[FDConfig],[APIServer],[Engine], [Scheduler], [PD Disaggregation], [Executor], [Graph Optimization], [Speculative Decoding], [RL], [Models], [Quantization], [Loader], [OP], [KVCache], [DataProcessor], [BugFix], [Docs], [CI], [Optimization], [Feature], [Benchmark], [Others], [XPU], [HPU], [GCU], [DCU], [Iluvatar], [Metax]]
  • You can add new tags based on the PR content, but the semantics must be clear.
• Format your code, run pre-commit before commit.
• Add unit tests. Please write the reason in this PR if no unit tests.
• Provide accuracy results.
• If the current PR is submitting to the release branch, make sure the PR has been submitted to the develop branch, then cherry-pick it to the release branch with the [Cherry-Pick] PR tag.

[paddle-bot]

Thanks for your contribution!

[codecov-commenter]

Codecov Report

❌ Patch coverage is 66.66667% with 4 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (develop@1c0a2b0). Learn more about missing BASE report.

Files with missing lines	Patch %	Lines
fastdeploy/model_executor/layers/sample/sampler.py	63.63%	3 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             develop    #6362   +/-   ##
==========================================
  Coverage           ?   68.22%           
==========================================
  Files              ?      391           
  Lines              ?    52248           
  Branches           ?     8146           
==========================================
  Hits               ?    35645           
  Misses             ?    13995           
  Partials           ?     2608           

Flag	Coverage Δ
GPU	68.22% <66.66%> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[Copilot]

Pull request overview

该 PR 旨在为开启 logprob 的场景引入（CUDA 下）更低开销的数据搬运方式，通过 pinned memory + 非阻塞 copy_ 尝试实现 logprobs 相关张量的异步 D2H 拷贝，并对 prompt_logprobs 的落盘/传递链路做了相应调整。

Changes:

• Sampler.gather_logprobs 在 CUDA 平台上使用 pinned CPU tensor，并以 copy_(..., blocking=False) 发起异步 D2H 拷贝。
• GPUModelRunner._get_prompt_logprobs_list 在使用异步拷贝结果前加入 paddle.device.synchronize()。
• save_output_topk 的 token_ids 入参从 sampler_output.sampled_token_ids 调整为 share_inputs["sampled_token_ids"]。

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.

File	Description
fastdeploy/worker/output.py	调整 LogprobsTensors.slice_rows 的实现（当前实现会强制 .cpu() 并重新 to_tensor）。
fastdeploy/worker/gpu_model_runner.py	prompt_logprobs 路径新增同步，确保异步 D2H copy 完成后再使用。
fastdeploy/model_executor/pre_and_post_process.py	save_output_topk 传入的 sampled token ids 来源调整为 share_inputs。
fastdeploy/model_executor/layers/sample/sampler.py	CUDA 下为 logprobs 相关张量引入 pinned memory + 非阻塞 D2H copy。

[Copilot]

LogprobsTensors.slice_rows 这里先对整块 Tensor 调用 .cpu() 再切片，并且又包了一层 paddle.to_tensor(...)，会导致不必要的数据拷贝/同步（在按 batch 循环 slice_rows 的场景下可能放大为多次整块拷贝），同时也会丢失上游可能使用的 pinned memory 优势。建议直接对原 Tensor 做切片并返回（保持原 place/pin 状态）；如果确实需要搬到 CPU，也应当先切片再 .cpu()，并避免对 Tensor 再次 to_tensor。

Suggested change

	paddle.to_tensor(self.logprob_token_ids.cpu()[start:end], place="cpu"),
	paddle.to_tensor(self.logprobs.cpu()[start:end], place="cpu"),
	paddle.to_tensor(self.selected_token_ranks.cpu()[start:end], place="cpu"),
	self.logprob_token_ids[start:end],
	self.logprobs[start:end],
	self.selected_token_ranks[start:end],

[Copilot]

这里使用 paddle.device.synchronize() 会进行全设备同步，会显著降低 overlap-schedule/异步拷贝的收益（尤其在循环里每个 request 都同步）。建议改为更细粒度的同步方式：例如使用 CUDA event 只等待本次 D2H copy 对应的 stream，或将 prompt_logprobs 的目标缓冲区改为 pinned memory 并直接对目标做非阻塞 copy，然后复用 runner 里已有的 event 同步机制。

Suggested change

	# Synchronize before using token_ids, logprobs and ranks to ensure async copy are completed.
	paddle.device.synchronize()
	# Synchronize current CUDA stream before using token_ids, logprobs and ranks
	paddle.device.cuda.current_stream().synchronize()

[Copilot]

注释里的英文语法建议调整："to ensure async copy are completed" 应改为 "to ensure async copies are completed" 或 "... is completed"，避免语法错误影响可读性。

Suggested change

	# Synchronize before using token_ids, logprobs and ranks to ensure async copy are completed.
	# Synchronize before using token_ids, logprobs and ranks to ensure async copies are completed.