How should I correctly use hpx::distributed::barrier to synchronize tasks across all nodes

[phil-skillwon]

I created an action that is a compute-intensive task, taking about 1-7 seconds to execute. I will execute this action 8 times across 2 nodes simultaneously and use hpx::distributed::barrier for synchronization across these tasks. I referred to the documentation for hpx::distributed::barrier:

https://hpx-docs.stellar-group.org/latest/html/libs/full/collectives/api/barrier.html?highlight=barrier#hpx-distributed-barrier

Here is part of my test code:

static std::string DISTR_BARRIER_URL = "/myDistrBarrier";
static constexpr size_t NUM_TASKS = 8;

static ResultRs<int, ErrorRs> myTask(const size_t taskId, const int64_t time) 
{
    TRACE("%s", __FUNCTION__);
    double result = 0.0, ts = 0.00;
    auto ticker = Ticker();
    ticker.restart();

	auto locId = static_cast<size_t>(hpx::get_locality_id());
	
	// These lines do not work.
	// auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, NUM_TASKS);
	// auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, NUM_TASKS, locId);
	// auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, NUM_TASKS, locId);
	// auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{0, 1}, locId);
	// auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, 1, locId);
    
	// These lines work correctly.
	// auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{0}, 0);
	// auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{1}, 1);
	auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{locId}, locId);
	
	// Simulate a compute-intensive task.
    while (true) 
    {
        for (int i = 0; i < 1000000; ++i)
        {
            result += std::sin(i) * std::cos(i);
        }
        ts = ticker.elapsed(Ticker::TimePrecision::PrecisionMs);
        if (ts > time) 
        {
            break;
        }
    }

    TRACE("task %ld completed in %d ms", taskId, time);
	DEBUG("task %ld barrier wait", taskId);
    // Wait for all tasks across nodes to reach this point.
    barr.wait();
	DEBUG("task %ld barrier wait done", taskId);
	
    return static_cast<int>(taskId);
}

struct MyTaskAction 
    :hpx::actions::make_action<decltype(&myTask), &myTask, MyTaskAction>::type 
{

};

// Part of the hpx_main function.
std::vector<hpx::id_type> localities = hpx::find_all_localities();
auto rootNodeId = localities[0]; 
auto workerNodeId = localities[1]; 

vector<hpx::future<ResultRs<int, ErrorRs>>> futVec;
for (size_t i = 0; i < NUM_TASKS; ++i) 
{
	auto fut = hpx::async<MyComputeTaskAction>(0 != i % 2 ? rootNodeId : workerNodeId, i, 10 * 1000); 
	futVec.push_back(std::move(fut));
}

hpx::wait_all(futVec);

My question is:

Why do any of the following lines cause barr.wait() in the myTask function to block indefinitely?

auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, NUM_TASKS);
auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, NUM_TASKS, locId);
auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{0, 1}, locId);
auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, 1, locId);

While these lines allow barr.wait() to return:

auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{0}, 0);
auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{1}, 1);
auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{locId}, locId);

I thought this line should also allow barr.wait() to return, but it doesn't synchronize tasks across nodes correctly in my test:

auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, NUM_TASKS, locId);

Only these three lines work for cross-node synchronization:

auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{0}, 0);
auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{1}, 1);
auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{locId}, locId);

How should I correctly use hpx::distributed::barrier to synchronize tasks across all nodes?

[phil-skillwon]

@hkaiser
Could you please take some time to look into this issue?

[hkaiser]

@hkaiser Could you please take some time to look into this issue?

Sorry for the late reply, I missed your original question. Could you please create minimal full test cases reproducing your issues for me to check?

[phil-skillwon]

DistributedBarrier.zip

@hkaiser
Sorry, my docker image is too large, and I'm unsure how to upload it.

Here is my test code(DistributedBarrier.zip), which consists of only one main.cpp file and uses CMake for building. The third-party libraries used are:

• asio-1.28.1.tar.bz2
• boost_1_84_0.tar.bz2
• hpx-1.10.0.tar.gz
• hwloc-2.10.0.tar.bz2
• gperftools-2.15.tar.gz

This test project may not work in your development environment because the .hpp files for libraries such as Asio and Boost exceed 25 MB, which prevents them from being uploaded to GitHub.
In the zip file I uploaded, I removed all Boost-related source files from the Boost directory.

Could you extract and recompile main.cpp in your environment?
To compile and run, navigate to the root directory of the test project and use:

cd /path/to/testProject
bash .vscode/AutoBuild.bash debug

The executable will be at:

/path/to/testProject/bin/debug/HpxDemo_d.elf

To run:

./HpxDemo_d.elf --hpx:threads=4 --hpx:localities=2 --hpx:agas=127.0.0.1:7910 --hpx:hpx=127.0.0.1:7910
./HpxDemo_d.elf --hpx:worker --hpx:agas=127.0.0.1:7910 --hpx:hpx=127.0.0.1:7911

My question is: Although the distributed barrier seems correctly used in my test, it doesn't work as expected. The following works:

auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, vector<size_t>{locId}, locId);

But this doesn't:

auto barr = hpx::distributed::barrier(DISTR_BARRIER_URL, NUM_TASKS);

Thank you very much for your help.

[hkaiser]

Well, some minimal C++ code would suffice. The overall environment is not required.

[phil-skillwon]

Here is my code.
main.cpp.md