vllm_omni.distributed.omni_connectors.transfer_adapter.base ¶

logger `module-attribute` ¶

logger = get_connector_logger(__name__)

OmniTransferAdapterBase ¶

Base class for managing data transfer via OmniConnector.

This class handles the core loop logic and connector interactions, but leaves the specific data processing (chunks, KV cache, etc.) to subclasses.

config `instance-attribute` ¶

config = config

connector `instance-attribute` ¶

connector = None

recv_thread `instance-attribute` ¶

recv_thread = threading.Thread(
    target=self.recv_loop, daemon=True
)

save_thread `instance-attribute` ¶

save_thread = threading.Thread(
    target=self.save_loop, daemon=True
)

stop_event `instance-attribute` ¶

stop_event = threading.Event()

collect_failed_receive_request_ids ¶

collect_failed_receive_request_ids() -> dict[str, str]

Drain and return requests whose received chunk was invalid.

collect_failed_send_request_ids ¶

collect_failed_send_request_ids() -> dict[str, str]

Drain and return the requests whose send gave up.

create_connector `classmethod` ¶

create_connector(model_config: Any)

get_finished_requests ¶

get_finished_requests()

Get finished loaded or saved requests

load ¶

load(*args, **kwargs)

Load request data from connector synchronously. To be implemented by subclasses.

load_async ¶

load_async(*args, **kwargs)

Register a request to load data. To be implemented by subclasses.

record_receive_failure ¶

record_receive_failure(
    request_id: str, reason: str
) -> None

Record an invalid received chunk that cannot be retried.

record_send_failure ¶

record_send_failure(
    request_id: str | None, reason: str
) -> None

Note that a chunk for request_id will never be delivered.

Called from the sender thread; drained by the scheduler thread, hence the lock. _waiting_since needs none because only the scheduler touches it.

recv_loop ¶

recv_loop()

Loop to poll for incoming data.

Process each pending request exactly once per pass. When no request made progress, back off 1 ms instead of tight-spinning on failed shm_open syscalls (which can burn a full CPU core).

save ¶

save(*args, **kwargs)

Save data to connector synchronously. To be implemented by subclasses.

save_async ¶

save_async(*args, **kwargs)

Submit data to be saved. To be implemented by subclasses.

save_loop ¶

save_loop()

Loop to send outgoing data.

shutdown ¶

shutdown()

Stop background loops and close the connector.