Anatomy of a Computer Cluster

By definition, a computer cluster is a set of machines connected to each other by a network working together to perform tasks as if they are a single powerful machine. A computer in a cluster is often termed a node. Most clusters have a few machines dedicated to executing management-related tasks that are called head nodes or management nodes. The compute nodes do the actual computations. All nodes execute their own operating systems with their own hardware such as CPUs, memory, and disks. Often the management nodes and compute nodes have different machine configurations. Depending on the type of storage used, there can be dedicated machines for storing data, as shown in Figure 3-1, or storage can be built into the compute nodes, as shown in Figure 3-2. Usually, a dedicated network is used within the nodes of the cluster.

When assembling a cluster of machines in a public cloud, the storage can derive from a storage service offered by the cloud provider. The network can be shared among other machines not present in the cluster depending on the setting. In general, the topology of the networking and the location of the VMs are not visible to the user. Figure 3-3 shows a VM cluster setup with a storage service.

SSH is the tool used to connect and execute commands in a cluster of machines. A user initially logs into a management node using an SSH connection. From there they can access the other nodes of the cluster. Depending on the access policies applied to the user, access can be restricted to other nodes. An application submitted to the cluster is called a job.

Dedicated Clusters

At this point, we have seen systems adopting two approaches to start and manage applications. In the first one, every application is considered as stand-alone without any connection to other applications. The other method involves applications sharing resources in a dedicated environment using common processes. With this approach, applications for a particular framework have dedicated access, but alternative applications using the same framework can share the resources. Let's see how this approach is used in both classic parallel frameworks and big data frameworks.

$ sudo apt install libopenmpi-dev

#include <mpi.h>
#include <stdio.h>
int main(int argc, char** argv) {
    // Initialize the MPI environment
BatchTLink    MPI_Init(NULL, NULL);
    // Get the number of processes
    int world_size;
    MPI_Comm_size(MPI_COMM_WORLD, &world_size);
    // Get the rank of the process
    int world_rank;
    MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);
    // Print off a hello world message
   printf("Hello world from process id %d out of %d processors
", world_rank, world_size);
    // Finalize the MPI environment.
    MPI_Finalize();
}

$ mpicc.openmpi -o task task.c

$ mpirun -np 4 ./task

$ Hello world from process id 1 out of 4 processors
$ Hello world from process id 0 out of 4 processors
$ Hello world from process id 2 out of 4 processors
$ Hello world from process id 3 out of 4 processors

$ cat nodes
example.com.1 slots=2
example.com.2 slots=2

mpirun -np 4 --hostfile nodes ./task

Big Data Systems

Big data frameworks are taking a mixed approach for dedicated environments nowadays. Some provide stand-alone applications, while others offer mechanisms for sharing applications in a dedicated environment. Apache Spark is one of the most widely used big data systems, so we will select it as an example.

Apache Spark cluster utilizes a master program and a set of worker programs. The master is run on a management node, while the workers run on the compute nodes. Once these programs are started, we can submit applications to the cluster. The master and workers manage the resources in the cluster to run multiple Spark applications within a dedicated environment. Figure 3-4 shows this architecture; arrows indicate communications between various components.

Spark applications have a central driver program that orchestrates the distributed execution of the application. This program needs to be run on a separate node in the cluster. It can be a compute node or another dedicated node.

Here is a small hands-on guide detailing how to start a cluster and run an example program with Apache Spark:

## download and extract a spark distribution, here we have selected 2.4.5 version
$ wget http://apache.mirrors.hoobly.com/spark/spark-2.4.5/spark-2.4.5-bin-hadoop2.7.tgz
$ tar -xvf spark-2.4.5-bin-hadoop2.7.tgz
## go into the spark extracted directory
$ cd spark-2.4.5-bin-hadoop2.7
## we need to change the slaves file to include the ip addresses of our hosts
$ cp conf/slaves.template conf/slaves
$ cat conf/slaves
localhost
# start the cluster
$ ./sbin/start-all.sh
# now execute an example program distributed with spark
$ ./bin/spark-submit --class org.apache.spark.examples.SparkPi --master spark://localhost:7077 examples/jars/spark-examples_2.11-2.4.5.jar 100

It will give a long output, and there will be a line at the end similar to the following:

$ Pi is roughly 3.1416823141682313

In the case of Spark, we started a set of servers on our machines and ran an example program on it. By adding compute node IP addresses to the conf/slaves file, we can deploy Spark on multiple machines. Spark must be installed on the same file location on all the machines, and password-less SSH needs to be enabled between them as in the OpenMPI example. If we do not specify a master URL separately, it will start the master program on the node where we ran the cluster start commands.

Shared Clusters

As one might expect, resource-sharing environments are more complicated to work with than dedicated environments. To manage and allocate resources fairly among different applications, a central resource scheduler is used. Some popular resource schedulers are Kubernetes, Apache Yarn, Slurm, and Torque. Kubernetes is a resource scheduler for containerized application deployment such as Docker. Apache Yarn is an application-level resource allocator for Hadoop applications. Slurm and Torque are user-level resource allocators for high-performance clusters including large supercomputers. Table 3-2 shows popular resource managers, their use cases, allocation types, and units.

The same examples can be executed on a cluster environment with some tweaks to the configurations and submit commands.

# Allocate a Slurm job with 4 nodes
$ salloc -N 4 
# Now run an Open MPI job on all the nodes allocated by Slurm
$ mpirun ./task

$ cat my_script.sh
#!/bin/sh
mpirun ./task
$ sbatch -N 4 my_script.sh
$ srun: jobid 1234 submitted
 

$ ./bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn --deploy-mode cluster examples/jars/spark-examples_2.11-2.4.5.jar 100

Life Cycle Steps

As we have seen in previous examples, it all starts at the command line with instructions to submit the application to the cluster. Sometimes frameworks provide UIs to submit applications. Once the command to submit an application is executed, the steps shown in Figure 3-5 are taken to distribute the application to the compute nodes. Following the application setup, a discovery step is used to figure out the details of various distributed parts of the application. The frameworks perform additional steps such as monitoring while the application executes.

Data Centers

We are living in an age where massive data centers with thousands of computers exist in different parts of the globe crunching data nonstop. They work round-the-clock producing valuable insights and solving scientific problems, executing many exaflops of computations daily and crunching petabytes of data, consuming about 3 percent of the total energy used worldwide. These massive data centers host traditional high-performance computing infrastructures, including supercomputers and public clouds. The location of a data center is critical to users and cloud providers. Without geographically close data centers, some applications cannot perform in different regions due to network performance degradation across long distances.

A data center has many components, ranging from networking, actual computers and racks, power distribution, cooling systems, backup power systems, and large-scale storage devices. Operating a data center efficiently and securely while minimizing any outages is a daunting task. Depending on the availability and other features of a data center, they are categorized into four tiers. Table 3.3 describes the different tiers and the expected availability of data centers.

Data center energy efficiency is an active research area, and even small improvements can reduce the environmental effects and save millions of dollars annually. A data center hosts computer clusters from a multitude of organizations. The individual organizations manage their computers and networking while the data center provides the overall infrastructure to host them, including connectivity to the outside world. Most data centers are built near the companies or the users they serve.

Power is one of the main expenses of running a data center. The availability of cheap power sources, especially green energy, is a big factor in choosing data center sites. Furthermore, places with low risk for natural disasters are chosen because it is costly to build systems that can withstand catastrophic events such as earthquakes or tsunamis. Cold climates can also reduce the cost of cooling, which is a big part of the overall power consumption for data centers. Table 3-4 shows some of the biggest data centers in the world and their power consumption.

Even though most of the online services we use every day are powered by data centers around the world, as users we rarely notice their existence. Application developers are also unaware of the physical locations of these data centers or how they operate. Nevertheless, they play a vital role in large-scale data analytics.

Physical Machines

Before clouds and VMs became mainstream, physical machines were the only choice for running applications. Now we need to rely on physical machines only rarely thanks to the ever-increasing popularity of public clouds, virtual machines, and containers. Physical machines are still used in HPC clusters due to performance reasons. Some high-performance applications need access to the physical hardware to perform optimally and cannot scale to hundreds of thousands of cores in virtual environments. Small-scale in-house clusters usually expose physical machines to the applications due to the complexity of maintaining private clouds.

The biggest advantage of physical machines is direct access to the hardware. This allows applications to access high-speed networking hardware, storage devices, CPU, and GPU features otherwise not available through virtualization layers. This allows applications written with special hardware features to achieve optimal performance. For mainstream computing, these features are not relevant as the cost of developing and maintaining such applications outweighs the benefits.

Network

The computer network is the backbone of a data center. It connects computers and carries data among them so applications can distribute functionality among many machines. Networks can carry data between data centers, within a data center, or simply within a rack of computers. Most data centers provide dedicated high-bandwidth connections to other data centers and designated areas. For example, the Citadel in Tahoe Reno provides 10Gbps network connections with 4 ms latency between Silicon Valley.

Within a data center, there can be many types of networks utilized by different clusters created by various companies. The machines are organized into shelves we call computer racks. A rack typically consists of many computers connected using a dedicated network switch. The network between racks is shared by multiple machines. Because of this, the networking speed is much higher in a rack than between machines of different racks.

Numerous networks are available nowadays, but the most notable currently deployed is Ethernet. In most cases, Ethernet runs at around 10Gbps, and there is hardware available that can deliver up to 100Gbps speeds. Specialty networks have been designed to run inside data centers such as InfiniBand, Intel Omni-path, and uGini. These networks provide ultra-low latencies and generally higher bandwidths compared to Ethernet. They are mostly used in high-performance computing clusters including large supercomputers and are considered a vital component in scaling applications to thousands of nodes and hundreds of thousands of cores. Because of their efficiency, more and more clusters in data centers are using such features to accelerate networking performance. Table 3-5 shows some of the fastest supercomputers in the world, their number of cores, and the network they use along with their performance.

Virtual Machines

The virtual machine (VM) was invented to facilitate application isolation and the sharing of hardware resources. Virtualization is the driving technology behind cloud computing, which is now the most dominant supplier of computation to the world. When utilizing a cloud computing environment, we will most probably work with virtual machines.

A virtual machine is a complete operating system environment either created on top of hardware directly or atop another operating system, as shown in Figure 3-6. The virtual machines we start on our laptops are running on our laptop OS. In production environments, they will be directly running on hardware using virtualization layers. The virtual machine is called a guest, and the actual machine that runs the VM is called the host.

Containers

Containers operate in a lightweight virtual environment within the Linux operating system. Figure 3-7 shows a high-level architecture of containers where they run as a thin layer on top of the host operating system. Containers are lightweight compared to VMs because they do not run their own operating system kernel. This lightweight virtual environment is created mainly using Linux kernel cgroups functionality and namespaces functionality. Control groups are used to limit and measure the resources such as CPU or memory of a set of processes. Namespaces can be used for limiting the visibility that a group of processes has in the rest of the system.

By default, when we log into a Linux system, it has only one namespace that includes all the processes and resources. Linux processes start with a single process called init , and this process begins with other processes that may in turn start even more. As such, init is the common parent of all processes creating a single hierarchy or tree, all of which share the resources and can see each other.

With cgroups we can create process hierarchies that have limits on the resources they have available. We can allocate resources such as CPUs and memory to these process groups. The resource abstractions are called subsystems, examples of which include CPU, memory, network, and input/output (I/O). These subsystems control how the processes within a cgroup access the resources.

Namespace functionality facilitates deployment of process hierarchies with isolated mount points, process IDs, and networks. With namespace functionality, processes will see only the mount points created for them or the process IDs of the group. When enough of these limits and isolations are applied, we can create a totally isolated environment that does not interact with other processes of the system. This is the approach taken by container technologies such as Docker [1] and LXC.

Processor, Random Access Memory, and Cache

The CPU is the heart of a computer. For decades, a processing chip had only a single CPU core. Now single CPU core processors are almost nonexistent, and modern computers are equipped with chips that have multiple CPU cores. These are called multicore processors. A CPU core can execute a single thread (task) of the program at a time, and when many cores are available, it can execute tasks simultaneously. This allows programs to run in parallel within a single machine. Older computers with a single CPU provided the illusion of multitasking by rapidly switching between different threads of execution. With multicore processors, many threads of executions can truly occur on their own. A CPU core needs access to important structures such as the cache and memory to execute a program. In the days when chips contained only one core, caches and memory buses were all exclusive to that lone CPU. In modern multicore processors, some of these features are shared among multiple cores, while others are exclusive to each.

Apart from the CPU, random access memory (RAM) or main memory is the most important resource for data-intensive applications because of the amount of data being processed. RAM is expensive compared to other forms of storage such as disks. A single machine can have many memory modules that are connected to the CPU through memory buses. Imagine a processor with multiple cores and a single large memory connected to the processor with a single bus. Since every core requires access to the memory, they all need to share the same memory bus, which can lead to slower than average memory access speeds. Modern processors have multiple buses connecting numerous memory modules to increase the memory access times and bandwidth.

Accessing the main memory is a costly task for a CPU core because the memory bus is shared and because of the high number of CPU cycles required even with exclusive access. To reduce this access time, a much faster memory called a cache is used between the CPU core and the main memory.

Cache

It takes multiple CPU clock cycles to read an item from the main memory and load it to a CPU register. As a result, CPUs are equipped with caches, which are faster memory modules placed closer to the cores. The purpose of the cache is to reduce the average cost (clock cycles) required to access data from the main memory. These caches prefetch the most likely memory regions the CPU may use in future instructions. The CPU core can only access the memory in its cache, and when a memory request is made, it is first loaded into the cache and then served to the CPU core. If an item is not found in a cache, the CPU needs many clock cycles to fetch it from the main memory compared to loading it from the cache.

Modern computers have three caches called L1, L2, and L3, with L1 being closest to the CPU core. Figure 3-8 shows a cache hierarchy seen in modern processors. There are separate L1 caches for instructions to execute and find data. The fastest caches are the L1 instruction cache and L1 data cache. They are usually built near the CPU core in the same chip. These caches are equipped with fast memory and are expensive to build and hence limited in capacity. Next, there is the L2 cache, which is bigger than the L1 cache, and each core has one. Finally, there is an even larger cache called L3, which is shared among the cores of the chip.

We will look at the importance of cache for data-intensive application in later chapters.

Nonuniform Memory Access

NUMA is widely used in modern computers with multiple processors. In a NUMA computer, all the available memory is not uniformly (latency) accessible from a single core. The main memory is connected to the CPU cores using data buses. Each chip of a multiprocessor NUMA computer has its own memory buses to connect with memory modules. This means some memory is local to certain cores. The main memory, along with the cores and devices that are connected closely, are called a NUMA node. Different processor architectures may have one or multiple NUMA nodes per processor. Figure 3-9 shows a two-CPU configuration with two NUMA nodes.

When a CPU core tries to access memory that is located in another NUMA node, the access can be slower than its local memory access. To utilize the memory efficiently, the locality of the memory must be considered by the programs. Note that NUMA only affects the performance of an application, and no matter how the memory is allocated, its correctness will not be impacted.

NUMA memory locality is preserved by the operating system (OS) to some extent. When a process runs on a NUMA node, the OS will allocate memory for that process on its local memory. If there is no space in the NUMA local memory, the memory will be allocated in another NUMA node, slowing down the performance. Applications are scheduled on different cores and sockets by the OS throughout its lifetime. If an application is scheduled on one NUMA node and later scheduled on another node due to memory requirements, the program's performance can lag. Operating systems provide mechanisms to attach processes to NUMA nodes to prevent such performance issues.

Resource managers offer mechanisms to preserve NUMA locality while allocating resources to applications. For high-performance applications, such configurations need to be taken into account to achieve the best performance possible.

Hard Disk

Disks allow for much cheaper permanent storage of applications. There are mechanical hard disks, solid-state drives (SSDs), and NVMe SSDs available, with mechanical hard disks being the cheapest and NVMe SSDs being the priciest. Mechanical disks are also slower compared to SSDs, while NVMe provides the best performance.

Multiple hard disks can be grouped together to form a single logical disk using Redundant Array of Independent Disk (RAID) technologies to ensure fault tolerance and increased read/write performance. Depending on the RAID configuration, different aspects such as I/O performance or fault tolerance can be given priority. Computers configured to analyze large volumes of data are equipped with many hard disks and multiple I/O controllers to fully utilize the disks. Most organizations have data that need to be archived for potential future applications. Tape drives can offer economical storage for archiving such data, at the cost of slow read performances.

When data sizes are bigger than available memory, data applications need to use the hard disks for storing temporary data of computations. It is important to configure fast disks for such operations to perform optimally. Also, when multiple processes (threads) of a data analytics application access the same hard disk, it can reduce the performance. Hard disks are optimized for sequential I/O operations, so the applications should try to avoid random I/O operations as much as possible.

GPUs

Most of us are familiar with graphic processing units thanks to video gaming. Originally GPUs were created to relieve the CPU from the burden of creating images for the monitor to display. With the gaming industry's ever-increasing demand for realistic environments inside games, more and more computing power was added to GPUs. Unlike CPUs, GPUs are equipped with hundreds of cores and thousands of hardware threads that can manipulate large chunks of data simultaneously. Eventually scientists realized the GPU's potential to do large computations efficiently for dense matrix multiplications. Dense matrix multiplication is important for training deep neural networks. Now GPUs power numerous deep learning workloads, and large data analytics tasks are executed on them regularly.

Mapping Resources to Applications

In most frameworks, applications can request specific resource requirements. Usually these are the memory and CPUs. For example, an application can say it has 12 computing tasks (Map tasks) and each of them requires 8GB of memory and 1 CPU core to execute. The framework transfers these application-specific requirements to cluster managers when executing the application.

For data analytics application tasks, it is important to adhere to stricter resource mapping like NUMA boundaries to execute efficiently. Many frameworks allow finer-grained control over how the application processes are mapped to the resources. A user can request a parallel process to bind to a single CPU throughout its execution. By binding to a CPU core, an application can avoid unnecessary OS scheduling overheads and cache misses. Linux OS uses mechanisms to map specific resources to specific processes in order to achieve this level of control. When using cluster resource managers and frameworks, the users do not need to go to the Linux details and can instead apply these policies through various configurations.

Kubernetes

Kubernetes is a resource manager specifically designed to work with containers. When set up on a cluster of computers, it can spawn and manage applications running on containers. These applications can be anything, ranging from data analytics to microservices to databases. Kubernetes provides the infrastructure to spawn containers, set up auxiliary functions such as networking, and set up distributed storage for the spawned containers to work together. Some of the most important functions of Kubernetes include the following:

• Spawning and managing containers
• Providing storage to the containers
• Handling networking among the containers
• Handling node failures by launching new containers
• Dynamic scaling of containers for applications

Kubernetes supports the container technologies Docker, containerd, and CRI-O. Docker is the first container technology and made containers famous. CRI-O is a lightweight container technology developed for Kubernetes, and containerd is a simplified container technology with an emphasis on robustness and portability.

Slurm

Slurm [2] is a widely used resource manager in high-performance clusters, including the largest supercomputers. Slurm works by allocating physical resources to an application. These can be anything, from compute nodes to CPUs or CPU cores, memory, disks, and GPUs. Slurm has a similar architecture to Kubernetes with a set of servers for managing the cluster and daemon processes to run on the cluster nodes. As shown in Figure 3-11, slurmctld is the main process that manages the cluster, and slurmdbd is for keeping the state of the cluster persistent to a database. The slurmd daemons run on the compute nodes managing the resources and the processes spawned on them.

Slurm provides an API and a set of command-line utilities wrapping these APIs to interact with it. Command-line utilities are the preferred choice for most users to interact with Slurm. Slurm supports allocations both in batch and interactive modes. In interactive mode, once resources are allocated, users can SSH to them and execute any processes they might need. In the batch mode, a distributed application is submitted up front and run by Slurm. Due to human involvement, interactive mode does not use the resources very efficiently. Many production clusters only allow batch submissions in the main compute nodes and permit interactive jobs on just a small number of nodes for debugging purposes.

Slurm has different built-in job scheduling algorithms. There are API points to define custom algorithms as well. By default, Slurm uses a first-come, first-serve job scheduling algorithm. Other algorithms such as Gang scheduling and backfill scheduling are available as well.

Yarn

Yarn was the first resource scheduler designed to work with big data systems. It originally derives from the Hadoop project. Initially Hadoop was developed with resource management, job scheduling, and a map-reduce application framework all as a single system. Because of this, Hadoop did not have a mechanism to work with external resource managers. When Hadoop was installed on a cluster, that cluster could only be used to run Hadoop applications. This was not an acceptable solution for many clusters where there were other types of applications that needed to share the same resources.

Yarn decoupled the resource management and job scheduling from map-reduce applications and evolved as a resource scheduler on its own for big data applications. MapReduce became another application that uses Yarn to allocate resources. As one might expect, Yarn has a similar architecture to Kubernetes and Slurm where there is a central server called ResourceManager and a set of daemon processes called NodeManagers that run on the worker nodes.

An application in Yarn consists of an application manager and a set of containers that runs the worker processes. A data analytics framework wanting to work in a Yarn environment should write the application manager code to acquire resources and manage the workers of its applications. Figure 3-12 shows this architecture with two application managers running in the cluster.

Similar to Slurm, Yarn has a queuing system to prioritize jobs and various scheduling objectives can be applied to the applications. One of the most popular considerations in Yarn applications is the data locality. Yarn also implements various job scheduling algorithms [3] such as fair scheduling and capacity scheduling to work in different settings.

Scheduling Policy

A scheduling policy dictates high-level requirements such as what type of resources are accessible to which users and priority of different applications. Say, for instance, we take a data analytics cluster that runs jobs each day to determine critical information for an organization. This cluster is also shared with data scientists who are doing experiments on their newest machine learning models. The organization can define a policy saying the critical jobs need to run no matter what state the cluster is in. When this policy is implemented, an experimental job run by a data scientist may be halted to run a higher priority job, or it may have to wait until the higher-priority jobs complete.

So, a scheduling policy contains a set of rules that specifies what actions need to be taken in case there are more resource requests than available.

Objective Functions

An object function determines whether a resource schedule is good or bad. Objective functions are created according to the policies specified by the organization. A cluster can run many jobs at any given time. These applications demand various resource requirements (number of processors, nodes) and execution times. At any given time, the free resources of a cluster can be scattered across many racks of nodes. Furthermore, some applications can benefit from special hardware support and placement of the worker processes. These requirements can be specified with the job scheduling algorithms to make better decisions about where to place applications. We can look at few common objective functions used in resource schedulers.

Algorithms

The goal of a scheduling algorithm is to select which jobs to run when the demand is higher than the available resources. The algorithm should select the jobs such that it creates a good schedule according to the objective functions defined. If we can run every job immediately because there are resources available, we do not need to use these algorithms.

There are many scheduling algorithms available in the literature. In practical systems, these algorithms are adapted to support various objective functions according to the applications they support.

Backfill Scheduling

The backfilling [6] algorithm tries to execute jobs that are ahead in the job queue in case it cannot execute the head of the queue due to resources not being available. When it picks a job that is down in the queue, it tries not to postpone the job at the head of the queue. This requires knowledge of the job execution times for each job.

Figure 3-13 shows the difference between FIFO and backfill scheduling. Imagine we have two resources (r) and four jobs. Jobs 1 to 4 take resources and time pairs of Job1 - {r = 1, t = 2}, Job2 - {r = 1, t = 4}, Job 3 - {r = 2, t = 3}, Job 4 - {r = 1, t = 2}. These are submitted in the order of 1 to 4. If we strictly follow the FIFO schedule, Job 4 needs to wait until Job 3 is completed. With backfill scheduling, we can start Job 4 after Job 1 completes. Note that the Job 3 start time does not change because of this.