collectives

collectives#

The collectives module exposes a set of distributed collective operations. Those can be used to exchange data between participating sites in a coordinated way. At this point the module exposes the following collective primitives:

See the API reference of the module for more details.

Hierarchical collectives#

Hierarchical communicators split a collective into sub-communicators arranged as a tree. This reduces the number of participants in each individual operation, but it also means that one user-visible hierarchical call can touch several internal communicators.

Using a hierarchical communicator#

Create a hpx::collectives::hierarchical_communicator with hpx::collectives::create_hierarchical_communicator. Every participating site has to call it with the same base name, number of sites, arity, generation, root site, and flat fallback threshold, and with its own site index. These arguments decide the tree of communicators that each site builds, so a mismatch can make the collective operations hang:

    // All sites create the communicator with the same arguments. The arity is
    // the number of children of each node in the tree of communicators.
    //
    // Below the flat fallback threshold (16 sites by default) the
    // communicator collapses to a single flat communicator. This example
    // passes 0 to disable that fallback, so that it builds a tree even on a
    // few localities, as long as the arity is smaller than the number of
    // localities.
    auto const comm = create_hierarchical_communicator(hierarchical_basename,
        num_sites_arg(num_localities), this_site_arg(this_locality),
        arity_arg(static_cast<std::size_t>(arity)), generation_arg(),
        root_site_arg(0), flat_fallback_threshold_arg(0));

The resulting communicator can be passed to the collective operations in place of a flat communicator. All of hpx::collectives::all_gather, hpx::collectives::all_reduce, hpx::collectives::all_to_all, hpx::collectives::broadcast_to and hpx::collectives::broadcast_from, hpx::collectives::gather_here and hpx::collectives::gather_there, hpx::collectives::reduce_here and hpx::collectives::reduce_there, hpx::collectives::scatter_to and hpx::collectives::scatter_from, hpx::collectives::inclusive_scan, hpx::collectives::exclusive_scan, and hpx::collectives::barrier accept it. The example below runs three different operations on one shared communicator. It keeps a single generation counter and increments it for every call, whatever the operation, as described in the next section:

    // A communicator that is shared between several operations needs an
    // explicit generation number on every call, and the numbers must be
    // strictly consecutive across all of those calls, starting at 1. Keep a
    // single counter and increment it for every call, whatever the
    // operation. Two-phase operations such as all_reduce and all_gather
    // reject the default generation even when the communicator is not shared.
    std::size_t generation = 0;

    // Sum up the locality ids. Every site receives the result.
    std::uint32_t const sum =
        all_reduce(comm, this_locality, std::plus<std::uint32_t>{},
            this_site_arg(this_locality), generation_arg(++generation))
            .get();

    // Collect the squares of the locality ids on every site.
    std::vector<std::uint32_t> const squares =
        all_gather(comm, this_locality * this_locality,
            this_site_arg(this_locality), generation_arg(++generation))
            .get();

    // Send a value from the root site to all other sites. Only the root site
    // calls broadcast_to; all others call broadcast_from.
    std::string message;
    if (this_locality == 0)
    {
        message = broadcast_to(comm, std::string("hello from the root"),
            this_site_arg(this_locality), generation_arg(++generation))
                      .get();
    }
    else
    {
        message = broadcast_from<std::string>(
            comm, this_site_arg(this_locality), generation_arg(++generation))
                      .get();
    }

The complete program is hierarchical_collectives.cpp. It runs hpx_main on all localities, so start it on several localities, for example with hpxrun.py -l 4 -t 1 bin/hierarchical_collectives -- --arity=2.

A few rules apply to all hierarchical collectives:

  • The arity is the number of children of each node in the tree. It must be at least 2 and defaults to 2. A small arity gives a deeper tree with fewer participants per sub-communicator, a large arity gives a flatter tree with fewer steps. The number of sites does not have to be a power of the arity.

  • If the number of sites is smaller than the flat fallback threshold (hpx::collectives::flat_fallback_threshold_arg, 16 by default), the communicator collapses to a single flat communicator spanning all sites. Pass flat_fallback_threshold_arg(0) to disable this fallback. The communicator is still a single flat communicator if the arity is not smaller than the number of sites.

  • Any site can be the root of the tree, but all sites have to pass the same root_site_arg. Only the root site may call the root-side operations (broadcast_to, gather_here, reduce_here, and scatter_to). The operations that take a root_site_arg although every site receives the result (all_gather, all_reduce, the scans, and barrier) accept only 0 (the default) or the root of the communicator, and report bad_parameter for any other site.

  • The futures returned by the asynchronous hierarchical operations are not fully non-blocking: walking the tree of communicators can wait internally before the returned future is ready.

Generation model#

Hierarchical collectives use two internal generations for each user generation. For a user generation k, the first internal phase uses 2k - 1 and the second uses 2k. Single-pass hierarchical operations and the inter-group exchange of hpx::collectives::all_to_all advance the communicator by two generations in one step so that shared hierarchical communicators stay aligned with two-phase collectives.

When a hierarchical communicator instance is shared between collective operations, callers must pass explicit, strictly consecutive positive generation numbers. Two-phase hierarchical collectives, including hpx::collectives::all_gather, hpx::collectives::all_reduce, hpx::collectives::all_to_all, hpx::collectives::inclusive_scan, hpx::collectives::exclusive_scan, and hpx::distributed::barrier, reject the default generation because their phase generations are derived from the explicit user generation.

Contents