Asynchronous channel communicator construction

[AndreyKulikov2022]

Expected Behavior

The code passes ,hpx::collectives::create_channel_communicator on each locality is constructed.

#include <hpx/hpx_main.hpp> // IWYU pragma: keep

#include <cstddef>  // for size_t, ptrdiff_t
#include <iostream> // for cout
#include <vector>   // for vector

#include <hpx/actions/base_action.hpp>              // for HPX_REGISTER_ACTION_2
#include <hpx/actions_base/plain_action.hpp>        // for HPX_PLAIN_ACTION
#include <hpx/async_distributed/async.hpp>          // for async
#include <hpx/collectives/argument_types.hpp>       // for that_site_arg, num_sites_arg
#include <hpx/collectives/channel_communicator.hpp> // for channel_communicator, get, set
#include <hpx/hpx_finalize.hpp>            // for terminate
#include <hpx/iostream.hpp>                // for cout, ostream
#include <hpx/naming_base/id_type.hpp>     // for id_type
#include <hpx/runtime.hpp>                 // for find_all_localities, get_os_thread_count
#include <hpx/serialization/serialize.hpp> // for operator>>, operator<<

void hpx_channel_communicator_test() {
    using hpx::collectives::num_sites_arg, hpx::collectives::that_site_arg;
    constexpr auto channel_communicator_name = "test/";
    const std::uint32_t n_localities = hpx::get_initial_num_localities();
    auto comm = hpx::collectives::create_channel_communicator(hpx::launch::sync,
                    channel_communicator_name, num_sites_arg(n_localities));
}

HPX_PLAIN_ACTION(hpx_channel_communicator_test, HPXChannelCommunicatorTestAction)

int main() {
    const std::vector<hpx::id_type> localities = hpx::find_all_localities();
    std::cout << localities.size() << " localities are used.\n";
    std::vector<hpx::future<void>> futures;
    futures.reserve(localities.size());
    for (const auto& locality : localities) {
        futures.push_back(hpx::async<HPXChannelCommunicatorTestAction>(locality));
    }
    hpx::wait_all(futures);
    std::cout<<"Done\n";
    return 0;
}

Actual Behavior

Often hangs. The issue occurred when updating from HPX 1.10 to 1.11

Steps to Reproduce the Problem

Run ./my_app

Sometimes multiple runs are required to catch the issue.

Observation: the hanging seems not to occur when (by luck) the construction happens on the localities in order Loc0, Loc1, Loc2.

Changing

auto comm = hpx::collectives::create_channel_communicator(hpx::launch::sync,
                    channel_communicator_name, num_sites_arg(n_localities));

to

auto comm = hpx::collectives::create_channel_communicator(channel_communicator_name, num_sites_arg(n_localities)).get()`

solves the issue, but if this is the intended way, it does not quite follow from the documentation.

Specifications

• HPX Version: 1.11.0
• Platform (compiler, OS, Architecture): g++13.2.0, Ubuntu 24.04.1 LTS, x86_64
• Platform2 : macOS 15.6.1, clang-1700.0.13.5, arm64-apple-darwin24.6.0.

[hkaiser]

Thanks for reporting! For now, please simply wait for the creation to finish (as you suggested). We will investigate.

[hkaiser]

@AndreyKulikov2022 I was thinking about this some more. The only reason why I could imagine for this to happen is if one of the localities tries to connect to the main communicator object only after it was destroyed by the root locality. Could you try adding a barrier (or similar) before the communicator goes out of scope such that all localities have to reach it before the communicator is allowed to be destroyed?

[0lem1of1]

@hkaiser I was able to successfully reproduce the deadlock locally by running the minimal test across two distinct localities on a single machine (--hpx:localities=2 --hpx:node=0 and --hpx:node=1).

As expected, it is a distributed race condition regarding the lifecycle of the comm object. Locality 0 successfully creates the local half of the communicator, immediately hits the end of the function scope (}), and destroys the comm object. By the time Locality 1 attempts to connect to test_channel, Locality 0 has already unregistered it, causing Locality 1 to hang indefinitely while waiting for the network handshake.

Adding an hpx::distributed::barrier right before the end of the hpx_channel_communicator_test() function forces Locality 0 to keep the comm object alive in memory until Locality 1 has successfully connected. With the barrier in place, the test passes and exits cleanly every time.

To move forward with a PR, how would you prefer this be handled at the architectural level? I see two potential paths:

Library-Side Fix (Implicit Synchronization): Update the core implementation of create_channel_communicator. If invoked with hpx::launch::sync, inject an implicit internal barrier so it does not return control to the user until all num_sites have securely connected.

AGAS Fix: Modify the destruction/reference counting logic of the communicator so that the underlying network channel stays alive in the background until the collective is fully formed, even if the local C++ object goes out of scope.

[hkaiser]

@hkaiser I was able to successfully reproduce the deadlock locally by running the minimal test across two distinct localities on a single machine (--hpx:localities=2 --hpx:node=0 and --hpx:node=1).

As expected, it is a distributed race condition regarding the lifecycle of the comm object. Locality 0 successfully creates the local half of the communicator, immediately hits the end of the function scope (}), and destroys the comm object. By the time Locality 1 attempts to connect to test_channel, Locality 0 has already unregistered it, causing Locality 1 to hang indefinitely while waiting for the network handshake.

Adding an hpx::distributed::barrier right before the end of the hpx_channel_communicator_test() function forces Locality 0 to keep the comm object alive in memory until Locality 1 has successfully connected. With the barrier in place, the test passes and exits cleanly every time.

Thanks for this analysis confirming my suspicions.

For your suggestions:

To move forward with a PR, how would you prefer this be handled at the architectural level? I see two potential paths:

Library-Side Fix (Implicit Synchronization): Update the core implementation of create_channel_communicator. If invoked with hpx::launch::sync, inject an implicit internal barrier so it does not return control to the user until all num_sites have securely connected.

This is viable, however would solve the problem only for synchronously created communicators. For the asynchronous case something similar might be possible as well by delaying making the communicator ready accordingly (the communicator is essentially a future to the global id representing it).

AGAS Fix: Modify the destruction/reference counting logic of the communicator so that the underlying network channel stays alive in the background until the collective is fully formed, even if the local C++ object goes out of scope.

That sounds interesting, however I currently know if it's possible to implement. Please feel free to investigate, though.

For now I think we should add a note to the documentation saying that the user is responsible to make sure the communicator doesn't go out of scope before all participating sites have reached it.

[0lem1of1]

@hkaiser I'm working on a permanent fix , in the meantime i have added a doc warning.

[AndreyKulikov2022]

@AndreyKulikov2022 I was thinking about this some more. The only reason why I could imagine for this to happen is if one of the localities tries to connect to the main communicator object only after it was destroyed by the root locality. Could you try adding a barrier (or similar) before the communicator goes out of scope such that all localities have to reach it before the communicator is allowed to be destroyed?

Sorry for a late and now redundant response: yes, the test always passes with a barrier.