ucxx::python::Worker Class Reference

Specialized Python implementation of a ucxx::Worker. More...

#include <worker.h>

Inheritance diagram for ucxx::python::Worker:

Public Member Functions
	Worker (const Worker &)=delete
Worker &	operator= (Worker const &)=delete
	Worker (Worker &&o)=delete
Worker &	operator= (Worker &&o)=delete
void	populateFuturesPool () override
	Populate the Python futures pool.
void	clearFuturesPool () override
	Clear the futures pool.
std::shared_ptr<::ucxx::Future >	getFuture () override
	Get a Python future from the pool.
RequestNotifierWaitState	waitRequestNotifier (uint64_t periodNs) override
	Block until a request event.
void	runRequestNotifier () override
	Notify Python futures of each completed communication request.
void	stopRequestNotifierThread () override
	Signal the notifier to terminate.
Public Member Functions inherited from ucxx::Worker
	Worker (const Worker &)=delete
Worker &	operator= (Worker const &)=delete
	Worker (Worker &&o)=delete
Worker &	operator= (Worker &&o)=delete
virtual	~Worker ()
	ucxx::Worker destructor.
ucp_worker_h	getHandle ()
	Get the underlying ucp_worker_h handle.
std::string	getInfo ()
	Get information about the underlying ucp_worker_h object.
void	initBlockingProgressMode ()
	Initialize blocking progress mode.
int	getEpollFileDescriptor ()
	Get the epoll file descriptor associated with the worker.
bool	arm ()
	Arm the UCP worker.
bool	progressWorkerEvent (const int epollTimeout=-1)
	Progress worker event while in blocking progress mode.
void	signal ()
	Signal the worker that an event happened.
bool	waitProgress ()
	Block until an event has happened, then progresses.
bool	progressOnce ()
	Progress the worker only once.
bool	progress ()
	Progress the worker until all communication events are completed.
void	registerDelayedSubmission (std::shared_ptr< Request > request, DelayedSubmissionCallbackType callback)
	Register delayed request submission.
bool	registerGenericPre (DelayedSubmissionCallbackType callback, uint64_t period=0)
	Register callback to be executed in progress thread before progressing.
bool	registerGenericPost (DelayedSubmissionCallbackType callback, uint64_t period=0)
	Register callback to be executed in progress thread before progressing.
bool	isDelayedRequestSubmissionEnabled () const
	Inquire if worker has been created with delayed submission enabled.
bool	isFutureEnabled () const
	Inquire if worker has been created with future support.
bool	isRequestAttributesEnabled () const noexcept
	Inquire if worker has been created with request attributes querying enabled.
BufferType	getCudaBufferType () const
	Get the preferred buffer type for CUDA allocations.
void	setProgressThreadStartCallback (std::function< void(void *)> callback, void *callbackArg)
	Set callback to be executed at the progress thread start.
void	startProgressThread (const bool pollingMode=false, const int epollTimeout=1)
	Start the progress thread.
void	stopProgressThread ()
	Stop the progress thread.
bool	isProgressThreadRunning ()
	Inquire if worker has a progress thread running.
std::thread::id	getProgressThreadId ()
	Get the progress thread ID.
size_t	cancelInflightRequests (uint64_t period=0, uint64_t maxAttempts=1)
	Cancel inflight requests.
void	scheduleRequestCancel (TrackedRequests trackedRequests)
	Schedule cancelation of inflight requests.
void	removeInflightRequest (std::shared_ptr< Request > request)
	Remove reference to request from internal container.
std::shared_ptr< TagProbeInfo >	tagProbe (const Tag tag, const TagMask tagMask=TagMaskFull, const bool remove=false) const
	Check for uncaught tag messages.
RequestTagBuilder	tagRecvBuilder (void *buffer, size_t length, Tag tag, TagMask tagMask)
	Create a builder for a tag receive operation.
RequestTagBuilder	tagRecvWithHandleBuilder (void *buffer, std::shared_ptr< TagProbeInfo > probeInfo)
	Create a builder for a tag receive operation using a message handle.
RequestTagBuilder	tagRecvWithHandleBuilder (void *buffer, size_t length, std::shared_ptr< TagProbeInfo > probeInfo)
	Create a builder for a tag receive operation using a message handle.
AddressBuilder	addressBuilder ()
	Create a builder for this worker's address.
EndpointBuilder	endpointBuilder (std::string ipAddress, uint16_t port)
	Create a builder for an endpoint to a worker listening on an IP and port.
EndpointBuilder	endpointBuilder (std::shared_ptr< Address > address)
	Create a builder for an endpoint to a worker located at a UCX address.
ListenerBuilder	listenerBuilder (uint16_t port, ucp_listener_conn_callback_t callback, void *callbackArgs)
	Create a builder for a listener on this worker.
void	registerAmAllocator (ucs_memory_type_t memoryType, AmAllocatorType allocator)
	Register allocator for active messages.
void	registerAmReceiverCallback (AmReceiverCallbackInfo info, AmReceiverCallbackType callback)
	Register receiver callback for active messages.
bool	amProbe (const ucp_ep_h endpointHandle) const
	Check for uncaught active messages.
RequestFlushBuilder	flushBuilder ()
	Create a builder for a flush operation.
Attributes	queryAttributes () const
	Get the worker's attributes.
Public Member Functions inherited from ucxx::Component
void	setParent (std::shared_ptr< Component > parent)
	Set the internal parent reference.
std::shared_ptr< Component >	getParent () const
	Get the internal parent reference.

Friends
std::shared_ptr<::ucxx::Worker >	createWorker (std::shared_ptr< Context > context, const bool enableDelayedSubmission, const bool enableFuture)
	Constructor of shared_ptr<ucxx::python::Worker>.

Additional Inherited Members
Protected Member Functions inherited from ucxx::Worker
	Worker (std::shared_ptr< Context > context, const bool enableDelayedSubmission=false, const bool enableFuture=false)
	Protected constructor of ucxx::Worker.
Protected Attributes inherited from ucxx::Worker
bool	_enableFuture
	Boolean identifying whether the worker was created with future capability.
bool	_enableRequestAttributes
	Whether request attributes (e.g. UCP debug info) are queried for each request.
std::mutex	_futuresPoolMutex {}
	Mutex to access the futures pool.
std::queue< std::shared_ptr< Future > >	_futuresPool {}
	Futures pool to prevent running out of fresh futures.
std::shared_ptr< Notifier >	_notifier {nullptr}
	Notifier object.
std::shared_ptr< internal::AmData >	_amData
	Worker data made available to Active Messages callback.
BufferType	_cudaBufferType {BufferType::Invalid}
	Preferred buffer type for CUDA allocations.
Protected Attributes inherited from ucxx::Component
std::shared_ptr< Component >	_parent {nullptr}
	A reference-counted pointer to the parent.

Detailed Description

Specialized Python implementation of a ucxx::Worker.

Specialized Python implementation of a ucxx::Worker, providing Python-specific functionality, such as notification of Python futures.

Member Function Documentation

clearFuturesPool()

void ucxx::python::Worker::clearFuturesPool ( )

overridevirtual

Clear the futures pool.

Clear the futures pool, ensuring all references are removed and thus avoiding reference cycles that prevent the ucxx::Worker and other resources from cleaning up on time.

This method is safe to be called even if object was created with enableFuture=false.

Reimplemented from ucxx::Worker.

getFuture()

std::shared_ptr<::ucxx::Future > ucxx::python::Worker::getFuture ( )

nodiscardoverridevirtual

Get a Python future from the pool.

Get a Python future from the pool. If the pool is empty, ucxx::python::Worker::populateFuturesPool() is called and a warning is raised, since that likely means the user is missing to call the aforementioned method regularly.

Returns

The shared_ptr<ucxx::python::Future> object

Reimplemented from ucxx::Worker.

populateFuturesPool()

void ucxx::python::Worker::populateFuturesPool ( )

overridevirtual

Populate the Python futures pool.

To avoid taking the Python GIL for every new future required by each ucxx::Request, the ucxx::python::Worker maintains a pool of futures that can be acquired when a new ucxx::Request is created. Currently the pool has a maximum size of 100 objects, and will refill once it goes under 50, otherwise calling this functions results in a no-op.

Exceptions

std::runtime_error

if object was created with enableFuture=false.

Reimplemented from ucxx::Worker.

runRequestNotifier()

void ucxx::python::Worker::runRequestNotifier ( )

overridevirtual

Notify Python futures of each completed communication request.

Notifies Python futures of each completed communication request of their new status. This method is intended to be used from the Python notifier thread, where the thread will call waitRequestNotifier() and block until some communication is completed, and then call this method to notify all futures. If this is notifying a Python future, the thread where this method is called from must be using the same Python event loop as the thread that submitted the transfer request.

Reimplemented from ucxx::Worker.

stopRequestNotifierThread()

void ucxx::python::Worker::stopRequestNotifierThread ( )

overridevirtual

Signal the notifier to terminate.

Signals the notifier to terminate, awakening the waitRequestNotifier() blocking call.

Reimplemented from ucxx::Worker.

waitRequestNotifier()

RequestNotifierWaitState ucxx::python::Worker::waitRequestNotifier ( uint64_t periodNs )

nodiscardoverridevirtual

Block until a request event.

Blocks until some communication is completed and a Python future is ready to be notified, shutdown was initiated or a timeout occurred (only if periodNs > 0). This method is intended for use from the Python notifier thread, where that thread will block until one of the aforementioned events occur.

Returns

RequestNotifierWaitState::Ready if some communication completed, RequestNotifierWaitStats::Timeout if a timeout occurred, or RequestNotifierWaitStats::Shutdown if shutdown has initiated.

Reimplemented from ucxx::Worker.

Friends And Related Symbol Documentation

createWorker

std::shared_ptr<::ucxx::Worker > createWorker ( std::shared_ptr< Context > context, const bool enableDelayedSubmission, const bool enableFuture )

friend

Constructor of shared_ptr<ucxx::python::Worker>.

The constructor for a shared_ptr<ucxx::python::Worker> object. The default constructor is made private to ensure all UCXX objects are shared pointers for correct lifetime management.

// context is `std::shared_ptr<ucxx::Context>`
auto worker = ucxx::python::createWorker(context, false, false);

Returns

The shared_ptr<ucxx::python::Worker> object

---

The documentation for this class was generated from the following file:

• /__w/ucxx/ucxx/cpp/python/include/ucxx/python/worker.h