Delayed Submission

Delayed Submission#

2026-03-31

12 min read time

Applies to Linux

The delayed submission module provides a mechanism to defer the submission of UCXX requests to the worker progress thread, rather than submitting them immediately from the calling thread. This is useful for thread-safety when multiple threads submit requests concurrently.

#include <ucxx/delayed_submission.h>

namespace ucxx

namespace ucxx#

Typedefs

typedef std::function<void()> DelayedSubmissionCallbackType#

A user-defined function to execute as part of delayed submission callback.

A user-defined function to execute in the scope of a ucxx::DelayedSubmission, allowing execution of custom code upon the completion of the delayed submission.

typedef uint64_t ItemIdType#

Type for identifying items in DelayedSubmissionCollection.

A 64-bit unsigned integer used to uniquely identify items in DelayedSubmissionCollection, allowing for tracking and cancellation of specific items.

template<typename T>
class BaseDelayedSubmissionCollection#
#include <delayed_submission.h>

Base type for a collection of delayed submissions.

Base type for a collection of delayed submission. Delayed submissions may have different purposes and this class encapsulates generic data for all derived types.

Public Functions

inline explicit BaseDelayedSubmissionCollection(
std::string name,
const bool enabled
)#

Constructor for a thread-safe delayed submission collection.

Construct a thread-safe delayed submission collection. A delayed submission collection provides two operations: schedule and process. The schedule() method will push an operation into the collection, whereas the process() will invoke all callbacks that were previously pushed into the collection and clear the collection.

Parameters:

  • name – [in] human-readable name of the collection, used for logging.

  • enabled – [in] whether the resource is enabled, if false an exception is raised when attempting to schedule a callable. Disabled instances of this class should only be used to provide a consistent interface among implementations.

BaseDelayedSubmissionCollection(
) = delete#
BaseDelayedSubmissionCollection(
const BaseDelayedSubmissionCollection&
) = delete#
BaseDelayedSubmissionCollection &operator=(
BaseDelayedSubmissionCollection const&
) = delete#
BaseDelayedSubmissionCollection(
BaseDelayedSubmissionCollection &&o
) = delete#
BaseDelayedSubmissionCollection &operator=(
BaseDelayedSubmissionCollection &&o
) = delete#
inline virtual ItemIdType schedule(
T item
)#

Register a callable or complex-type for delayed submission.

Register a simple callback, or complex-type with a callback (requires specialization), for delayed submission that will be executed when the request is in fact submitted when process() is called.

Raise an exception if false was specified as the enabled argument to the constructor.

Throws:

std::runtime_error – if _enabled is false.

Parameters:

item – [in] the callback that will be executed by process() when the operation is submitted.

Returns:

the ID of the scheduled item which can be used cancelation requests.

inline void process(
)#

Process all pending callbacks.

Process all pending generic. Generic callbacks are deemed completed when their execution completes.

inline void cancel(
ItemIdType id
)#

Cancel a pending callback.

Cancel a pending callback and thus do not execute it, unless the execution has already begun, in which case cancelation cannot be done.

Throws:

std::runtime_error – if the item is being processed and canceling is not possible anymore.

Parameters:

id – [in] the ID of the scheduled item, as returned by schedule().

Protected Functions

virtual void scheduleLog(
ItemIdType id,
T item
) = 0#

Log message during schedule().

Log a specialized message while schedule() is being executed.

Parameters:

  • id – [in] the ID of the scheduled item, as returned by schedule().

  • item – [in] the callback that was passed as argument to schedule().

virtual void processItem(
ItemIdType id,
T item
) = 0#

Process a single item during process().

Method called by process() to process a single item of the collection.

Parameters:

  • id – [in] the ID of the scheduled item, as returned by schedule().

  • item – [in] the callback that was passed as argument to schedule() when the first registered.

Protected Attributes

std::string _name = {"undefined"}#

The human-readable name of the collection, used for logging.

bool _enabled = {true}#

Whether the resource required to process the collection is enabled.

ItemIdType _itemId = {0}#

The item ID counter, used to allow cancelation.

std::optional<ItemIdType> _processing{std::nullopt}#

The ID of the item being processed, if any.

std::deque<std::pair<ItemIdType, T>> _collection = {}#

The collection.

std::set<ItemIdType> _canceled = {}#

IDs of canceled items.

std::mutex _mutex = {}#

Mutex to provide access to _collection.

class RequestDelayedSubmissionCollection : public ucxx::BaseDelayedSubmissionCollection<std::pair<std::shared_ptr<Request>, DelayedSubmissionCallbackType>>#
#include <delayed_submission.h>

A collection of delayed request submissions.

A collection of delayed submissions used specifically for message transfer ucxx::Request submissions.

Public Functions

explicit RequestDelayedSubmissionCollection(
std::string name,
const bool enabled
)#

Constructor of a collection of delayed request submissions.

Construct a collection of delayed submissions used specifically for message transfer ucxx::Request submissions.

Parameters:

  • name – [in] the human-readable name of the type of delayed submission for debugging purposes.

  • enabled – [in] whether delayed request submissions should be enabled.

Protected Functions

virtual void scheduleLog(
ItemIdType id,
std::pair<std::shared_ptr<Request>, DelayedSubmissionCallbackType> item
) override#

Log message during schedule().

Log a specialized message while schedule() is being executed.

Parameters:

  • id – [in] the ID of the scheduled item, as returned by schedule().

  • item – [in] the callback that was passed as argument to schedule().

virtual void processItem(
ItemIdType id,
std::pair<std::shared_ptr<Request>, DelayedSubmissionCallbackType> item
) override#

Process a single item during process().

Method called by process() to process a single item of the collection.

Parameters:

  • id – [in] the ID of the scheduled item, as returned by schedule().

  • item – [in] the callback that was passed as argument to schedule() when the first registered.

class GenericDelayedSubmissionCollection : public ucxx::BaseDelayedSubmissionCollection<DelayedSubmissionCallbackType>#
#include <delayed_submission.h>

A collection of delayed submissions of generic callbacks.

A collection of delayed submissions used specifically for execution of generic callbacks at pre-defined stages of the progress loop.

Public Functions

explicit GenericDelayedSubmissionCollection(
std::string name
)#

Constructor of a collection of delayed submissions of generic callbacks.

Construct a collection of delayed submissions used specifically for execution of generic callbacks at pre-defined stages of the progress loop.

Parameters:

name – [in] the human-readable name of the type of delayed submission for debugging purposes.

Protected Functions

virtual void scheduleLog(
ItemIdType id,
DelayedSubmissionCallbackType item
) override#

Log message during schedule().

Log a specialized message while schedule() is being executed.

Parameters:

  • id – [in] the ID of the scheduled item, as returned by schedule().

  • item – [in] the callback that was passed as argument to schedule().

virtual void processItem(
ItemIdType id,
DelayedSubmissionCallbackType callback
) override#

Process a single item during process().

Method called by process() to process a single item of the collection.

Parameters:

  • id – [in] the ID of the scheduled item, as returned by schedule().

  • item – [in] the callback that was passed as argument to schedule() when the first registered.

class DelayedSubmissionCollection#
#include <delayed_submission.h>

A collection of delayed submissions of multiple types.

A collection of delayed submissions of multiple types used by the owner to manage each of the delayed submission types via specialized methods.

Public Functions

explicit DelayedSubmissionCollection(
bool enableDelayedRequestSubmission = false
)#

Default delayed submission collection constructor.

Construct an empty collection of delayed submissions. Despite its name, a delayed submission registration may be processed right after registration, thus effectively making it an immediate submission.

Parameters:

enableDelayedRequestSubmission – [in] whether request submission should be enabled, if false, only generic callbacks are enabled.

DelayedSubmissionCollection(
) = delete#
DelayedSubmissionCollection(
const DelayedSubmissionCollection&
) = delete#
DelayedSubmissionCollection &operator=(
DelayedSubmissionCollection const&
) = delete#
DelayedSubmissionCollection(
DelayedSubmissionCollection &&o
) = delete#
DelayedSubmissionCollection &operator=(
DelayedSubmissionCollection &&o
) = delete#
void processPre(
)#

Process pending delayed request submission and generic-pre callback operations.

Process all pending delayed request submissions and generic callbacks. Generic callbacks are deemed completed when their execution completes. On the other hand, the execution of the delayed request submission callbacks does not imply completion of the operation, only that it has been submitted. The completion of each delayed request submission is handled externally by the implementation of the object being processed, for example by checking the result of ucxx::Request::isCompleted().

Generic callbacks may be used to to pass information between threads on the subject that requests have been in fact processed, therefore, requests are processed first, then generic callbacks are.

void processPost(
)#

Process all pending generic-post callback operations.

Process all pending generic-post callbacks. Generic callbacks are deemed completed when their execution completes.

void registerRequest(
std::shared_ptr<Request> request,
DelayedSubmissionCallbackType callback
)#

Register a request for delayed submission.

Register a request for delayed submission with a callback that will be executed when the request is in fact submitted when processPre() is called.

Throws:

std::runtime_error – if delayed request submission was disabled at construction.

Parameters:

  • request – [in] the request to which the callback belongs, ensuring it remains alive until the callback is invoked.

  • callback – [in] the callback that will be executed by processPre() when the operation is submitted.

ItemIdType registerGenericPre(
DelayedSubmissionCallbackType callback
)#

Register a generic callback to execute during processPre().

Register a generic callback that will be executed when processPre() is called. Lifetime of the callback must be ensured by the caller.

Parameters:

callback – [in] the callback that will be executed by processPre().

Returns:

the ID of the scheduled item which can be used cancelation requests.

ItemIdType registerGenericPost(
DelayedSubmissionCallbackType callback
)#

Register a generic callback to execute during processPost().

Register a generic callback that will be executed when processPost() is called. Lifetime of the callback must be ensured by the caller.

Parameters:

callback – [in] the callback that will be executed by processPre().

Returns:

the ID of the scheduled item which can be used cancelation requests.

void cancelGenericPre(
ItemIdType id
)#

Cancel a generic callback scheduled for processPre() execution.

Cancel the execution of a generic callback that has been previously scheduled for execution during processPre(). This can be useful if the caller of registerGenericPre() has given up and will not anymore be able to guarantee the lifetime of the callback.

Parameters:

id – [in] the ID of the scheduled item, as returned by registerGenericPre().

void cancelGenericPost(
ItemIdType id
)#

Cancel a generic callback scheduled for processPost() execution.

Cancel the execution of a generic callback that has been previously scheduled for execution during processPos(). This can be useful if the caller of registerGenericPre() has given up and will not anymore be able to guarantee the lifetime of the callback.

Parameters:

id – [in] the ID of the scheduled item, as returned by registerGenericPos().

bool isDelayedRequestSubmissionEnabled(
) const#

Inquire if delayed request submission is enabled.

Check whether delayed submission request is enabled, in which case registerRequest() may be used to register requests that will be executed during processPre().

Returns:

true if a delayed request submission is enabled, false otherwise.

Private Members

GenericDelayedSubmissionCollection _genericPre{"generic pre"}#

The collection of all known generic pre-progress operations.

GenericDelayedSubmissionCollection _genericPost{"generic post"}#

The collection of all known generic post-progress operations.

RequestDelayedSubmissionCollection _requests{"request", false}#

The collection of all known delayed request submission operations.

bool _enableDelayedRequestSubmission = {false}#

Inflight Requests#

#include <ucxx/inflight_requests.h>

namespace ucxx

class InflightRequests#

Handle tracked requests.

Handle tracked requests, providing functionality so that its owner can modify those requests, performing operations such as insertion, removal and cancelation.

Public Functions

InflightRequests(
) = default#

Default constructor.

~InflightRequests(
)#

Destructor.

Cancels all inflight requests before destruction.

size_t size(
)#

Query the number of pending inflight requests.

Returns:

The number of pending inflight requests.

void insert(
std::shared_ptr<Request> request
)#

Insert an inflight requests to the container.

Parameters:

request – [in] a std::shared_ptr<Request> with the inflight request.

void merge(
TrackedRequestsPtr trackedRequests
)#

Merge containers of inflight requests with the internal containers.

Merge containers of inflight requests obtained from InflightRequests::release() of another object with the internal containers.

Parameters:

trackedRequests – [in] containers of tracked inflight requests to merge with the internal tracked inflight requests.

void remove(
const Request *const request
)#

Remove an inflight request from the internal container.

Remove the reference to a specific request from the internal container. This should be called when a request has completed and the InflightRequests owner does not need to keep track of it anymore. The raw pointer to a ucxx::Request is passed here as opposed to the usual std::shared_ptr<ucxx::Request> used elsewhere, this is because the raw pointer address is used as key to the requests reference, and this is called called from the object’s destructor.

Parameters:

request – [in] raw pointer to the request

size_t cancelAll(
)#

Issue cancelation of all inflight requests and clear the internal container.

Issue cancelation of all inflight requests known to this object and clear the internal container. The total number of canceled requests is returned.

Returns:

The total number of canceled requests.

TrackedRequestsPtr release(
)#

Releases the internally-tracked containers.

Releases the internally-tracked containers that can be merged into another InflightRequests object with InflightRequests::merge(). Effectively leaves the internal state as a clean, new object.

Returns:

The internally-tracked containers.

size_t getCancelingSize(
)#

Get count of requests in process of cancelation.

After cancelAll() is called the requests are scheduled for cancelation but may not complete immediately, therefore they are tracked until cancelation is completed. This method returns the count of requests in process of cancelation and drops references to those that have completed.

Returns:

The count of requests that are in process of cancelation.