Async

2026-03-31

12 min read time

Applies to Linux

This page provides ucxx class references for the async communication API in the ucxx._lib_async module. This module provides asyncio-compatible classes for building high-level communication applications.

ApplicationContext

class ucxx._lib_async.ApplicationContext(config_dict={}, progress_mode=None, enable_delayed_submission=None, enable_python_future=None, connect_timeout=None)

The context of the Asyncio interface of UCX.

clear_progress_tasks()

Return type:

None

property config

UCX configuration options as a dict.

continuous_ucx_progress(event_loop=None)

Guarantees continuous UCX progress

Use this function to associate UCX progress with an event loop. Notice, multiple event loops can be associate with UCX progress.

This function is automatically called when calling create_listener() or create_endpoint().

event_loop: asyncio.event_loop, optional

The event loop to evoke UCX progress. If None, asyncio.get_event_loop() (asyncio.new_event_loop() in Python 3.10+) is used.

create_endpoint(ip_address, port, endpoint_error_handling=True, connect_timeout=5.0)

Create a new endpoint to a server

ip_address: str

IP address of the server the endpoint should connect to

port: int

IP address of the server the endpoint should connect to

endpoint_error_handling: boolean, optional

If True (default) enable endpoint error handling raising exceptions when an error occurs, may incur in performance penalties but prevents a process from terminating unexpectedly that may happen when disabled. If False endpoint endpoint error handling is disabled.

connect_timeout: float

Timeout in seconds for exchanging peer info. In some cases, exchanging peer information may hang indefinitely, a timeout prevents that. If the chosen value is too high it may cause the operation to be stuck for too long rather than quickly raising a TimeoutError that may be recovered from by the application, but under high-load a higher timeout may be helpful to prevent exchanging peer info from failing too fast.

Endpoint

The new endpoint

create_endpoint_from_worker_address(address, endpoint_error_handling=True)

Create a new endpoint to a server

address: UCXAddress endpoint_error_handling: boolean, optional

If True (default) enable endpoint error handling raising exceptions when an error occurs, may incur in performance penalties but prevents a process from terminating unexpectedly that may happen when disabled. If False endpoint endpoint error handling is disabled.

Endpoint

The new endpoint

create_listener(callback_func, port=0, endpoint_error_handling=True, connect_timeout=5.0)

Create and start a listener to accept incoming connections

callback_func is the function or coroutine that takes one argument – the Endpoint connected to the client.

Notice, the listening is closed when the returned Listener goes out of scope thus remember to keep a reference to the object.

callback_func: function or coroutine

A callback function that gets invoked when an incoming connection is accepted

port: int, optional

An unused port number for listening, or 0 to let UCX assign an unused port.

endpoint_error_handling: boolean, optional

If True (default) enable endpoint error handling raising exceptions when an error occurs, may incur in performance penalties but prevents a process from terminating unexpectedly that may happen when disabled. If False endpoint endpoint error handling is disabled.

connect_timeout: float

Timeout in seconds for exchanging peer info. In some cases, exchanging peer information may hang indefinitely, a timeout prevents that. If the chosen value is too high it may cause the operation to be stuck for too long rather than quickly raising a TimeoutError that may be recovered from by the application, but under high-load a higher timeout may be helpful to prevent exchanging peer info from failing too fast.

Listener

The new listener. When this object is deleted, the listening stops

enable_delayed_submission = Ellipsis

enable_python_future = Ellipsis

get_config()

Returns all UCX configuration options as a dict.

dict

The current UCX configuration options

get_ucp_worker()

Returns the underlying UCP worker handle (ucp_worker_h)

as a Python integer.

get_ucxx_worker()

Returns the underlying UCXX worker pointer (ucxx::Worker*)

as a Python integer.

get_worker_address()

progress_mode = Ellipsis

recv(buffer, tag)

Receive directly on worker without a local Endpoint into buffer.

buffer: exposing the buffer protocol or array/cuda interface

The buffer to receive into. Raise ValueError if buffer is smaller than nbytes or read-only.

tag: hashable, optional

Set a tag that must match the received message.

start_notifier_thread()

stop_notifier_thread()

Stop Python future notifier thread

Stop the notifier thread if context is running with Python future notification enabled via UCXPY_ENABLE_PYTHON_FUTURE=1 or ucxx.init(…, enable_python_future=True).

Warning

When the notifier thread is enabled it may be necessary to explicitly call this method before shutting down the process or or application, otherwise it may block indefinitely waiting for the thread to terminate. Executing ucxx.reset() will also run this method, so it’s not necessary to have both.

property ucp_context_info

Low-level UCX info about this endpoint as a string.

property ucp_worker

The underlying UCP worker handle (ucp_worker_h) as a Python integer.

property ucp_worker_info

Return low-level UCX info about this endpoint as a string.

property ucxx_worker

The underlying UCXX worker pointer (ucxx::Worker*) as a Python integer.

property worker_address

Endpoint

class ucxx._lib_async.Endpoint(endpoint, ctx, tags=None)

An endpoint represents a connection to a peer

Please use create_listener() and create_endpoint() to create an Endpoint.

abort(period=10000000000, max_attempts=1)

Close the communication immediately and abruptly.

Useful in destructors or generators’ finally blocks.

Despite the attempt to close communication immediately, in some circumstances, notably when the parent worker is running a progress thread, a maximum timeout may be specified for which the close operation will wait. This can be particularly important for cases where the progress thread might be attempting to acquire the GIL while the current thread owns that resource.

Notice, this functions doesn’t signal the connected peer to close. To do that, use Endpoint.close().

period: int

maximum period to wait (in ns) for internal endpoint operations to complete, usually two operations (pre and post) are involved thus the maximum perceived timeout should be multiplied by two.

max_attempts: int

maximum number of attempts to close endpoint, only applicable if worker is running a progress thread and period > 0.

property alive

am_recv()

Receive from connected peer via active messages.

am_send(buffer)

Send buffer to connected peer via active messages.

buffer: exposing the buffer protocol or array/cuda interface

The buffer to send. Raise ValueError if buffer is smaller than nbytes.

close(period=10000000000, max_attempts=1)

Close the endpoint cleanly.

This will attempt to flush outgoing buffers before actually closing the underlying UCX endpoint.

A maximum timeout and number of attempts may be specified to prevent the underlying Endpoint object from failing to acquire the GIL, see abort() for details.

period: int

maximum period to wait (in ns) for internal endpoint operations to complete, usually two operations (pre and post) are involved thus the maximum perceived timeout should be multiplied by two.

max_attempts: int

maximum number of attempts to close endpoint, only applicable if worker is running a progress thread and period > 0.

close_after_n_recv(n, count_from_ep_creation=False)

Close the endpoint after n received messages.

n: int

Number of messages to received before closing the endpoint.

count_from_ep_creation: bool, optional

Whether to count n from this function call (default) or from the creation of the endpoint.

property closed

Is this endpoint closed?

get_ucp_endpoint()

Returns the underlying UCP endpoint handle (ucp_ep_h)

as a Python integer.

get_ucp_worker()

Returns the underlying UCP worker handle (ucp_worker_h)

as a Python integer.

get_ucxx_endpoint()

Returns the underlying UCXX endpoint pointer (ucxx::Endpoint*)

as a Python integer.

get_ucxx_worker()

Returns the underlying UCXX worker pointer (ucxx::Worker*)

as a Python integer.

is_alive()

recv(buffer, tag=None, force_tag=False)

Receive from connected peer into buffer.

buffer: exposing the buffer protocol or array/cuda interface

The buffer to receive into. Raise ValueError if buffer is smaller than nbytes or read-only.

tag: hashable, optional

Set a tag that must match the received message. Currently the tag is hashed together with the internal Endpoint tag that is agreed with the remote end at connection time. To enforce using the user tag, make sure to specify force_tag=True.

force_tag: bool

If true, force using tag as is, otherwise the value specified with tag (if any) will be hashed with the internal Endpoint tag.

recv_multi(tag=None, force_tag=False)

Receive from connected peer into buffer.

tag: hashable, optional

Set a tag that must match the received message. Currently the tag is hashed together with the internal Endpoint tag that is agreed with the remote end at connection time. To enforce using the user tag, make sure to specify force_tag=True.

force_tag: bool

If true, force using tag as is, otherwise the value specified with tag (if any) will be hashed with the internal Endpoint tag.

recv_obj(tag=None, allocator=bytearray)

Receive from connected peer that calls send_obj().

As opposed to recv(), this function returns the received object. Data is received into a buffer allocated by allocator.

The transfer includes an extra message containing the size of obj, which increases the overhead slightly.

tag: hashable, optional

Set a tag that must match the received message. Notice, currently UCX-Py doesn’t support a “any tag” thus tag=None only matches a send that also sets tag=None.

allocator: callabale, optional

Function to allocate the received object. The function should take the number of bytes to allocate as input and return a new buffer of that size as output.

>>> await pickle.loads(ep.recv_obj())

send(buffer, tag=None, force_tag=False)

Send buffer to connected peer.

buffer: exposing the buffer protocol or array/cuda interface

The buffer to send. Raise ValueError if buffer is smaller than nbytes.

tag: hashable, optional

Set a tag that the receiver must match. Currently the tag is hashed together with the internal Endpoint tag that is agreed with the remote end at connection time. To enforce using the user tag, make sure to specify force_tag=True.

force_tag: bool

If true, force using tag as is, otherwise the value specified with tag (if any) will be hashed with the internal Endpoint tag.

send_multi(buffers, tag=None, force_tag=False)

Send buffer to connected peer.

buffer: exposing the buffer protocol or array/cuda interface

The buffer to send. Raise ValueError if buffer is smaller than nbytes.

tag: hashable, optional

Set a tag that the receiver must match. Currently the tag is hashed together with the internal Endpoint tag that is agreed with the remote end at connection time. To enforce using the user tag, make sure to specify force_tag=True.

force_tag: bool

If true, force using tag as is, otherwise the value specified with tag (if any) will be hashed with the internal Endpoint tag.

send_obj(obj, tag=None)

Send obj to connected peer that calls recv_obj().

The transfer includes an extra message containing the size of obj, which increases the overhead slightly.

obj: exposing the buffer protocol or array/cuda interface

The object to send.

tag: hashable, optional

Set a tag that the receiver must match.

>>> await ep.send_obj(pickle.dumps([1,2,3]))

set_close_callback(callback_func, cb_args=None, cb_kwargs=None)

Register a user callback function to be called on Endpoint’s closing.

Allows the user to register a callback function to be called when the Endpoint’s error callback is called, or during its finalizer if the error callback is never called.

Once the callback is called, it’s not possible to send any more messages. However, receiving messages may still be possible, as UCP may still have incoming messages in transit.

callback_func: callable

The callback function to be called when the Endpoint’s error callback is called, otherwise called on its finalizer.

cb_args: tuple or None

The arguments to be passed to the callback function as a tuple, or None (default).

cb_kwargs: dict or None

The keyword arguments to be passed to the callback function as a dict, or None (default).

Example >>> ep.set_close_callback(lambda: print(“Executing close callback”))

property ucp_endpoint

The underlying UCP endpoint handle (ucp_ep_h) as a Python integer.

property ucp_worker

The underlying UCP worker handle (ucp_worker_h) as a Python integer.

property ucxx_endpoint

The underlying UCXX endpoint pointer (ucxx::Endpoint*) as a Python integer.

property ucxx_worker

Returns the underlying UCXX worker pointer (ucxx::Worker*) as a Python integer.

property uid

The unique ID of the underlying UCX endpoint

Listener

class ucxx._lib_async.Listener(listener, ident, active_clients)

A handle to the listening service started by create_listener()

The listening continues as long as this object exist or .close() is called. Please use create_listener() to create an Listener.

property active_clients

close()

Closing the listener

property closed

Is the listener closed?

property ip

The listening network IP address

property port

The listening network port