dynamo.nixl_connect.Connector#

Core class for managing the connection between workers in a distributed environment. Use this class to create readable and writable operations, or read and write data to remote workers.

This class is responsible for interfacing with the NIXL-based RDMA subsystem and providing a “Pythonic” interface with which to utilize GPU Direct RDMA accelerated data transfers between models hosted by different workers in a Dynamo pipeline. The connector provides two methods of moving data between workers:

  • Preparing local memory to be written to by a remote worker.

  • Preparing local memory to be read by a remote worker.

In both cases, local memory is registered with the NIXL-based RDMA subsystem via the Descriptor class and provided to the connector. The connector then configures the RDMA subsystem to expose the memory for the requested operation and returns an operation control object. The operation control object, either a ReadableOperation or a WritableOperation, provides RDMA metadata (RdmaMetadata) via its .metadata() method, functionality to query the operation’s current state, as well as the ability to cancel the operation prior to its completion.

The RDMA metadata must be provided to the remote worker expected to complete the operation. The metadata contains required information (identifiers, keys, etc.) which enables the remote worker to interact with the provided memory.

Warning

RDMA metadata contains a worker’s address as well as security keys to access specific registered memory descriptors. This data provides direct memory access between workers, and should be considered sensitive and therefore handled accordingly.

Example Usage#

    @async_on_start
    async def async_init(self):
      runtime = dynamo_context["runtime"]

      self.connector = dynamo.nixl_connect.Connector(runtime=runtime)
      await self.connector.initialize()

Tip

See ReadOperation, ReadableOperation, WritableOperation, and WriteOperation for additional examples.

Methods#

begin_read#

async def begin_read(
    self,
    remote_metadata: RdmaMetadata,
    local_descriptors: Descriptor | list[Descriptor],
) -> ReadOperation:

Creates a ReadOperation for transferring data from a remote worker.

To create the operation, the serialized request from a remote worker’s ReadableOperation along with a matching set of local memory descriptors which reference memory intended to receive data from the remote worker must be provided. The serialized request must be transferred from the remote to the local worker via a secondary channel, most likely HTTP or TCP+NATS.

Once created, data transfer will begin immediately.

Disposal of the object will instruct the RDMA subsystem to cancel the operation, therefore the operation should be awaited until completed unless cancellation is intended.

Use .wait_for_completion() to block the caller until the operation has completed or encountered an error.

begin_write#

async def begin_write(
    self,
    local_descriptors: Descriptor | list[Descriptor],
    remote_metadata: RdmaMetadata,
) -> WriteOperation:

Creates a WriteOperation for transferring data to a remote worker.

To create the operation, the serialized request from a remote worker’s WritableOperation along with a matching set of local memory descriptors which reference memory to be transferred to the remote worker must be provided. The serialized request must be transferred from the remote to the local worker via a secondary channel, most likely HTTP or TCP+NATS.

Once created, data transfer will begin immediately.

Disposal of the object will instruct the RDMA subsystem to cancel the operation, therefore the operation should be awaited until completed unless cancellation is intended.

Use .wait_for_completion() to block the caller until the operation has completed or encountered an error.

create_readable#

def create_readable(
    self,
    local_descriptors: Descriptor | list[Descriptor],
) -> ReadableOperation:

Creates a ReadableOperation for transferring data to a remote worker.

To create the operation, a set of local memory descriptors must be provided that reference memory intended to be transferred to a remote worker. Once created, the memory referenced by the provided descriptors becomes immediately readable by a remote worker with the necessary metadata. The metadata required to access the memory referenced by the provided descriptors is accessible via the operation’s .metadata() method. Once acquired, the metadata needs to be provided to a remote worker via a secondary channel, most likely HTTP or TCP+NATS.

Disposal of the object will instruct the RDMA subsystem to cancel the operation, therefore the operation should be awaited until completed unless cancellation is intended.

Use .wait_for_completion() to block the caller until the operation has completed or encountered an error.

create_writable#

def create_writable(
    self,
    local_descriptors: Descriptor | list[Descriptor],
) -> WritableOperation:

Creates a WritableOperation for transferring data from a remote worker.

To create the operation, a set of local memory descriptors must be provided which reference memory intended to receive data from a remote worker. Once created, the memory referenced by the provided descriptors becomes immediately writable by a remote worker with the necessary metadata. The metadata required to access the memory referenced by the provided descriptors is accessible via the operation’s .metadata() method. Once acquired, the metadata needs to be provided to a remote worker via a secondary channel, most likely HTTP or TCP+NATS.

Disposal of the object will instruct the RDMA subsystem to cancel the operation, therefore the operation should be awaited until completed unless cancellation is intended.

Use .wait_for_completion() to block the caller until the operation has completed or encountered an error.

Properties#

is_cuda_available#

@cached_property
def is_cuda_available(self) -> bool:

Gets True when CUDA is available for the selected array module (most likely CuPy); otherwise False.

name#

@property
def name(self) -> str | None:

Gets the Dynamo component name used by the connector.

namespace#

@property
def namespace(self) -> str:

Gets the Dynamo namespace used by the connector.

runtime#

def runtime(self) -> dynamo.runtime.DistributedRuntime:

Gets the Dynamo distributed runtime instance associated with the connector.

On this page