[ckpt] feat: add checkpoint-engine abstraction

[wuxibin89]

What does this PR do?

Add Checkpoint Engine abstraction.

Overview

Checkpoint Engine is an unified abstract layer to synchronize weights between various training backends and inference backends. It provides three unified APIs:

• send_weights: get named tensors from generator and send them in streaming manner.
• receive_weights: return a tensor generator that yield named tensors in streaming manner.
• get_weights: return a tensor generator that yield named tensors in streaming manner, used for each inference instance update weight independently from local cache (e.g share memory, disk).

For more detail, see verl/checkpoint_engine/README.md.

verl core

checkpoint engine

[gemini-code-assist]

Code Review

This pull request introduces a new Checkpoint Engine abstraction, which is a significant and well-structured addition for managing model weight synchronization. The implementations for NCCL and NIXL backends are comprehensive. However, I've identified several critical issues related to resource management (socket and context leaks) in both nccl_checkpoint_engine.py and nixl_checkpoint_engine.py, and improper process termination in nixl_checkpoint_engine.py. Additionally, there's a high-severity issue in base.py where a concrete implementation violates the async contract of its abstract base class. Addressing these issues is crucial for the stability and correctness of this new engine.

[gemini-code-assist]

The finish method does not clean up network resources like the ZMQ socket or the socket used for finding a free port. This will lead to resource leaks, which is critical for long-running services. The ZMQ context is also created locally in _start_zmq_server and _connect_zmq_client and not terminated. It should be created as an instance attribute in __init__ and terminated in finish using self.context.term().

Suggested change

	def finish(self):
	"""Destroy the NCCL process group if rebuild_group is True."""
	if self.rebuild_group:
	if self.rank >= 0:
	collective.destroy_collective_group(self.group_name)
	self.rank = None
	self.world_size = None
	self.send_buf = None
	self.recv_buf = None
	def finish(self):
	"""Destroy the NCCL process group if rebuild_group is True."""
	if self.rebuild_group:
	if self.rank >= 0:
	collective.destroy_collective_group(self.group_name)
	self.rank = None
	self.world_size = None
	self.send_buf = None
	self.recv_buf = None
	if hasattr(self, "socket"):
	self.socket.close()
	if hasattr(self, "listen_sock"):
	self.listen_sock.close()