Waiting for Single Messages in Flow Graph

Intel® oneAPI Threading Building Blocks Developer Guide and API Reference

Download PDF

ID 772616

Date 3/31/2025

Version

Public

A newer version of this document is available. Customers should click here to go to the newest version.

Visible to Intel only — GUID: GUID-A13E3A96-5170-4654-A2DB-89D30A076850

View Details

Waiting for Single Messages in Flow Graph

Description
API
Example

Description

This feature adds a new try_put_and_wait interface to the receiving nodes in the Flow Graph. This function puts a message as an input into a Flow Graph and waits until all work related to that message is complete. try_put_and_wait may reduce latency compared to calling graph::wait_for_all since graph::wait_for_all waits for all work, including work that is unrelated to the input message, to complete.

node.try_put_and_wait(msg) performs node.try_put(msg) on the node and waits until the work on msg is completed. Therefore, the following conditions are true:

Any task initiated by any node in the Flow Graph that involves working with msg or any other intermediate result computed from msg is completed.
No intermediate results computed from msg remain in any buffers in the graph.

CAUTION:

To prevent try_put_and_wait calls from infinite waiting, avoid using buffering nodes at the end of the Flow Graph since the final result will not be automatically consumed by the Flow Graph.

CAUTION:

The multifunction_node and async_node classes are not currently supported by this feature. Including one of these nodes in the Flow Graph may cause try_put_and_wait to exit early, even if the computations on the initial input message are still in progress.

API

Header


#define TBB_PREVIEW_FLOW_GRAPH_FEATURES // macro option 1
#define TBB_PREVIEW_FLOW_GRAPH_TRY_PUT_AND_WAIT // macro option 2
#include <oneapi/tbb/flow_graph.h>

Synopsis


namespace oneapi {
    namespace tbb {
        template <typename Output, typename Policy = /*default-policy*/>
        class continue_node {
        public:
            bool try_put_and_wait(const continue_msg& input);
        }; // class continue_node

        template <typename Input, typename Output = continue_msg, typename Policy = /*default-policy*/>
        class function_node {
        public:
            bool try_put_and_wait(const Input& input);
        }; // class function_node

        template <typename T>
        class overwrite_node {
        public:
            bool try_put_and_wait(const T& input);
        }; // class overwrite_node

        template <typename T>
        class write_once_node {
        public:
            bool try_put_and_wait(const T& input);
        }; // class write_once_node

        template <typename T>
        class buffer_node {
        public:
            bool try_put_and_wait(const T& input);
        }; // class buffer_node

        template <typename T>
        class queue_node {
        public:
            bool try_put_and_wait(const T& input);
        }; // class queue_node

        template <typename T, typename Compare = std::less<T>>
        class priority_queue_node {
        public:
            bool try_put_and_wait(const T& input);
        }; // class priority_queue_node

        template <typename T>
        class sequencer_node {
        public:
            bool try_put_and_wait(const T& input);
        }; // class sequencer_node

        template <typename T, typename DecrementType = continue_msg>
        class limiter_node {
        public:
            bool try_put_and_wait(const T& input);
        }; // class limiter_node

        template <typename T>
        class broadcast_node {
        public:
            bool try_put_and_wait(const T& input);
        }; // class broadcast_node

        template <typename TupleType>
        class split_node {
        public:
            bool try_put_and_wait(const TupleType& input);
        }; // class split_node
    } // namespace tbb
} // namespace oneapi

Member Functions


template <typename Output, typename Policy>
bool continue_node<Output, Policy>::try_put_and_wait(const continue_msg& input)

Effects: Increments the count of input signals received. If the incremented count is equal to the number of known predecessors, performs the body function object execution.