4.16. Distributed Systems

4.16.1. Introduction

As an HPC visualization toolkit, Viskores is designed to be executed in distributed systems. This is a key requirement for resource-expensive applications where a computational job can be partitioned into different tasks, which are then assigned to different processes located on potentially different nodes. Those nodes, as a composite, will normally form a cluster of computers or a supercomputer. These computing tasks can communicate and coordinate with each other through the use of a middleware library, which in the case of Viskores is the DIY library. Furthermore, both to launch jobs and ultimately to communicate between tasks, Viskores delegates to MPI (Message Passing Interface).

Despite the fact that only some Viskores filters can run out of the box as a distributed job, many other Viskores components can run as a distributed job by manually partitioning the computational problem data and assigning one of those partitions to each task.

4.16.2. DIY

DIY is a block-parallel library for implementing scalable algorithms that can execute both in-core and out-of-core. The same program can be executed with one or more threads per MPI process, seamlessly combining distributed-memory message passing with shared-memory thread parallelism. The abstraction enabling these capabilities is block parallelism: blocks and their message queues are mapped onto processing elements, consisting of either MPI processes or threads, and are migrated between memory and storage by the DIY runtime. Complex communication patterns, including neighbor exchange, merge reduction, swap reduction, and all-to-all exchange, are possible in DIY both in-core and out-of-core.

A full description of using DIY to perform distributed visualization is beyond the scope of this guide. For a full description, refer to the documentation provided by DIY. The basic procedure of any DIY algorithm is to first define a set of blocks, assign them to the ranks of an MPI job, and define neighborhood relationships between them. The following example demonstrates defining a set of blocks, one per MPI rank.

Example 4.175 Communication setup of an example DIY application.
 1  viskoresdiy::mpi::communicator comm;
 2  viskores::cont::EnvironmentTracker::SetCommunicator(comm);
 3
 4  auto nblocks = comm.size();
 5  std::vector<int> gids;
 6
 7  viskoresdiy::RoundRobinAssigner assigner(comm.size(), nblocks);
 8  assigner.local_gids(comm.rank(), gids);
 9
10  // In our example nblocks == num_ranks, thus gids.size() == 1
11  auto gid = gids[0];
12
13  // The link will be eventually freed by DIY.
14  auto link = new viskoresdiy::Link;
15
16  // Connect each blocks with itself.
17  viskoresdiy::BlockID neighbor;
18  neighbor.gid = gid;
19  neighbor.proc = assigner.rank(neighbor.gid);
20  link->add_neighbor(neighbor);

Did You Know?

When using DIY objects from inside Viskores, use the objects in the mangled viskoresdiy namespace rather than the diy namespace. Viskores uses this mangled namespace to prevent conflicts if it is used with another library or executable that uses a different version of DIY.

Any distributed algorithm needs to establish a “communicator” that defines the group of processes that are communicating. Viskores uses the viskores::cont::EnvironmentTracker class to manage a global communicator used within Viskores. Example 4.175, line 2 demonstrates establishing the communicator.

class EnvironmentTracker

Maintain MPI controller, if any, for distributed operation.

EnvironmentTracker is a class that provides static API to track the global MPI controller to use for operating in a distributed environment.

Public Static Functions

static void SetCommunicator(const viskoresdiy::mpi::communicator &comm)

Set a global communicator to be used by Viskores.

static const viskoresdiy::mpi::communicator &GetCommunicator()

Get the global communicator to be used by Viskores.

Communication in DIY is managed by the viskoresdiy::Master object. References to the defined blocks are added to the viskoresdiy::Master object. You can then run an operation on each of these blocks using the foreach method, which is given a function to execute on each block. This function is provided with a proxy that enables communicating data with other nodes using a variety of communication patterns. These communications do not happen right away but rather are queued for later exchange. This exchange is done by calling the free function viskores::cont::DIYMasterExchange(). The following example, which builds on the previous one, finds the median value of an array on each rank and then finds the maximum median value. The blocks and communication used by these examples are outlined in Figure 4.5.

void viskores::cont::DIYMasterExchange(viskoresdiy::Master &master, bool remote = false)

Wraps viskoresdiy::Master::exchange by setting its appropriate viskoresdiy::MemoryManagement.

Example 4.176 Example DIY application that finds the maximum of the medians of different ArrayHandle objects.
 1  viskoresdiy::Master master(comm);
 2
 3  struct MyBlock
 4  {
 5    viskores::cont::ArrayHandle<viskores::Int32> in;
 6  };
 7
 8  MyBlock block{ inputArrayHandle };
 9  master.add(gid, &block, link);
10
11  master.foreach (
12    [&](MyBlock* b, const viskoresdiy::Master::ProxyWithLink& cp)
13    {
14      viskores::cont::Algorithm::Sort(b->in);
15      cp.enqueue(cp.link()->target(0), b->in);
16    });
17  viskores::cont::DIYMasterExchange(master);
18  master.foreach (
19    [&](MyBlock* b, const viskoresdiy::Master::ProxyWithLink& cp)
20    {
21      cp.dequeue(cp.link()->target(0).gid, b->in);
22
23      auto median_idx = (b->in.GetNumberOfValues() / 2) - 1;
24      auto median = viskores::cont::ArrayGetValue(median_idx, b->in);
25
26      cp.all_reduce(median, viskoresdiy::mpi::maximum<viskores::Int32>());
27    });
28  viskores::cont::DIYMasterExchange(master);
29
30  if (comm.rank() == 0)
31  {
32    std::cout << "Max(median): "
33              << master.proxy(master.loaded_block()).get<viskores::Int32>() << std::endl;
34  }
_images/DIYApp.png

Figure 4.5 Communication topology of the example DIY application shown in Example 4.175 and Example 4.176.

Common Errors

Normally, the DIY exchange process is done by calling the viskoresdiy::Master::exchange method. However, when using DIY with Viskores, the exchange should instead be done by calling viskores::cont::DIYMasterExchange(). This function allows Viskores to enable the state needed to interface between DIY and Viskores data without otherwise affecting exchanges that happen outside of Viskores.

4.16.3. Object Serialization

When data are transferred among ranks, the format needs to be packed in a way the message layer understands. Objects that might have complex structure typically need to be converted to one or more buffers through serialization.

DIY provides out-of-the-box serialization of common C++ standard-library types such as std::vector and std::string. Viskores also provides serialization for common Viskores data types such as viskores::cont::ArrayHandle and viskores::cont::DataSet. This list is not exhaustive, since Viskores also provides DIY serialization for many other data types. For custom data types, the user can specify how to serialize and deserialize the desired type by defining an additional template specialization for struct viskoresdiy::Serialization. An example of this can be found in Example 4.177.

Example 4.177 Example DIY application that demonstrates how to serialize custom data types in DIY.
 1struct TimedCoords
 2{
 3  viskores::cont::ArrayHandle<viskores::UInt64> TimeStamps;
 4  viskores::cont::ArrayHandle<viskores::Vec3i> Coordinates;
 5};
 6
 7namespace viskoresdiy
 8{
 9template<>
10struct Serialization<TimedCoords>
11{
12  static void save(BinaryBuffer& bb, const TimedCoords& p)
13  {
14    viskoresdiy::save(bb, p.TimeStamps);
15    viskoresdiy::save(bb, p.Coordinates);
16  }
17  static void load(BinaryBuffer& bb, TimedCoords& p)
18  {
19    viskoresdiy::load(bb, p.TimeStamps);
20    viskoresdiy::load(bb, p.Coordinates);
21  }
22};
23}
24
25void compute(viskoresdiy::Master& master, viskoresdiy::Link* link, int gid)
26{
27  TimedCoords timedCoords;
28  master.add(gid, &timedCoords, link);
29
30  master.foreach (
31    [&](TimedCoords* tc, const viskoresdiy::Master::ProxyWithLink& cp)
32    {
33      *tc = ComputeLocalCoords(gid);
34      cp.enqueue(cp.link()->target(0), *tc);
35    });
36  viskores::cont::DIYMasterExchange(master);
37  master.foreach (
38    [&](TimedCoords* tc, const viskoresdiy::Master::ProxyWithLink& cp)
39    {
40      cp.dequeue(cp.link()->target(0).gid, *tc);
41      auto expectedVec = ComputeLocalCoords(gid);
42      if (*tc != expectedVec)
43      {
44        std::cerr << "ERROR: recieved incorrect vec values." << std::endl;
45      }
46    });
47}

4.16.4. GPU-aware MPI

Modern HPC GPUs allow direct GPU-to-GPU communication. This provides GPUs with an efficient mechanism to directly send data stored in their device memory to the target GPU’s device memory. This is a significant departure from the traditional approach, where the NIC is solely accessible from the CPU, constraining applications to a costly GPU communication pattern consisting of first copying the desired data from device memory to host memory, transferring it over the network, and then copying the received data from host memory to device memory.

Both major GPU parallel platforms, ROCm and CUDA, provide APIs that support direct GPU-to-GPU communication. Nevertheless, to avoid vendor lock-in, Viskores does not directly use these APIs. Instead, Viskores delegates to MPI, which implements a unified and standardized API for GPU-to-GPU communication. Consequently, Viskores provides users with the capability to use direct GPU-to-GPU communication during MPI-based distributed executions of Viskores applications.

Viskores can autonomously determine whether GPU-to-GPU communication is possible. Consequently, it does not provide a specific API to control this type of communication. The communication type is selected on each call to the free function viskores::cont::DIYMasterExchange(), demonstrated in Example 4.176. The function internally decorates viskoresdiy::Master::exchange so that it can perform GPU-to-GPU communication when the situation allows it.

This GPU-aware MPI feature can be enabled with the flag VISKORES_ENABLE_GPU_MPI=ON. Enabling this feature on target supercomputers often requires additional setup that depends on the particular system. Refer to the target system’s documentation for further information.