Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix limiter_node decrementer #647

Merged
merged 8 commits into from
Dec 13, 2021
Merged

Fix limiter_node decrementer #647

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
ab8698b
Fix limiter_node decrementer
Iliamish Nov 10, 2021
e512424
Fix align
Iliamish Nov 10, 2021
91a7606
add new variable for check unused decrements
Iliamish Nov 11, 2021
ae2b441
remove iostream
Iliamish Nov 11, 2021
ce04660
remove unnesessary cast
Iliamish Nov 11, 2021
f51fb90
align whitespaces
Iliamish Nov 12, 2021
ee3b7b6
Fix whitespace alignment
Nov 22, 2021
f620936
Revert workaround for GCC
Nov 22, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 48 additions & 21 deletions include/oneapi/tbb/flow_graph.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1886,6 +1886,7 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
size_t my_threshold;
size_t my_count; // number of successful puts
size_t my_tries; // number of active put attempts
size_t my_future_decrement; // number of active decrement
reservable_predecessor_cache< T, spin_mutex > my_predecessors;
spin_mutex my_mutex;
broadcast_cache< T > my_successors;
Expand All @@ -1894,12 +1895,19 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
threshold_regulator< limiter_node<T, DecrementType>, DecrementType > decrement;

graph_task* decrement_counter( long long delta ) {
if ( delta > 0 && size_t(delta) > my_threshold ) {
delta = my_threshold;
}

{
spin_mutex::scoped_lock lock(my_mutex);
if( delta > 0 && size_t(delta) > my_count ) {
if ( delta > 0 && size_t(delta) > my_count ) {
if( my_tries > 0 ) {
my_future_decrement += (size_t(delta) - my_count);
}
my_count = 0;
}
else if( delta < 0 && size_t(-delta) > my_threshold - my_count ) {
else if ( delta < 0 && size_t(-delta) > my_threshold - my_count ) {
my_count = my_threshold;
}
else {
Expand All @@ -1924,23 +1932,34 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
input_type v;
graph_task* rval = NULL;
bool reserved = false;
{
spin_mutex::scoped_lock lock(my_mutex);
if ( check_conditions() )
++my_tries;
else
return NULL;
}

{
spin_mutex::scoped_lock lock(my_mutex);
if ( check_conditions() )
++my_tries;
else
return NULL;
}

//SUCCESS
// if we can reserve and can put, we consume the reservation
// we increment the count and decrement the tries
if ( (my_predecessors.try_reserve(v)) == true ){
reserved=true;
if ( (rval = my_successors.try_put_task(v)) != NULL ){
if ( (my_predecessors.try_reserve(v)) == true ) {
reserved = true;
if ( (rval = my_successors.try_put_task(v)) != NULL ) {
{
spin_mutex::scoped_lock lock(my_mutex);
++my_count;
if ( my_future_decrement ) {
if ( my_count > my_future_decrement ) {
my_count -= my_future_decrement;
my_future_decrement = 0;
}
else {
my_future_decrement -= my_count;
my_count = 0;
}
}
--my_tries;
my_predecessors.try_consume();
if ( check_conditions() ) {
Expand Down Expand Up @@ -1988,8 +2007,8 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
public:
//! Constructor
limiter_node(graph &g, size_t threshold)
: graph_node(g), my_threshold(threshold), my_count(0), my_tries(0), my_predecessors(this)
, my_successors(this), decrement(this)
: graph_node(g), my_threshold(threshold), my_count(0), my_tries(0), my_future_decrement(0),
my_predecessors(this), my_successors(this), decrement(this)
{
initialize();
}
Expand Down Expand Up @@ -2071,7 +2090,6 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
}

graph_task* rtask = my_successors.try_put_task(t);

if ( !rtask ) { // try_put_task failed.
spin_mutex::scoped_lock lock(my_mutex);
--my_tries;
Expand All @@ -2085,22 +2103,31 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
else {
spin_mutex::scoped_lock lock(my_mutex);
++my_count;
if ( my_future_decrement ) {
if ( my_count > my_future_decrement ) {
my_count -= my_future_decrement;
my_future_decrement = 0;
}
else {
my_future_decrement -= my_count;
my_count = 0;
}
}
--my_tries;
Comment on lines 2104 to 2116
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This part of the code is a complete (both semantic and syntactic) duplication of lines 1951-1963 in the newer version of the same file. I would recommend moving them into separate limiter_node-specific helper method, which will be named correspondingly. E.g., account_successful_forward.

There is also similar (slightly incomplete) code duplication about forwarding failure: see lines 2094-2101 and 1983-1994 in the newer version of the file. Please consider moving them as well.

Of course, it is mainly for better code readability and maintainability in the future, and not a showstopper for now. If there is no time or desire implementing this "small" improvement, consider leaving a TODO comment, something like: "// TODO: consider moving duplicated logic about processing of successful and unsuccessful task forward into separate functions."

}
}
return rtask;
}

graph& graph_reference() const override { return my_graph; }

void reset_node( reset_flags f) override {
void reset_node( reset_flags f ) override {
my_count = 0;
if(f & rf_clear_edges) {
if ( f & rf_clear_edges ) {
my_predecessors.clear();
my_successors.clear();
}
else
{
my_predecessors.reset( );
else {
my_predecessors.reset();
}
decrement.reset_receiver(f);
}
Expand Down
51 changes: 37 additions & 14 deletions test/tbb/test_limiter_node.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,18 +125,6 @@ struct put_dec_body : utils::NoAssign {

};

template< typename Sender, typename Receiver >
void make_edge_impl(Sender& sender, Receiver& receiver){
#if __GNUC__ < 12 && !TBB_USE_DEBUG
// Seemingly, GNU compiler generates incorrect code for the call of limiter.register_successor in release (-03)
// The function pointer to make_edge workarounds the issue for unknown reason
auto make_edge_ptr = tbb::flow::make_edge<int>;
make_edge_ptr(sender, receiver);
#else
tbb::flow::make_edge(sender, receiver);
#endif
}

template< typename T >
void test_puts_with_decrements( int num_threads, tbb::flow::limiter_node< T >& lim , tbb::flow::graph& g) {
parallel_receiver<T> r(g);
Expand Down Expand Up @@ -367,7 +355,7 @@ void test_reserve_release_messages() {
broad.try_put(1); //failed message retrieved.
g.wait_for_all();

make_edge_impl(limit, output_queue); //putting the successor back
tbb::flow::make_edge(limit, output_queue); //putting the successor back

broad.try_put(1); //drop the count

Expand Down Expand Up @@ -465,7 +453,7 @@ void test_try_put_without_successors() {
}
);

make_edge_impl(ln, fn);
tbb::flow::make_edge(ln, fn);

g.wait_for_all();
CHECK((counter == i * try_put_num / 2));
Expand Down Expand Up @@ -517,6 +505,35 @@ void test_deduction_guides() {
}
#endif

void test_decrement_while_try_put_task() {
constexpr int threshold = 50000;

tbb::flow::graph graph{};
std::atomic<int> processed;
tbb::flow::input_node<int> input{ graph, [&](tbb::flow_control & fc) -> int {
static int i = {};
if (i++ >= threshold) fc.stop();
return i;
}};
tbb::flow::limiter_node<int, int> blockingNode{ graph, 1 };
tbb::flow::multifunction_node<int, std::tuple<int>> processing{ graph, tbb::flow::serial,
[&](const int & value, typename decltype(processing)::output_ports_type & out) {
if (value != threshold)
std::get<0>(out).try_put(1);
processed.store(value);
}};

tbb::flow::make_edge(input, blockingNode);
tbb::flow::make_edge(blockingNode, processing);
tbb::flow::make_edge(processing, blockingNode.decrementer());

input.activate();

graph.wait_for_all();
CHECK_MESSAGE(processed.load() == threshold, "decrementer terminate flow graph work");
}


//! Test puts on limiter_node with decrements and varying parallelism levels
//! \brief \ref error_guessing
TEST_CASE("Serial and parallel tests") {
Expand All @@ -537,6 +554,12 @@ TEST_CASE("Test continue_msg reception") {
test_continue_msg_reception();
}

//! Test put message on decrementer port does not stop message flow
//! \brief \ref error_guessing
TEST_CASE("Test try_put to decrementer while try_put to limiter_node") {
test_decrement_while_try_put_task();
}

//! Test multifunction_node connected to limiter_node
//! \brief \ref error_guessing
TEST_CASE("Multifunction connected to limiter") {
Expand Down