Add async checkpoint feature

[VincentCheungKokomo]

• feat(checkpoint): add async DCP checkpoint saving

1. Use PyTorch DCP async_save to offload checkpoint file writes to a background process, allowing training to continue while checkpoint I/O is in progress.
2. Create a dedicated gloo process group for async checkpoint collectives to avoid interfering with the training communication group.
3. Write async checkpoints to an .incomplete directory first, then rename after all ranks finish, so partially written checkpoints are never exposed as valid checkpoints.
4. Keep at most one async checkpoint in flight and wait for pending saves before the next checkpoint or trainer shutdown.
5. Reuse a shared-memory staging cache in XtunerCacheWriter to reduce CPU buffer allocation and process handoff overhead between checkpoints.
6. Add optional file write lock slots to throttle concurrent DCP file writes and reduce storage contention during async checkpointing.
7. Store model and optimizer DCP state under a single weights directory and update trainer, RL worker, resume logic, and tests for the merged layout.

[HAOCHENYE]

In [2]: fw = FileSystemWriter("./")

In [3]: from torch.distributed.checkpoint.staging import AsyncStager, BlockingAsyncStager

In [4]: isinstance(fw, AsyncStager)
Out[4]: True

is _CachingStagingWriter necessary?

[HAOCHENYE]

Is this injection necessary?

[VincentCheungKokomo]

cache_staged_state_dict keeps pinned staging buffers on the FileSystemWriter instance. XTuner creates one writer per checkpoint path, so carry the cache across writers to preserve steady-state async_save launch performance.

[HAOCHENYE]

Could we hold the storage writer all the time if async_save is enabled?

[HAOCHENYE]

@hhaAndroid Please evaluate whether we actually have a need to not store the optimizer

[HAOCHENYE]

Where is the purpose of using a committed future + dcp future here? I understand that committed simply involves recording the time consumed, printing some logs, and finally moving the path, which are all lightweight operations. What is the purpose of such an implementation?