vllm.distributed.kv_transfer.kv_connector.v1.umbp.runtime.base

Backend-neutral runtime contract for UMBP.

Classes:

• IUMBPRuntime –
• UMBPSchedulerHandle –
• UMBPWorkerHandle –

IUMBPRuntime

Bases: Protocol

Methods:

• create_scheduler_handle –

  Create a scheduler-side handle.

• create_worker_handle –

  Create a worker-side handle.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

class IUMBPRuntime(Protocol):
    def create_scheduler_handle(
        self,
        namespace: str,
        topology: RankTopology,
        layout: KVLayoutDescriptor,
    ) -> UMBPSchedulerHandle:
        """Create a scheduler-side handle."""

    def create_worker_handle(
        self,
        namespace: str,
        topology: RankTopology,
        layout: KVLayoutDescriptor,
    ) -> UMBPWorkerHandle:
        """Create a worker-side handle."""

create_scheduler_handle(namespace, topology, layout)

Create a scheduler-side handle.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def create_scheduler_handle(
    self,
    namespace: str,
    topology: RankTopology,
    layout: KVLayoutDescriptor,
) -> UMBPSchedulerHandle:
    """Create a scheduler-side handle."""

create_worker_handle(namespace, topology, layout)

Create a worker-side handle.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def create_worker_handle(
    self,
    namespace: str,
    topology: RankTopology,
    layout: KVLayoutDescriptor,
) -> UMBPWorkerHandle:
    """Create a worker-side handle."""

UMBPSchedulerHandle

Bases: Protocol

Methods:

• clear –

  Clear all objects visible to this scheduler handle.

• close –

  Release scheduler-side resources.

• lookup –

  Return authoritative existence for each key.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

class UMBPSchedulerHandle(Protocol):
    def lookup(self, keys: Sequence[str]) -> Sequence[bool]:
        """Return authoritative existence for each key."""

    def clear(self) -> bool:
        """Clear all objects visible to this scheduler handle."""

    def close(self) -> None:
        """Release scheduler-side resources."""

clear()

Clear all objects visible to this scheduler handle.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def clear(self) -> bool:
    """Clear all objects visible to this scheduler handle."""

close()

Release scheduler-side resources.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def close(self) -> None:
    """Release scheduler-side resources."""

lookup(keys)

Return authoritative existence for each key.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def lookup(self, keys: Sequence[str]) -> Sequence[bool]:
    """Return authoritative existence for each key."""

UMBPWorkerHandle

Bases: Protocol

Methods:

• close –

  Release worker-side resources.

• load –

  Start loading the requested plans.

• poll –

  Return a finished job without blocking, or None if still pending.

• publish –

  Accept a completed store; a runtime may defer visibility until then.

• register_buffers –

  Register each unique KV allocation with the runtime.

• store –

  Start storing the requested plans.

• wait –

  Block until the job is final and its buffers are safe to reuse.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

class UMBPWorkerHandle(Protocol):
    def register_buffers(self, kv_caches: dict[str, torch.Tensor]) -> None:
        """Register each unique KV allocation with the runtime."""

    def load(self, plans: Sequence[BlockTransferPlan]) -> TransferJobState:
        """Start loading the requested plans."""

    def store(self, plans: Sequence[BlockTransferPlan]) -> TransferJobState:
        """Start storing the requested plans."""

    def wait(self, job: TransferJobState) -> TransferJobState:
        """Block until the job is final and its buffers are safe to reuse."""

    def poll(self, job: TransferJobState) -> TransferJobState | None:
        """Return a finished job without blocking, or None if still pending."""

    def publish(self, job: TransferJobState) -> None:
        """Accept a completed store; a runtime may defer visibility until then."""

    def close(self) -> None:
        """Release worker-side resources."""

close()

Release worker-side resources.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def close(self) -> None:
    """Release worker-side resources."""

load(plans)

Start loading the requested plans.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def load(self, plans: Sequence[BlockTransferPlan]) -> TransferJobState:
    """Start loading the requested plans."""

poll(job)

Return a finished job without blocking, or None if still pending.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def poll(self, job: TransferJobState) -> TransferJobState | None:
    """Return a finished job without blocking, or None if still pending."""

publish(job)

Accept a completed store; a runtime may defer visibility until then.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def publish(self, job: TransferJobState) -> None:
    """Accept a completed store; a runtime may defer visibility until then."""

register_buffers(kv_caches)

Register each unique KV allocation with the runtime.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def register_buffers(self, kv_caches: dict[str, torch.Tensor]) -> None:
    """Register each unique KV allocation with the runtime."""

store(plans)

Start storing the requested plans.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def store(self, plans: Sequence[BlockTransferPlan]) -> TransferJobState:
    """Start storing the requested plans."""

wait(job)

Block until the job is final and its buffers are safe to reuse.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/umbp/runtime/base.py

def wait(self, job: TransferJobState) -> TransferJobState:
    """Block until the job is final and its buffers are safe to reuse."""