Skip to content

Commit

Permalink
wait_xxx_nothrow functions return whether one of the futures is excep…
Browse files Browse the repository at this point in the history 
…tional
  • Loading branch information
@hkaiser
hkaiser committed Apr 10, 2022
1 parent 9b43f25 commit b4c2142
Show file tree
Hide file tree
Showing 18 changed files with 390 additions and 303 deletions.
11 changes: 6 additions & 5 deletions libs/core/algorithms/include/hpx/parallel/algorithms/make_heap.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -344,11 +344,12 @@ namespace hpx { namespace parallel { inline namespace v1 {
policy.executor(), op, shapes);

// Required synchronization per level
hpx::wait_all_nothrow(workitems);

// collect exceptions
util::detail::handle_local_exceptions<
ExPolicy>::call(workitems, errors, false);
if (hpx::wait_all_nothrow(workitems))
{
// collect exceptions
util::detail::handle_local_exceptions<
ExPolicy>::call(workitems, errors, false);
}
}

if (!errors.empty())
Expand Down
26 changes: 14 additions & 12 deletions libs/core/algorithms/include/hpx/parallel/util/foreach_partitioner.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -138,25 +138,27 @@ namespace hpx { namespace parallel { namespace util {
std::pair<Items1, Items2>&& items, F&& f, FwdIter last)
{
// wait for all tasks to finish
hpx::wait_all_nothrow(hpx::get<0>(items), hpx::get<1>(items));

// always rethrow if inititems/workitems have at least one
// exceptional future
handle_local_exceptions::call(hpx::get<0>(items));
handle_local_exceptions::call(hpx::get<1>(items));

if (hpx::wait_all_nothrow(
hpx::get<0>(items), hpx::get<1>(items)))
{
// always rethrow if inititems/workitems have at least one
// exceptional future
handle_local_exceptions::call(hpx::get<0>(items));
handle_local_exceptions::call(hpx::get<1>(items));
}
return f(HPX_MOVE(last));
}

template <typename F, typename Items, typename FwdIter>
static FwdIter reduce(Items&& items, F&& f, FwdIter last)
{
// wait for all tasks to finish
hpx::wait_all_nothrow(items);

// always rethrow if items has at least one exceptional future
handle_local_exceptions::call(items);

if (hpx::wait_all_nothrow(items))
{
// always rethrow if items has at least one exceptional
// future
handle_local_exceptions::call(items);
}
return f(HPX_MOVE(last));
}
};
Expand Down
21 changes: 12 additions & 9 deletions libs/core/algorithms/include/hpx/parallel/util/partitioner.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -283,22 +283,25 @@ namespace hpx { namespace parallel { namespace util {
static R reduce(Items&& workitems, F&& f)
{
// wait for all tasks to finish
hpx::wait_all_nothrow(workitems);

// always rethrow workitems has at least one exceptional future
handle_local_exceptions::call(workitems);

if (hpx::wait_all_nothrow(workitems))
{
// always rethrow workitems has at least one exceptional
// future
handle_local_exceptions::call(workitems);
}
return f(HPX_MOVE(workitems));
}

template <typename Items>
static void reduce(Items&& workitems, hpx::util::empty_function)
{
// wait for all tasks to finish
hpx::wait_all_nothrow(workitems);

// always rethrow workitems has at least one exceptional future
handle_local_exceptions::call(workitems);
if (hpx::wait_all_nothrow(workitems))
{
// always rethrow workitems has at least one exceptional
// future
handle_local_exceptions::call(workitems);
}
}

template <typename Items1, typename Items2, typename F>
Expand Down
14 changes: 7 additions & 7 deletions libs/core/algorithms/include/hpx/parallel/util/partitioner_with_cleanup.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -86,13 +86,13 @@ namespace hpx { namespace parallel { namespace util {
static R reduce(Items&& workitems, F&& f, Cleanup&& cleanup)
{
// wait for all tasks to finish
hpx::wait_all_nothrow(workitems);

// always rethrow if 'errors' is not empty or workitems has
// exceptional future
handle_local_exceptions::call_with_cleanup(
workitems, HPX_FORWARD(Cleanup, cleanup));

if (hpx::wait_all_nothrow(workitems))
{
// always rethrow if 'errors' is not empty or workitems has
// exceptional future
handle_local_exceptions::call_with_cleanup(
workitems, HPX_FORWARD(Cleanup, cleanup));
}
return f(HPX_MOVE(workitems));
}

Expand Down
30 changes: 18 additions & 12 deletions libs/core/algorithms/include/hpx/parallel/util/scan_partitioner.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,10 @@ namespace hpx { namespace parallel { namespace util {
}

// Wait for all f1 tasks to finish
hpx::wait_all_nothrow(workitems);
if (hpx::wait_all_nothrow(workitems))
{
handle_local_exceptions::call(workitems, errors);
}

// perform f2 sequentially in one go
f2results.resize(workitems.size());
Expand Down Expand Up @@ -343,12 +346,14 @@ namespace hpx { namespace parallel { namespace util {
return R();
#else
// wait for all tasks to finish
hpx::wait_all_nothrow(workitems, finalitems);

// always rethrow if 'errors' is not empty or 'workitems' or
// 'finalitems' have an exceptional future
handle_local_exceptions::call(workitems, errors);
handle_local_exceptions::call(finalitems, errors);
if (hpx::wait_all_nothrow(workitems, finalitems) ||
!errors.empty())
{
// always rethrow if 'errors' is not empty or 'workitems' or
// 'finalitems' have an exceptional future
handle_local_exceptions::call(workitems, errors);
handle_local_exceptions::call(finalitems, errors);
}

try
{
Expand Down Expand Up @@ -376,11 +381,12 @@ namespace hpx { namespace parallel { namespace util {
return R();
#else
// wait for all tasks to finish
hpx::wait_all_nothrow(finalitems);

// always rethrow if 'errors' is not empty or
// 'finalitems' have an exceptional future
handle_local_exceptions::call(finalitems, errors);
if (hpx::wait_all_nothrow(finalitems) || !errors.empty())
{
// always rethrow if 'errors' is not empty or 'finalitems'
// have an exceptional future
handle_local_exceptions::call(finalitems, errors);
}

try
{
Expand Down
Loading

0 comments on commit b4c2142

Please sign in to comment.