Data Packing

First, we need to convert the data into binary format. We call this step data packing or serialization, and it depends on how we keep the data in the memory. In case our data is a C-style array, it is already in a contiguous byte buffer in the memory. If it is a table structure in a format like Apache Arrow, the data may be in multiple buffers each allocated for holding a data column of the table. In the case of Apache Arrow, there can be auxiliary arrays associated with the data arrays to hold information like null values and offsets. But the data is still in a set of contiguous buffers. In both a C-style array and an Arrow table, we can directly send the byte buffers without converting the original data structure.

If our data is in a more generic type such as a Java object, it is not in a contiguous buffer, and we may need to use an object serialization technology to convert it to a byte buffer. At the receiver, we need to unpack the data in the byte format and create the expected objects. To do this, we need information such as data types and the size of the message.

Protocol

A protocol is a contract between a sender and receiver so that they can successfully send messages and understand them. There are many messaging protocols available to suit our needs. Some are designed to work across hardware, operating systems, and programming languages. These protocols are based on open standards, with HTTP being familiar to most developers. Since such protocols are open, they sometimes present overheads for certain applications. For example, HTTP is a request-response protocol, but not every use case needs a response. The advantage of open protocols is that they allow various systems to be integrated in a transparent way.

Data-intensive frameworks tend to use internal protocols that are optimized for their use cases. The details of these protocols are not exposed to the user. Standard protocols are too expensive for the fine-grained operations occurring between the parallel processes of these frameworks.

There are two broad messaging protocols used for transferring data inside clusters. In the literature on messaging, they are called Eager Protocol and Rendezvous Protocol.

• Eager Protocol—The sender assumes there is enough memory at the receiving side for storing the messages it sends. When this assumption is made, we can send messages without synchronizing with the receiver. The Eager Protocol is more efficient at transferring small messages.
• Rendezvous Protocol—The sender first indicates to the receiver their intention to send a message. If the receiver has enough memory/processing to receive the message, it delivers an OK message back to the sender. Upon receiving this signal, the sender transmits the message. This protocol is mostly used for transferring larger messages.

The header contains the information needed by the receiver to process the message. Some common information in the header is listed here:

• Source identifier—Many sources may be sharing the same channel.
• Destination identifier—If there are many processing units at the receiver, we need an identifier to target them.
• Length—Length of the message. If the header supports variable-length information, we need to include its length.
• Data type—This helps the receiver to unpack the data.

As we can see, a header of a message can take up a substantial amount of space. If we specify the previous information as integers, the header will take up 16 bytes. If our messages are very small, we are using a considerable amount of bandwidth for the headers. After the receiver constructs the message, it can hand it over to the upper layers of the framework for further processing.

Control Messages

Data analytics frameworks use control messages for managing the applications in a distributed setting. The control messages are active in the initialization, monitoring, and teardown of the applications. Most are peer-to-peer messages, while some advanced systems have more complex forms to optimize for large-scale deployments. We touched upon control messages in an earlier chapter where we discussed resource management.

An example of a control message is a heartbeat used for detecting failures at parallel workers. In some distributed systems, parallel workers send a periodic message to indicate their active status to a central controller, and when these messages are missing, the central controller can determine whether a worker has failed.

External Data Sources

Data processing systems need to communicate with distributed storage and queuing systems to read and write data. This happens through the network as data are hosted in distributed data storages. Distributed storage such as file systems, object storages, SQL databases, and NoSQL databases provide their own client libraries so users can access data over the network. Apart from requiring the URL of these services, most APIs completely hide the networking aspect from the user.

We discussed accessing data storage in Chapter 2, so we will not go into more detail here. An example is accessing an HDFS file. In this case, the user is given a file system client to read or write a file in HDFS where the user does not even know whether that data is transferred through the network. In the HDFS case, the API provided is a file API. Databases offer either SQL interfaces or fluent APIs to access them. Message queues are a little different from other storage systems since they provide APIs that expose the messaging to the user. Reading and writing data to message queues is straightforward.

Data Transfer Messages

Once an application starts and data is read, the parallel workers need to communicate from time to time to synchronize their state. Let us take the popular big data example of counting words. To make things simple, we will be counting the total number of words instead of counting for individual words. Now, assume that we have a set of large text files stored in a distributed file system and only a few machines are available to perform this calculation.

Our first step would be to start a set of processes to do the computation. We can get the help of a resource manager as described in Chapter 3 and thereby manage a set of processes. Once we start these processes, they can read the text files in parallel. Each process will read a separate set of text files and count the number of words in them.

We will have to sum the word counts in different processes to get the overall total. We can use the network to send the counts to a single process, as shown in Figure 6-2. This can be achieved with a TCP library. First, we designate a process that is going to receive the counts. Then we establish a TCP connection to the designated receive process and send the counts as messages. Once the designated process receives messages from all other processes, it can calculate the total sums.

The previous example uses point-to-point messages, but collectively they are doing a summation of values distributed in a set of processes. A distributed framework can provide point-to-point messages as well as higher-level communication abstractions we call distributed operators.

• Point-to-point messages—This is the most basic form of messaging possible in distributed applications. The example we described earlier was implemented using point-to-point messages. These are found in programs that need fine-grained control over how parallel processes communicate.
• Distributed operations—Frameworks implement common communication patterns as distributed operations. In modern data-intensive frameworks, distributed operations play a vital role in performance and usability. They are sometimes called collective operations [1].

Most data-intensive frameworks do not provide point-to-point message abstractions between the parallel processes. Instead, they expose only distributed operations. Because of this, in the next few sections, we will focus our attention on distributed operations.

Task Graph

Distributed operations can be embedded into a task graph–based program as a link between two tasks. Figure 6-3 details a computation expressed as a graph with a source task connected to a task receiving the reduced values through a Reduce link.

At runtime, this graph is transformed into one that executes on a distributed set of resources. Sources and reduced tasks are user-defined functions. The Reduce link between these nodes represents the distributed operation between them. The framework schedules the source and targets of the distributed operation in the resources according to the scheduling information. Figure 6-3 illustrates the transformed graph with four source task instances connected to a Reduce task instance by the Reduce distributed operation.

As we discussed in Chapter 5, the task graph–based execution is used in programs with a distributed data abstraction such as Apache Spark. The following Apache Spark example uses the Resilient Distributed Data (RDD) API to demonstrate a Reduce operation.

In this example, we have an array and partition it to 2 (parallelism). Then we do a Reduce operation on the data in the array. Note that Spark by default does the summation over all elements of the array and produces a single integer value as the sum. Spark can choose to run this program in two nodes of the cluster or in the same process depending on how we configure it. This API allows Spark to hide the distributed nature of the operation from the users.

import java.util.Arrays;
 
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
 
public class SparkReduce {
 public static void main(String[] args) throws Exception {
   JavaSparkContext sc = new JavaSparkContext();
 
   // create a list and parallelize it to 2
   JavaRDD<Integer> list = sc.parallelize(
       Arrays.asList(4, 1, 3, 4, 5, 7, 7, 8, 9, 10), 2);
 
   // get the sum
   Integer sum = list.reduce((accum, n) -> (accum + n));
 
   System.out.println("Sum: " + sum);
 }
}

Parallel Processes

As previously described, the MPI implementations provide distributed operations so that we can write our program as a set of parallel processes. Table 6-1 shows a popular implementation of the MPI standard [2]. The standard provides APIs for point-to-point communication as well as array-based distributed operations called collectives.

With the default MPI model, the user programs a process instead of a dataflow graph. The distributed operations are programmed explicitly by the user between these processes in the data analytics application. Computations are carried out as if they are local computations identifying the parallelism of the program. This model is synonymous with the popular bulk synchronous parallel (BSP) programming model. In Figure 6-4, we see there are computation phases and distributed operations for synchronizing state among processes.

In the following source code, we have an MPI_reduce operation. Every parallel process generates a random integer array, and a single reduced array is produced at Process 0. The target is specified in the operation (in this case the 0th process), and the data type is set to Integer. Since the operation uses the MPI_COMM_WORLD , it will use all the parallel processes as sources. This API is a little more involved than the Spark version in terms of the parameters provided to the operation. Also, the semantics are different compared to Spark as the MPI program does an element-wise reduction of the arrays in multiple processes.

 
#include <stdio.h>
#include <stdlib.h>
#include <mpi.h>
#include <assert.h>
#include <time.h>
 
int main(int argc, char** argv) {
 MPI_Init(NULL, NULL);
 
 int world_rank;
 MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);
 
 int int_array[4];
 int i;
 printf("Process %d array: ", world_rank);
 for (i = 0; i < 4; i++) {
   int_array[i] = rand() % 100;
   printf("%d ", int_array[i]);
 }
 printf("
");
 
 int reduced[4] = {0};
 // Reduce each value in array to process 0
 MPI_Reduce(&int_array, &reduced, 4, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
 
 // Print the result
 if (world_rank == 0) {
   printf("Reduced array: ");
   for (i = 0; i < 4; i++) {
     printf("%d ", reduced[i]);
   }
   printf("
");
 }
 
 MPI_Finalize();
}

With MPI's parallel process management and the ability to communicate between them, we can build various models of distributed computing. We can even develop task graph–based executions using the functions provided by MPI [3]. Since MPI provides only the basic and most essential functions for parallel computing, its implementations are efficient. Even on large clusters, microsecond latencies are observed for distributed operations.

On the other hand, the simplistic nature of these functions creates a barrier for new users. Most MPI implementations lack enterprise features such as gathering statistics and web UIs, which in turn make them unattractive for mainstream users. With deep learning frameworks, MPI is increasingly used for training larger models and continues to gain momentum in data analytics.

Data Abstractions

The semantics of a distributor operation depend on the data abstractions and application domain they represent. As discussed in Chapter 4, arrays and tables are the two most common structures used in data-intensive applications.

• Arrays—We use arrays to represent mathematical structures such as vectors, matrices, and tensors. Distributed operations on arrays define the common communication patterns used in parallel calculations over these mathematical structures.
• Tables—The operations associated with tables originate from relational algebra. Distributed operations on tables include these relational algebra operations and common data transfer patterns such as shuffle.

Distributed Operation API

As distributed operations are built on messaging, they need the information we discussed for messaging and more to produce higher-level functionality.

Streaming Operations

Streaming systems are always modeled as task graphs, and they run as long-running tasks and communication links. So, the streaming operations are running continuously, connecting the tasks by taking input from source tasks, and producing output to the receiving tasks.

Batch Operations

Two variants of batch operations are possible depending on whether the operations are designed to process in-memory data or large data.

• In-memory data—For in-memory data, an operation can take the complete data at once and produce the output. This simplifies the operation API. For example, array-based operations in MPI are designed around this idea.
• Large data—To handle large data that does not fit into the memory, the operation needs to take in data piece by piece and produce output the same way. The operation design is much more involved, as it needs to take partial input, process this using external storage, and take more input until it reaches the end. At the end, it still needs to produce output one chunk at a time.

Broadcast

The simplest approach to implement Broadcast is to create N connections to the target processes and send the value through. In this model, if it takes time to put the Broadcast value into the network link, it will take times to completely send the value to all N processes. Figure 6-12 illustrates this approach. It is termed a flat tree as it forms a tree with the source at the root and all the other workers at the leaves. This method works better if N is small and takes much longer when either N or the message size increases.

There are methods that can perform much better for many processes. One such option is to arrange the processes in a binary tree, as shown in Figure 6-13, and send the values through the tree. The source sends the value to two targets; these in turn simultaneously send the value to four targets, which transfer the value to eight targets, and so on until it reaches all targets. The parallelism of the operation increases exponentially, and as it expands, it uses more of the total available bandwidth of the network. Theoretically, it takes about steps to broadcast the values to all the nodes. When N is large, this can reduce the time required significantly.

The tree approach works well only for smaller values that do not utilize the network fully. In the case of large messages, there is still a bottleneck in this approach since every node other than the leaves tries to send the value to two processes at the same time. The operation can be further optimized for larger data by using methods such as chaining and double trees. It is important to note that some computer networks have built-in capabilities to broadcast values, and implementations sometimes exploit these features.

Reduce

Depending on the message sizes, different routing algorithms can be utilized to optimize the performance of the Reduce operation. For smaller values, routing based on tree structures is optimal. The tree-based routing algorithms work by arranging the participating processes in an inverted tree. The root of the tree is the target that receives the final reduced value, and the rest of the tree are the sources producing the data to be reduced. As in the broadcast case, the reduction can be done in log N steps for N parallel processes.

AllReduce

The simplest form of AllReduce implementation is a Reduce operation followed by a Broadcast. This works well for small messages. When the message size is large, however, the network is not fully utilized. To optimize the network bandwidth utilization, a ring-based routing algorithm or recursive doubling algorithm can be used.

Ring Algorithm

In this version, data from each process is sent around a virtual ring created by the i^th process connecting to the where is the number of processes. In the first step, each process sends its data to its connected process. In the next step, each process sends the data it received in the previous step to its connected process. This is done for steps. The algorithm fully utilizes the available network bandwidth by activating all the network links equally at every step. It is best suited for large data reductions. Figure 6-14 shows a reduced set of values in four processes in three steps.

Recursive Doubling

Figure 6-15 shows the recursive doubling algorithm. Each process will start by sending data to a process at distance 1. For example, process 0 will send data to process 1. At each step , this distance will double until steps are completed. This algorithm takes steps to communicate the data between the processes. Figure 6-15 demonstrates how this algorithm works for four processes. The algorithm has better latency characteristics than the previous one but does not utilize the network as efficiently.

Gather and AllGather Collective Algorithms

Gather collective algorithms are inverted versions of the Broadcast algorithms. We can use flat trees or binary trees for small message sizes. Gather produces a large dataset when data size in individual processes or the number of parallel processes increases. Chaining algorithms are available for larger messages.

For AllGather operations, Gather followed by Broadcast is a viable option for small messages with a small number of processes. AllGather can transfer larger messages as it needs to distribute the gathered value to all the other processes. For these cases, one can use the ring algorithm or the recursive doubling algorithm described in the AllReduce.

Scatter and AllToAll Collective Algorithms

Scatter utilizes collective algorithms found in Broadcast for sending the values present in a single process to multiple processes. AllToAll can involve many communications as a result of sending data from each process to all the other processes. It can be optimized using a chaining algorithm. This algorithm works by arranging the processes in a dynamic chain configuration. At the k^th step, the process sends data to the where is the number of processes. We will discuss this algorithm later along with shuffle.

Partitioning Data

For larger datasets, partitioning can take a considerable amount of time. The common partitioning schemes used in data-intensive frameworks are hash partitioning and radix partitioning.

Hash Partitioning

Assume we are going to create partitions from a dataset, and these are numbered from to . We can apply a hashing function to the attribute value we are using to partition the data. The new partition of a record with attribute value can be calculated by the following:

This operation can be time-consuming when large numbers of records need to be processed. If we have multiple attributes, we can use those values in the hash function. A hash function such as murmur or modulo [6] can perform the hashing.

histogram[input >> (32 - N)]++

Handling Large Data

Handling data larger than memory means we cannot keep all the data at once in memory. Therefore, we need to design the shuffle operation to take a stream of inputs to handle data larger than memory. Also, its output should be streamed out piece by piece. Since we are taking a stream of inputs, we can keep only part of the data in the memory at any instant.

To keep the parts that are not in memory, shuffle operation needs to tap into disk storage. Disks are orders of magnitude slower than the main memory and cheaper per storage unit by comparison.

The code works by reading a part of the data into memory and inserting that into the shuffle operation, as shown in Figure 6-17. The shuffle operation creates partitions and send them to the correct destinations. Once the receivers have these partitions, they can write them to disk so that they have enough memory to handle the next batch. Then we read the proceeding partitions from the disk and repeat the same until all the data are read from the disk.

At this point, the operation can start outputting data in a streaming fashion one partition at a time. Shuffle has an option to sort records based on a key associated with that record. When sorting is enabled, an external sorting algorithm can be used. For example, at the receiver we can save data to the disk after sorting. When reading, we merge the sorted partition data.

Next, we will look at a few algorithms for implementing shuffle operation.

Fetch-Based Algorithm (Asynchronous Algorithm)

The fetch-based algorithm is a popular approach adopted by current big data systems that adopts asynchronous execution. We will discuss it in Chapter 7. Figure 6-18 illustrates how the input data to the operation is split into buffers. When the buffer becomes full, the data is spilled to the disk. To achieve maximum performance from a disk, read and write needs to be done with larger chunks and on contiguous regions.

The sources update a central server with the data location of partitions, and the targets get to know about the data to receive from this server. Bear in mind central synchronization does hold the risk of introducing potential performance bottlenecks when there are many parallel workers trying to synchronize all at once. After the target fetches data from multiple sources, they merge them utilizing the disk.

Distributed Synchronization Algorithm

The shuffle can be implemented as a distributed synchronization algorithm without a single point for coordination, as shown in Figure 6-19. This algorithm works by arranging the tasks in a chain configuration. The algorithm is self-synchronizing and can be used in high-performance clusters.

The advantage of this is that it does not have the problem of overloading a single node with multiple senders transmitting to the same location. Also, it can utilize all the networking links fully without trying to send the messages through the same networking link at any given time. Unfortunately, it requires every node to participate in communication in synchronization with each other. If one node becomes slow, it can affect the whole algorithm. Such slowdowns can be due to factors such as garbage collection, network congestion, and operating system noise.

Shuffle deals with substantial amounts of data over phases of computation, communication, and disk I/O. By overlapping these phases, resources can be utilized to the fullest and improve the throughput. Most implementations found in big data systems support such overlapping to various degrees.

Join Algorithms

In most databases, the join algorithm occurs in a single machine, whereas with big data systems the joins take place in a distributed setting at a much larger scale. It is important to understand the mechanics of a join in a single machine to apply it on a larger scale. Hash join and sorted merge join algorithms are the most popular. Many frameworks implement both as they can be efficient for several types of joins with varying parameters. Depending on the situation, the correct join can be selected by the framework or the user.

Sorted Merge Join

The naive approach to joining two datasets is to take one record from a dataset and compare it against the entire next relationship to find if there are records matching the join condition. The pseudocode for this is shown in the following with datasets A and B:

For a in A
     For b in B
         If a and b matches join condition
            Add to output a,b

This approach needs a comparison, which can become huge as the size of the datasets increases. Imagine joining a million tuples against another million tuples. The previous algorithm requires 1,000 billion comparisons to complete. To avoid that many comparisons, the datasets can be sorted using the joining keys. The sorting can be done in time, and after this one scan through both datasets, it can join them together. If the number of records in relation A is N and the number of records in relation B is M, the complexity will be .

The algorithm is shown next and can be described in just a few steps. Initially, both datasets are sorted with the join key. Note that this algorithm works only for equijoins. The algorithm defines a function called advance that scans a given dataset starting with a given index. This function outputs the records with a key equivalent to the starting index of the scan. Since the datasets are sorted, it only needs to look at the consecutive elements. The moment it sees anything different, it can return the current set.

function sortMergeJoin(dataset L, dataset R, key k)
    dataset output = {}
    // sort L and R with key k
    list lSorted = sort(L, k) 
    list rSorted = sort(R, k)
    key lKey, rKey
    set lSubset, rSubset 
    advance(lSubset, lSorted, lKey, k)
    advance(rSubset, rSorted, rKey, k)
    while not empty(lSubset) and not empty(rightSubset)
        if lKey == rKey 
            add cartesian product of lSubset and rSubset to output
            advance(lSubset, lSorted, lKey, k)
            advance(rSubset, rSorted, rKey, k)
        else if left_key < right_key
           advance(lSubset, lSorted, lKey, k)
        else 
           advance(rSubset, rSorted, rKey, k)
    return output
function advance(subset out, sortedData inout, key out, k in)
    key = sortedData[0].key(k)
    subset = {}
    while not empty(sortedData) and sortedData[0].key(k) == key
        subset.insert(sorted[0])
        sortedData.remove(0)

After the advance function finds records for each dataset, if the keys are equal, the Cartesian product is taken of the two outputs of the advance, and then the advance function is run on both datasets. If the keys of a dataset are less than the other dataset, the algorithm runs the advance routine on the dataset with the lesser key starting from the previous end index. This assumes an ascending order sorting of the relationships. The algorithm does this until it reaches the end of one dataset. Figure 6-22 shows two relationships with two fields per record. The first field is taken as the joining key.

The output of the equijoin will be as follows:

[3,5,1], [3,5,2], [3,5,3], [3,6,1], [3,6,2], [3,6,3], [5,1,1], [5,0,1], [8,0,1], [8,9,1].

Notice how it takes the Cartesian product of the matching records of the two datasets. This algorithm works much more efficiently compared to two sets done naively for larger datasets. Sorting an enormous number of records can take a significant amount of time. When data larger than memory is involved, an external sorting algorithm is of great help.

Distributed Joins

The idea behind the distributed join is simple. It first redistributes data with the same keys from both datasets into the same processes. After this, the data in each process is joined separately using one of the join algorithms mentioned earlier [8]. Figure 6-23 shows an example of such where Process 0 and Process 1 has two tables 0 and 1. Process 0 has the 0^th partition of these tables, and Process 1 has the 1^st partition. First the data is shuffled so that Keys 1 and 2 go to Process 0 and Keys 3 and 4 are sent to Process 1. After the shuffle, the tables are joined locally.

Performance of Joins

We described the basic structure of a join here, and in practice they are implemented to efficiently use the cache and reduce the TLB misses. First, the tables are partitioned into small datasets. This is done so that individual partitions fit into the cache to make sorting or hashing efficient.

A join can produce a dataset that is many times bigger than the input datasets. In classic implementations, the hash join is slightly better than the sort-merge join. Sort-merge join is a more popular implementation as it can handle all the join scenarios. The performance of these two algorithms depends on the implementation and programming environment.

The performance evaluations of the algorithms found in research literature are done mostly in ideal conditions. In practice, factors such as data representation in the memory and size of the data and whether disks are used can determine which implementation is faster. With such a complex set of circumstances, it is better to consult the performance numbers of specific frameworks and their recommendations to figure out the correct algorithm for certain situations.

There are many other optimizations available as well. For example, if one relationship is comparatively small, instead of reshuffling both tables, we can send the small table to all the processes and do the join at each process. This avoids the shuffling of the larger table that can take a considerable time.

Memory Considerations

Memory management is key to efficient messaging, as most of the overhead occurs when in-memory user-defined data types are transferred to the network buffers. This is especially problematic for higher-level languages such as Java and Python, where objects are created in heap memory. When numerous small objects are made in heap memory, transferring them to the network buffers can be costly due to many random memory accesses that occur.

Java-based frameworks use object serialization technologies such as Kryo or Avro to serialize user-defined objects into byte format. For these applications, serialization incurs a significant overhead when large numbers of small messages are transferred.

On the other hand, high-performance frameworks restrict the users to define data types that fit into the networking buffers without additional processing. For example, compact formats like Apache Arrow use buffers that can be directly applied in network transfers. This is the most efficient form of designing applications as there is no need to pack the user-defined object into a buffer.

Message Coalescing

Small messages in the network lead to the underutilization of network bandwidth. To fully take advantage of the network, messages must be as big as possible. Message size is dictated by the applications, and frameworks can coalesce or group these small messages to increase network bandwidth use.

Sending groups of messages increases bandwidth utilization, but individual messages can observe higher latencies. This effect is more important for streaming applications where latency is critical. Also, message coalescing requires the use of CPU and additional memory.

Compression

Message compression can be applied to reduce the number of bytes transferred through the network. There are many compression schemes available with varying degrees of compression capabilities. Usually when a compression ratio of an algorithm is high, it demands more computing power to operate. A user can choose the algorithm based on the availability of spare CPU power. Since compression is a compute-intensive operation, it is not suitable for every situation.

Compression should be used when a data transfer is bounded by the network bandwidth and there are spare CPU cycles. Another situation where compression can be used is when network transfers incur monetary costs compared to using the CPU.

When records are small, it is harder for the communication algorithms to utilize the full bandwidth of the network. Larger size records can saturate the network, so we may consider compression algorithms in such cases. We should note that small messages can be combined to form larger messages.

Blocking Operations

When invoked, blocking operations wait until a network operation fully completes. The biggest issue with blocking operations comes from the fact that network operations work with distributed resources and can take considerable time to complete.

When the network operation is the only task the program has to do, there is no harm in waiting. But this is a rare circumstance in a complex program. Usually a program needs to communicate with multiple entities simultaneously, and they have other tasks that can complete while network operations are pending. The shuffle operation is one such application, where the targets receive multiple values from various sources. While they obtain values from the source, the program can sort and store the already received data into the disk.

If we are working with blocking operations, we need to use separate threads for doing other tasks while some threads wait for network operations. Spawning threads unnecessarily can seriously affect the performance of a program, especially when there are many of them. Most data-intensive applications use a conservative approach for creating threads and try to keep the thread count close to the number of CPUs assigned to a process. To achieve this objective, they have no choice but to use nonblocking APIs for communication.

Nonblocking Operations

A nonblocking operation does not have the requirement of creating separate threads for handling each concurrent connection. A carefully written nonblocking networking program can use up to N number of threads for handling N connections. In a nonblocking API, calls such as read or write do not wait for completion. Because of this, we need to check periodically to see whether there is data ready in the network channels. This checking is called a progression loop and can be found in nonblocking network APIs.

Once an operation has begun, unless the progression loop is invoked, the network will not advance. Such progression loops allow programmers to control the threads used for networking operations. This applies from low-level to high-level APIs offered for handling messages. The following is an example of how this principle can be applied to a distributed operation AllReduce:

// this call returns immediately without finishing,  
// it returns a reference to the pending nonblocking operation
R = IAllReduce(dataset) 
// loop until the operation is complete, we use the reference to check
While (not isComplete(R)) {
   // do other work   
}

The isComplete call is necessary as it performs the network handling and the completion of the operation. Without this, the operation cannot progress as there is no other thread to check on the network. Various frameworks provide different names for the isComplete method such as test , probe , select , etc. But they all achieve the same goal of checking whether the operation is complete and proceeding with the underlying network operations.