Struct CoordinatedWorker 

pub struct CoordinatedWorker { /* private fields */ }

Leader’s view of a worker with coordination state.

Coordination State vs Execution State

CoordinatedWorker maintains coordination state - the leader’s view of what handles a worker has and how to route transfers. This is distinct from execution state which DirectWorker maintains for actual transfer execution.

State Type	Owner	Purpose
Execution	DirectWorker	Handles needed by TransferManager to execute
Coordination	CoordinatedWorker	Leader’s tracking for routing decisions

When the inner worker is a DirectWorker, handles exist in both places. This duplication is intentional:

• DirectWorker needs handles to call TransferManager
• CoordinatedWorker provides uniform API for local AND remote workers
• VeloWorkerClient is stateless, so leader must track handles somewhere

Usage

// Leader creates CoordinatedWorker wrapping actual worker
let worker = CoordinatedWorker::new(
    Box::new(direct_worker),
    rank,
    host_instance,
);

// After configure_layouts RPC, populate coordination state
worker.apply_layout_response(&response)?;

// Leader can now query handles for routing
if let Some(g2) = worker.local_g2() {
    // Route G2 transfers through this worker
}

Remote Handle Mappings

For cross-leader transfers (e.g., Prefill pulling from Decode), the leader imports remote worker metadata and stores rank-aware mappings:

// Prefill leader imports Decode workers' metadata
worker.import_remote_metadata(decode_leader_id, decode_rank, metadata).await?;

// Later, execute transfer using stored mapping
worker.transfer_from_remote(
    decode_leader_id,
    decode_rank,
    LogicalLayoutHandle::G2,  // source
    src_block_ids,
    LogicalLayoutHandle::G2,  // destination
    dst_block_ids,
    options,
)?;

Implementations

impl CoordinatedWorker

pub fn new( inner: Box<dyn Worker>, rank: usize, host_instance: InstanceId, ) -> Self

Create a new CoordinatedWorker wrapping an existing Worker.

pub fn rank(&self) -> usize

Get this worker’s rank.

pub fn host_instance(&self) -> InstanceId

Get the instance ID of the process hosting this worker.

pub fn inner(&self) -> &dyn Worker

Get a reference to the underlying Worker.

pub fn set_local_g1(&self, handle: LayoutHandle) -> Result<()>

Set the local G1 (GPU KV) handle.

Arguments

• handle - G1 layout handle

Errors

Returns error if G1 handle was already set.

pub fn set_local_g2(&self, handle: LayoutHandle) -> Result<()>

Set the local G2 (Host) handle.

Arguments

• handle - G2 layout handle

Errors

Returns error if G2 handle was already set.

pub fn set_local_g3(&self, handle: LayoutHandle) -> Result<()>

Set the local G3 (Disk) handle.

Arguments

• handle - G3 layout handle

Errors

Returns error if G3 handle was already set.

pub fn apply_layout_response( &self, response: &WorkerLayoutResponse, ) -> Result<()>

Apply layout response from configure_layouts RPC.

This is the primary way to populate coordination state. After the leader sends a configure_layouts RPC to the worker, the response contains the handles that were created. This method extracts those handles from the serialized metadata.

Arguments

• response - The WorkerLayoutResponse from configure_layouts RPC

Example

// Leader calls configure_layouts on worker
let response = worker_client.configure_layouts(config).await?;

// Populate coordination state from response
coordinated_worker.apply_layout_response(&response)?;

pub fn local_g1(&self) -> Option<LayoutHandle>

Get the local G1 handle if set.

pub fn local_g2(&self) -> Option<LayoutHandle>

Get the local G2 handle if set.

pub fn local_g3(&self) -> Option<LayoutHandle>

Get the local G3 handle if set.

pub async fn import_remote_metadata( &self, remote_leader_id: InstanceId, remote_rank: usize, metadata: SerializedLayout, ) -> Result<()>

Import metadata from a remote worker and store handle mappings.

This is called when the leader receives metadata from another leader’s workers during cross-leader coordination (e.g., prefill→decode).

Arguments

• remote_leader_id - Instance ID of the remote leader
• remote_rank - Rank of the remote worker under its leader
• metadata - Serialized layout metadata from the remote worker

pub fn resolve_remote_handle( &self, remote_leader_id: InstanceId, remote_rank: usize, logical_type: LogicalLayoutHandle, ) -> Option<LayoutHandle>

Look up physical handle for a remote transfer.

Arguments

• remote_leader_id - Instance ID of the remote leader
• remote_rank - Rank of the remote worker
• logical_type - Logical layout type (G1/G2/G3)

pub fn has_remote_metadata( &self, remote_leader_id: InstanceId, remote_rank: usize, ) -> bool

Check if remote metadata has been imported for a specific remote worker.

pub fn transfer_from_remote( &self, remote_leader_id: InstanceId, remote_rank: usize, src_logical: LogicalLayoutHandle, src_block_ids: Vec<BlockId>, dst_logical: LogicalLayoutHandle, dst_block_ids: Arc<[BlockId]>, options: TransferOptions, ) -> Result<TransferCompleteNotification>

Execute transfer from a remote worker.

This method looks up the remote handle from stored mappings and executes an RDMA transfer to pull data from the remote worker.

Arguments

• remote_leader_id - Instance ID of the remote leader
• remote_rank - Rank of the source worker under its leader
• src_logical - Source logical layout type (e.g., G2)
• src_block_ids - Block IDs on the remote to pull
• dst_logical - Destination logical layout type on this worker
• dst_block_ids - Destination block IDs
• options - Transfer options

Trait Implementations

impl ObjectBlockOps for CoordinatedWorker

fn has_blocks( &self, keys: Vec<SequenceHash>, ) -> BoxFuture<'static, Vec<(SequenceHash, Option<usize>)>>

Check if blocks exist in object storage.

fn put_blocks( &self, keys: Vec<SequenceHash>, src_layout: LogicalLayoutHandle, block_ids: Vec<BlockId>, ) -> BoxFuture<'static, Vec<Result<SequenceHash, SequenceHash>>>

Put blocks to object storage.

fn get_blocks( &self, keys: Vec<SequenceHash>, dst_layout: LogicalLayoutHandle, block_ids: Vec<BlockId>, ) -> BoxFuture<'static, Vec<Result<SequenceHash, SequenceHash>>>

Get blocks from object storage.

fn put_blocks_with_layout( &self, keys: Vec<SequenceHash>, _layout: PhysicalLayout, _block_ids: Vec<BlockId>, ) -> BoxFuture<'static, Vec<Result<SequenceHash, SequenceHash>>>

Put blocks to object storage using a resolved physical layout.

fn get_blocks_with_layout( &self, keys: Vec<SequenceHash>, _layout: PhysicalLayout, _block_ids: Vec<BlockId>, ) -> BoxFuture<'static, Vec<Result<SequenceHash, SequenceHash>>>

Get blocks from object storage into a resolved physical layout.