Dask is a popular parallel and distributed computing framework, which rivals Apache Spark to enable task-based scalable processing of big data. The Dask Distributed library forms the basis of this computing engine and provides support for adding new communication devices. It currently has two communication devices: one for TCP and the other for high-speed networks using UCX-Py—a Cython wrapper to UCX. This paper presents the design and implementation of a new communication backend for Dask—called MPI4Dask—that is targeted for modern HPC clusters built with GPUs. MPI4Dask exploits mpi4py over MVAPICH2-GDR, which is a GPU-aware implementation of the Message Passing Interface (MPI) standard. MPI4Dask provides point-to-point asynchronous I/O communication coroutines, which are non-blocking concurrent operations defined using the async/await keywords from the Python’s asyncio framework. Our latency and throughput comparisons suggest that MPI4Dask outperforms UCX by 6× for 1 Byte message and 4× for large messages (2 MBytes and beyond) respectively. We also conduct comparative performance evaluation of MPI4Dask with UCX using two benchmark applications: 1) sum of cuPy array with its transpose, and 2) cuDF merge. MPI4Dask speeds up the overall execution time of the two applications by an average of 3.47× and 3.11× respectively on an in-house cluster built with NVIDIA Tesla V100 GPUs for 1 – 6 Dask workers. We also perform scalability analysis of MPI4Dask against UCX for these applications on TACC’s Frontera (GPU) system with upto 32 Dask workers on 32 NVIDIA Quadro RTX 5000 GPUs and 256 CPU cores. MPI4Dask speeds up the execution time for cuPy and cuDF applications by an average of 1.71× and 2.91× respectively for 1 − 32 Dask workers on the Frontera (GPU) system.

Modern HPC systems built with emerging multi-/many -core architectures have high core-counts and deep memory hierarchies. It is challenging to design communication libraries on these systems with the conventional wisdom of using OS processes as the basic building block to build communication algorithms. Instead, the next generation of communication libraries should treat hardware as the “first-class citizen” and utilize the underlying topology as the basic building block. Driven by this overarching principle, we present a framework for Optimized Shared Memory Processing (OSMP) and communication for these platforms. An abstract representation of the underlying hardware topology is maintained by OSMP in the form of a topology tree, which is later exploited by runtime libraries to execute communication operations in a topology-aware manner. This can be done by simply traversing the topology tree with an existing communication primitive as the base-case. OSMP does not mandate any changes to the original communication algorithm. We focus on collective operations such as barrier, reduction, and broadcast as candidate communication patterns. We demonstrate the efficacy of OSMP by decoupling the implementation of collective algorithms and system topology and evaluate it on four state-of-the-art multi-tmany-core architectures: Intel Cascade Lake, AMD Rome, ARM A64fx and IBM POWER9. Results show that even the basic algorithms can be made topology-aware by exploiting OSMP. This provides significant benefits over state-of-the-art algorithm implementations for intra-node communication. Using various micro-benchmarks and applications, we demonstrate that our proposed designs can achieve up to 7.8× improvements at the micro-benchmark level, and 15% for applications over state-of-the-art intra-node collective communication designs employed by production MPI libraries.

Python is emerging as a popular language in the data science community due to its ease-of-use, vibrant community, and rich set of libraries. Dask is a popular Python-based distributed computing framework that allows users to process large amounts of data on parallel hardware. The Dask distributed package is a non-blocking, asynchronous, and concurrent library that offers support for distributed execution of tasks on datacenter and HPC environments. A few key requirements of designing high-performance communication backends for Dask distributed is to provide scalable support for coroutines that are unlike regular Python functions and can only be invoked from asynchronous applications. In this paper, we present Blink-a high-performance communication library for Dask on high-performance RDMA networks like InfiniBand. Blink offers a multi-layered architecture that matches the communication requirements of Dask and exploits high-performance interconnects using a Cython wrapper layer to the C backend. We evaluate the performance of Blink against other counterparts using various micro-benchmarks and application kernels on three different cluster testbeds with varying interconnect speeds. Our micro-benchmark evaluation reveals that Blink outperforms other communication backends by more than 3× for message sizes ranging from 1 Byte to 64 KByte, and by a factor of 2× for message sizes ranging from 128 KByte to 8 MByte. Using various application-level evaluations, we demonstrate that Dask achieves up to 7% improvement in application throughput (e.g., total worker throughput).

Modern multi-/many-cores offer higher core-density, hardware multi-threading, deeper memory hierarchies, and diverse architectural capabilities. While emerging cloud-based HPC systems are able to deliver near-native performance, they bring more diversity to the architectures. The Message Passing Interface (MPI) offers the flexibility to arbitrarily bind application processes to CPU cores, however the static nature of these binding policies typically does not take applications' communication patterns and underlying machine architecture into consideration. This lack of association between the dynamic nature of applications and architectural diversity offered by modern processors makes it difficult for the application developers and MPI designers to exploit modern multi-/many-core systems to their full potential. In this paper, we propose a set of low-level benchmarking based approaches and MPI-level designs to infer vendor-specific machine characteristics e.g., physical to virtual machine topologies, and dynamic communication patterns of the applications. By utilizing this information, we propose two novel algorithms to construct efficient MPI mappings for any given architecture and application communication pattern. The proposed designs are implemented in the MVAPICH2 MPI library and are evaluated on three different architectures using various micro-benchmarks and application kernels. We demonstrate up to 2X performance improvement for MPI collectives, and up to 3.5X and 26% improvement for NAS-CG and miniAMR application kernels, respectively.

Data-parallelism has become an established paradigm to train DNNs that fit inside GPU memory on large-scale HPC systems. However, model-parallelism is required to train out-of-core DNNs. In this paper, we deal with emerging requirements brought forward by very large DNNs being trained using high-resolution images common in digital pathology. To address these, we propose, design, and implement GEMS; a GPU-Enabled Memory-Aware Model-Parallelism System. We present several design schemes like GEMS-MAST, GEMS-MASTER, and GEMS-Hybrid that offer excellent speedups over state-of-the-art systems like Mesh-TensorFlow and FlexFlow. Furthermore, we combine model-parallelism and data-parallelism to train a 1000-1ayer ResNet-lk model using 1,024 Volta V100 GPUs with 97.32% scaling-efficiency. For the real-world histopathology whole-slide-image (WSI) of 100,000 x 100,000 pixels, we train custom ResNet-110-v2 on image tiles of size 1024 x 1024 and reduce the training time from seven hours to 28 minutes.

Emerging multi-/many-cores such as Intel Xeon and Xeon Phi are widely being adopted for modern large-scale supercomputing systems. The architectural features such as high core density, mesh interconnects, deeper memory hierarchies and hardware multi-threading offered by these systems provide opportunities for application developers to exploit more parallelism. However, it also poses significant challenges for the MPI runtimes to optimize communication performance. One of the major challenges involves optimizing collective communication for dense multi-/many-core processors. Traditionally, MPI runtimes have used send/recv, direct shared-memory ("double-copy") or kernel-assisted ("single-copy") mechanisms for intra-node collective communication. However, existing collective designs that are based on these mechanisms suffer from several bottlenecks such as multiple copies, per message handshake, and kernel-level lock contention that limit their performance. In this paper, we first characterize the bottlenecks associated with the aforementioned approaches in designing collectives in MPI. Then, we propose efficient "Shared-address space"-based designs to implement different MPI collectives. Finally, we show the efficacy of our approach by implementing various MPI collectives. Our proposed designs show up to 11x, 50x, 17x, and 5x performance improvement for Bcast, Scatter, Gather, and Alltoall over other state-of-the-art MPI libraries on different multi-/many-core architectures.

The recent advent of the NVLink interconnect and Peripheral Component Interconnect express (PCIe) switch has resulted in the creation of extremely dense Graphics Processing Unit (GPU) systems like Cray CS-Storm and NVIDIA DGX. In addition to the extremely high computational capability and communication capacity within a single machine, these systems expose novel capabilities like performing load-store operations from remote GPU memory across interconnects. While researchers have proposed solutions that take advantage of load-store semantics to provide support for high-performance datatype processing on CPUs, there exists no scholarly work on how one can orchestrate such high-performance datatype-based communication for GPU-resident data. In this paper, we take up this challenge and propose high-performance and architecture-aware designs for GPU-based non-contiguous datatype processing that uses the load-store semantics exposed by modern dense GPU systems. We demonstrate that the proposed solutions can reduce the overhead of datatype processing by up to 4.7X compared to the state-of-the-art schemes for GPU-based MILC communication kernel on an NVLink2-enabled dense GPU system. For a weather forecast application kernel, the proposed designs demonstrate up to 9.9X faster HaloExchange kernel among 64 GPUs over state-of-the-art designs. The proposed adaptive scheme also reports 10% higher throughput than existing designs for a 2D Jacobi solver on 16 GPUs. To the best of our knowledge, this is the first scholarly work that takes advantage of the zero-copy based load-store semantics to perform high-performance GPU to GPU derived datatype communication on modern dense GPU systems.

Derived datatypes are commonly used in MPI applications to exchange non-contiguous data among processes. However, state-of-the-art MPI libraries do not offer efficient processing of derived datatypes and often rely on packing and unpacking the data at the sender and the receiver processes. This approach incurs the cost of extra copies and increases overall communication latency. While zero-copy communication schemes have been proposed for contiguous data, applying such techniques to non-contiguous data transfers bring forth several new challenges. In this work, we address these challenges and propose FALCON - Fast and Low-overhead Communication designs for intra-node MPI derived datatypes processing. We show that the memory layouts translation of derived datatypes introduce significant overheads in the communication path and propose novel solutions to mitigate such bottlenecks. We also find that the current MPI datatype routines cannot fully take advantage of the zero-copy mechanisms, and propose enhancements to the MPI standard to address these limitations. The experimental evaluations show that our proposed designs achieve up to 3 times improved intra-node communication latency and bandwidth over state-of-the-art MPI libraries. We also evaluate our designs with communication kernels of popular scientific applications such as MILC, WRF, NAS MG, and 3D-Stencil on three different multi-/many-core architectures and show up to 5.5 times improvement over state-of-the-art designs employed by production MPI libraries.

Message Passing Interface (MPI), thus far, has remained a dominant programming model to program large-scale scientific applications. Collective communication operations in MPI are of significant importance due to their communication intensive nature and use in scientific applications. With the emergence of multi-/many-core systems and rise of deep learning applications, it is important to revisit MPI collectives, particularly MPI Allreduce to exploit vast parallelism offered by modern architectures. In this paper, we take up this challenge and propose Scalable and Adaptive designs for Large message Reduction collectives (SALaR). We focus on MPI Allreduce due to its use in deep learning frameworks and propose new designs that can significantly improve its performance by exploiting architectural features of modern multi-/many-cores in tandem with high-throughput network such as InfiniBand. We also propose a theoretical model to analyze communication and computation cost and use these insights to guide our designs. The evaluation of the proposed SALaR based designs shows significant performance gains over state-of-the-art designs on a wide variety of micro-benchmarks and applications.

State-of-the-art designs for the hierarchical reduction collective operation in MPI that work on the concept of distributed address spaces incur the cost of intermediate copies inside the MPI library to stage the data between processes. Such additional copies can severely affect the performance especially on emerging many-core architectures like Intel Xeon/Xeon Phi and OpenPOWER. In this paper, we take up this challenge and study the trade-offs involved in designing high-performance and scalable, "shared address space"-based communication primitives on top of XPMEM using basic point-to-point primitives in MPI. We then redesign the reduction collective operations using the knowledge gained from the initial studies. Our proposed designs at the collective level enable a process to offload communication and computation operations to intra-node peers without the need for additional intermediate copies resulting in a truly "zero-copy" design for MPI_Reduce and MPI_Allreduce. We further develop a theoretical model to analytically study the impact such designs can have on the performance of collective communication primitives. We evaluate the proposed designs with microbenchmarks, HPC, and Deep Learning applications on three different multi-/many-core architectures (Broadwell, Knights Landing, and OpenPOWER). The proposed designs show up to 3x improvement in latency of Reduce and Allreduce benchmarks, up to 37% improvement in the runtime of MiniAMR, and up to 19% reduction in the training time of AlexNet deep neural network compared to existing state-of-the-art MPI libraries. To the best of our knowledge, this is the first research work that studies the impact of XPMEM based shared address space designs on the performance of collective operations in a distributed memory programming model like MPI at scale.

Manycore processors such as Intel Knights Landing (KNL), the second generation Xeon Phi many-core processor from Intel comes equipped with up to 288 threads and 16 gigabytes of high-bandwidth on-chip multi-channel DRAM (MCDRAM) that bear the potential to significantly improve the performance of both compute-bound and memory-bound applications. For this potential to be realized, it is imperative to exploit KNL's highly threaded environment and careful use of the limited MCDRAM resource. In this work, we focus on achieving effective utilization of KNL's resources through the design of a kernel-based communication engine that makes use of multiple kernel threads and generic work request abstraction scheme to accelerate MPI data movement operations. Being a kernel-based approach, our designs are application-pattern agnostic and aim to have minimal contention with the application's compute and memory requirements. We have compared our proposed designs with other prevalent schemes employed by modern MPI libraries. The experimental evaluation shows that the proposed designs provide up to 2.5X improvement at the microbenchmark-level and improve the total execution time of the MPI+OpenMP version of HPCG by up to 15% when compared with other approaches. Furthermore, using the CNTK Deep Learning framework, we demonstrate a significant improvement over existing approaches in the total training time (execution time) with the Multi-level perceptron (MLP) model and MNIST image recognition dataset.

Broadcast operations (e.g. MPI_Bcast) have been widely used in deep learning applications to exchange a large amount of data among multiple graphics processing units (GPUs). Recent studies have shown that leveraging the InfiniBand hardware-based multicast (IB-MCAST) protocol can enhance scalability of GPU-based broadcast operations. However, these initial designs with IB-MCAST are not optimized for multi-source broadcast operations with large messages, which is the common communication scenario for deep learning applications. In this paper, we first model existing broadcast schemes and analyze their performance bottlenecks on GPU clusters. Then, we propose a novel broadcast design based on message streaming to better exploit IB-MCAST and NVIDIA GPUDirect RDMA (GDR) technology for efficient large message transfer operation. The proposed design can provide high overlap among multi-source broadcast operations. Experimental results show up to 68% reduction of latency compared to state-of-the-art solutions in a benchmark-level evaluation. The proposed design also shows near-constant latency for a single broadcast operation as a system grows. Furthermore, it yields up to 24% performance improvement in the popular deep learning framework, Microsoft CNTK, which uses multi-source broadcast operations; notably, the performance gains are achieved without modifications to applications. Our model validation shows that the proposed analytical model and experimental results match within a 10% range. Our model also predicts that the proposed design outperforms existing schemes for multi-source broadcast scenarios with increasing numbers of broadcast sources in large-scale GPU clusters.

Availability of large data sets like ImageNet and massively parallel computation support in modern HPC devices like NVIDIA GPUs have fueled a renewed interest in Deep Learning (DL) algorithms. This has triggered the development of DL frameworks like Caffe, Torch, TensorFlow, and CNTK. However, most DL frameworks have been limited to a single node. In order to scale out DL frameworks and bring HPC capabilities to the DL arena, we propose, S-Caffe; a scalable and distributed Caffe adaptation for modern multi-GPU clusters. With an in-depth analysis of new requirements brought forward by the DL frameworks and limitations of current communication runtimes, we present a co-design of the Caffe framework and the MVAPICH2-GDR MPI runtime. Using the co-design methodology, we modify Caffe's workflow to maximize the overlap of computation and communication with multi-stage data propagation and gradient aggregation schemes. We bring DL-Awareness to the MPI runtime by proposing a hierarchical reduction design that benefits from CUDA-Aware features and provides up to a massive 133x speedup over OpenMPI and 2.6x speedup over MVAPICH2 for 160 GPUs. S-Caffe successfully scales up to 160 K-80 GPUs for GoogLeNet (ImageNet) with a speedup of 2.5x over 32 GPUs. To the best of our knowledge, this is the first framework that scales up to 160 GPUs. Furthermore, even for single node training, S-Caffe shows an improvement of 14\% and 9\% over Nvidia's optimized Caffe for 8 and 16 GPUs, respectively. In addition, S-Caffe achieves up to 1395 samples per second for the AlexNet model, which is comparable to the performance of Microsoft CNTK.