Dynamic reconfiguration of noniterative scientific applications: A case study with HPG aligner

3.1. Overview of the DMR API

A dynamic reconfiguration system relies on two main components: an RMS and a parallel runtime. In particular, the DMR API depends on the OmpSs ⁸ runtime (called Nanos++) and Slurm (Yoo et al., 2003) workload manager. As shown in Figure 1, the parallel runtime supports the execution of the application and allows it to inform the RMS that it can be resized by issuing a check status request. On the other hand, the RMS is aware of the resource utilization and the queue of pending jobs. When the RMS receives a check status request, it inspects the global status of the system, decides whether any rescaling action should be done, and informs of this decision to the parallel runtime. If a rescale action is due, the RMS, the parallel runtime, and the application will collaborate to continue the application execution with a different number of processes.

OPEN IN VIEWER

Figure 1.

Communication scheme of the DMR API reconfiguration. DMR API: Dynamic Management of Resources Application Programming Interface.

A general schema of a malleable job using the DMR API is depicted in Listing 1. Once the data have been initialized (line 2), the main loop is entered (line 7). At the beginning of each main loop iteration, the application calls the DMR API dmr_check_status() function (line 8) to inform the resource manager that a resizing can be performed. If no action is returned, the execution will continue normally (line 13). Else, if the RMS responds with a resizing action, either expand or shrink, a handler of a new communicator and the number of processes created in that communicator (handlerNProcs) will be returned. Using this information, the application will set the data dependencies and the communication pattern in order to redistribute the data among the new processes (line 10).

OPEN IN VIEWER

Listitng 1.

General schema of a malleable application using the DMR API.

3.2. Overview of HPG aligner

HPG aligner is a bioinformatics application for fast and accurate mapping of RNA sequences on a cluster of computers (Martínez et al., 2013). It employs several MPI processes, which exploit node-level parallelism through portable operating system interface (POSIX) threads. One of these MPI multithreaded processes is configured as a writer, and the others as workers (see Figure 2). Each worker operates over a particular part of the input file, composed by short RNA fragments (reads) produced by a next generation sequencing (NGS) sequencer. At the beginning of the execution, each worker calculates the indexes of its part of the input file. Then, each worker performs its computation over its self-assigned reads and sends the alignments it obtains to the writer. Meanwhile, the writer saves the reported alignments to disk.

OPEN IN VIEWER

Figure 2.

HPG aligner original version workflow.

4.1. Adapting the HPG aligner workflow

The original implementation of HPG aligner performs a static distribution among all the processes of the whole data set (i.e. all the RNA reads obtained from an NGS sequencer). This strategy is not desirable for a malleable application, since it would complicate both the redistribution of the yet-to-be-done work among the new processes and the determination of the synchronization point that all the processes must reach to evaluate whether a reconfiguration should be performed. Thus, we have redesigned the HPG aligner workflow in order to split the input workload into a user-defined number of chunks, which will be dynamically assigned to the worker processes.

To allow HPG aligner to dynamically distribute the data set, a new process is created, the manager, which is in charge of the distribution of the input data set chunks among the worker processes. In this new workflow, in Figure 3, when a worker has finished its previously assigned work, it asks the manager for more work. Then, the manager computes the index of the next data set chunk to be processed and assigns it to that worker.

OPEN IN VIEWER

Figure 3.

HPG aligner malleable version workflow.

Moreover, the HPG aligner workflow has been also modified in order to allow the processes to periodically reach a malleability point, that is, a synchronization point where the RMS can be asked if a reconfiguration should take place. This may be attained by instructing all the HPG aligner processes to reach the malleability point after a given number, n, of chunks have been processed. Figure 3 illustrates this iterative schema: All the HPG aligner processes cooperate to compute n chunks of reads, and then all of them reach the malleability point, where a reconfiguration could be triggered. These two steps, processing and reaching the malleability point, are repeated until all the chunks have been processed.

To implement this iterative schema and synchronize all the processes at a malleability point, the next strategy and communication schema have been employed. After dispatching n chunks, the manager stops distributing work. Instead, whenever a worker asks for more work, the manager signals to proceed to the malleability point. The worker, for its part, after receiving this signal, propagates it to the writer. This communication schema, see Figure 4, ensures that all processes will eventually reach the malleability point after n chunks are processed.

OPEN IN VIEWER

Figure 4.

Communication schema of the processes in the HPG aligner malleable version.

The algorithms used to implement the described strategy and communication schema are depicted next using an MPI pseudo-code. In this pseudo-code, the first argument of the MPI send calls is the data to be transferred, and the second, its destination. Likewise, the first argument of the MPI receive calls is a buffer for the data to be received, and the second argument, the senders from whom that data are received.

Algorithm 1 shows the pseudo-code corresponding to a manager iteration. While there are chunks to be processed (line 1), the manager waits for a work petition from any worker (line 2), and in answer to each of these, it provides the index of the next chunk to be processed (line 3). When all the chunks of the current iteration have been assigned, a special signal (a − 1) is sent to all the workers (lines 4–6) in order to synchronize them at the malleability point (line 7).

OPEN IN VIEWER

Algorithm 1.

Pseudo-code of a manager iteration.

Algorithm 2 describes the pseudo-code of a worker iteration. First, the worker informs the manager that it is ready to process more work (line 2). Then, it waits until the manager sends back the index of the next chunk of reads assigned to it (line 3). If a − 1 index is received (lines 4–6), it will propagate this signal to the writer and proceed to the malleability point (line 9). Otherwise, it will process the assigned chunk and send its results to the writer.

OPEN IN VIEWER

Algorithm 2.

Pseudo-code of a worker iteration.

Finally, Algorithm 3 describes the pseudo-code of the writer iteration. The writer listens for data from any worker (line 2). If a − 1 is received, a counter is increased, and when the value of this counter reaches the number of workers, it will proceed to the malleability point (lines 3–6). Otherwise, if an actual result is received, it will write it to disk (line 8).

OPEN IN VIEWER

Algorithm 3.

Pseudo-code of the writer iteration.

4.2. HPG aligner data redistribution patterns

In the previous section, a malleable workflow was proposed for HPG aligner, so that it could periodically reach a malleability point to ask the RMS whether it should be reconfigured or not before continuing processing the input data. In this section, we will discuss how the data of the HPG aligner processes could be redistributed in the event of a reconfiguration.

HPG aligner reconfigurations consider two data structures in their data redistributions. To begin with, all the HPG aligner processes share some identical information (common data). Furthermore, HPG aligner workers, during their execution, keep the mapping information to the reference genome (worker data). This data structure is used by the HPG aligner workers to improve their performance and their mapping quality. It is worth notice that although the workers do not require their particular worker data to be initially populated, HPG aligner uses the aggregated data from all the workers in a later stage in order to be able to correctly align those reads that could not be previously aligned.

Common data and worker data, hereafter referred as cData and wData respectively, must be distributed from the current processes to the new processes. For this reason, we have defined two different patterns depending on the reconfiguration type. If the reconfiguration is an expansion, that is, the application will be executed on more processes after the reconfiguration, the next strategy is followed:

Each initial w_i worker sends both its cData and wData to the new w_i′ worker.

Figure 5 illustrates this redistribution pattern when expanding from 4 to 6 processes. The initial workers send their cData and wData to the peer MPI ranks in the new communicator (wide arrows in the figure). At the same time, the initial manager and writer send their cData to the remaining newly spawned processes (narrow arrows in the figure).

OPEN IN VIEWER

Figure 5.

Data redistribution when expanding from 4 to 6 processes.

Each initial worker preserves its wData by sending it to another worker in the new communicator. Since, after a reconfiguration, the number of workers changes, not all the new workers will receive wData. We proceed in this manner as we have experimentally determined that there was no appreciable difference in HPG aligner performance or accuracy if some new workers did not have previous wData. In fact, distributing the same wData to more than one new process incurred a performance penalty because of the communication overhead.

When the action is a shrinking, the application will be executed on fewer processes after the reconfiguration. In this case, the already populated wData of those workers that will be removed should be preserved (as this information is paramount to a later stage of HPG aligner). HPG aligner implements an efficient parallel merge (Medina et al., 2016), which uses a minimum spanning tree pattern to merge all the wData into the first worker (see Figure 6). Therefore, over the different options that could be employed to preserve and distribute this information, we have chosen to first call HPG aligner parallel merge, and then follow the next strategy:

The first worker, w₀, will send its cData and wData to the new first worker, w_0′. (The aggregated wData will be preserved on this worker.)

OPEN IN VIEWER

Figure 6.

HPG aligner parallel merge of the worker data structures.

Figure 7 illustrates the redistribution pattern when shrinking from 6 to 4 processes. The first worker sends its cData and wData to the MPI rank 0 in the new communicator (wide arrow in the figure), and the remaining processes send their cData to their peer MPI ranks (narrow arrows in the figure).

OPEN IN VIEWER

Figure 7.

Data redistribution when shrinking from 6 to 4 processes.

4.3. HPG aligner malleable version outline using the DMR API

Once the HPG aligner modified workflow has been presented and the data redistribution patterns have been defined, we outline the HPG aligner malleable main loop, which proceeds as follows (see Listing 2). When a process finishes its part of processing the first n chunks (line 5), it will reach the malleability point and call the DMR API dmr_check_status( ) function. If no resize action is planned, the current processes will proceed with the next n chunks. Else, if an expanding or shrinking action is scheduled, the corresponding data redistribution pattern, described in the previous section, is applied. Since the actual data redistribution is performed by the DMR API runtime, all that is required is to indicate in the source code which data have to be distributed and to whom using the DMR API OmpSs-like #pragma directive. After the data redistribution, the new processes will proceed with the next n chunks.

OPEN IN VIEWER

Listing 2.

HPG aligner malleable version outline using the DMR API.

Notice that the dmr_check_status() function allows defining some malleability conditions that the RMS will consider. These are (i) the minimum number of processes (MIN), (ii) the maximum number of processes (MAX), and (iii) the preferred number of processes (PREF) that should be assigned to this application. If a resize is due, the function will return: the action, the new number of processes (in handlerNProcs), and the new MPI communicator (in handler).

The #pragma omp task directives, as previously stated, are used to indicate which data have to be redistributed and to whom. For example, the directive:

#pragma omp task in(cData) onto(handler, myRank) indicates that the cData data structure has to be conveyed from the current process to the new myRank MPI process in the handler communicator.

4.4. Validation of the proposed HPG aligner malleable version

To compare the original and the malleable versions of HPG aligner outputs, we have executed both versions 10 times with the next parameters and collected their execution times and results.

The input to both versions consisted of a data set with 20 million RNA reads of 100 nucleotides (generated with benchmarker for evaluating the effectiveness of RNA-seq software (BEERS) (Grant et al., 2011)), and the reference genome GRCh37.p73 human genome. ⁹

For the malleable version, we configured the number of chunks to be equal to the number of initial workers in order to mimic a static data set distribution and to prevent any reconfiguration.

The execution times were virtually the same for both versions, with close-to-negligible time differences of 0.28%, which were due to the malleability code overhead (Iserte et al., 2018).

Since the HPG aligner output depends on the order in which the reads are processed (Martinez et al., 2015), it is not possible to obtain the same results on different parallel executions. However, we can consider that the obtained results are comparable if they present a similar accuracy. Moreover, the accuracy deviation of the results obtained by the malleable version with respect to the ones of the original version was a negligible ±0.2%, we can conclude that both outputs are comparable.

5.1. Experimental setup

The experiments were performed using the Marenostrum IV supercomputer at the Barcelona Supercomputing Center (BSC). Each cluster node integrates 2 Intel Xeon Platinum 8160 sockets (24 cores at 2.10 GHz each) for a total of 48 cores with 96GiB of RAM. The nodes are interconnected through a 100Gb/s Intel Omni-Path network.

Of the 50 nodes of a standard queue in Marenostrum IV, 1 node was used to run the Slurm manager daemon and the other 49 nodes were used to run the jobs.

The following software versions were used: MPICH 3.2, OmpSs 15.06, and Slurm 15.08.

5.2. Experiments with a production-size data set

The input data set used in these experiments is a production-size one consisting of 40 million RNA reads of 100 nucleotides, 8 GiB in total. It was generated using BEERS (Grant et al., 2011).

Since the malleable version of HPG aligner has to inform the RMS of its minimum, maximum, and preferred number of processes, we have first experimentally determined these for the already described experimental setup. For this purpose, we have launched the HPG aligner malleable version with different numbers of processes and obtained their execution times, which are shown in Table 1. In order to evaluate the fitness of each configuration, we use a heuristic metric, referred as gain slope, which, unless there is a relevant gain between a configuration and the next one, decreases rapidly as the number of required nodes is increased.

OPEN IN VIEWER

Table 1.

Execution time and gain difference of the HPG aligner malleable version when executed with an input data set of 40 million reads of 100 nucleotides for different numbers of MPI multithreaded processes.

No. of processes	No. of cores	Execution time (s)	Gain (%)
3	144	238	—
6	288	68	71.43
12	576	40	11.76
24	1152	44	−1.68
48	2304	60	−6.72

The gain slope, which is reported in the fourth column of Table 1, is computed as

t i − 1 − t i t 0

where t_i is the execution time of row i of the table. Using this metric, the minimum, preferred, and maximum number of processes are determined as follows:

The minimum number of processes is selected as the first number of processes for which the gain slope is greater than 75%. In this configuration, since there is no such case, the strictly minimum number of processes has been chosen: 3 processes (a manager, a worker, and a writer).

Regarding the job submission, jobs of the fixed workload have been submitted according to the next two premises: Users want to have their results as fast as possible, and they cannot know in advance how the cluster workload will evolve. Therefore, the fixed jobs have been submitted with 12 MPI multithreaded processes each, because with this configuration they should finish faster than if submitted with any other number of processes (as shown in Table 1). It should be noted that although on this experimental setup, using 6 processes will lead to better completion times (similar to the achieved by their malleable counterparts), in a production environment, where the other running jobs will exhibit completely different characteristics, this could be not the case.

Jobs of the malleable workload have been submitted using a Slurm range submission of 312 MPI multithreaded processes, and a preferred number of processes of 6. This way, the resource manager will decide the number of nodes that will be initially assigned to the job, favoring the upper limit whenever possible. In practice, when a malleable job is being processed, although it could be started with 12 processes, it will usually be shrunk to 6 processes (the preferred number of processes) as long as there are other jobs waiting in the queue.

These differences on how the resources are (dynamically) allocated pose an impact on the cluster average resource utilization, as can be seen in Figure 8, where the malleable workloads present a lower average resource utilization than that of their fixed counterparts. As stated previously, the jobs of a workload are submitted at different times to the queue, simulating a real scenario. Besides, nothing prevents the malleable workload to make use of all the resources available at a given moment. Therefore, the lower average resource utilization on the malleable workloads indicates that these workloads achieve a better job completion throughput than their fixed counterparts.

OPEN IN VIEWER

Figure 8.

Average resource utilization on different fixed and malleable workloads with an increasing number of jobs.

If the job execution time is to be compared, since the jobs on the malleable workloads can be executed with 6 or even 3 processes, the average execution time of these should be lower than that of their fixed counterparts. This can be seen in Figure 9, where the slowdown in the average execution time of the malleable jobs reaches 60%.

OPEN IN VIEWER

Figure 9.

Average job execution time on different fixed and malleable workloads with an increasing number of jobs.

But, when a job is submitted to a cluster, not only its execution time is taken into account; the time elapsed from the submission to the execution (waiting time) has to be also considered. Jobs are expected to be initiated earlier on a malleable workload because currently running jobs may release part of their resources in favor of the pending jobs. As expected, Figure 10 shows that the average waiting time in a malleable workload can be up to 77% lower than that in a fixed workload.

OPEN IN VIEWER

Figure 10.

Average job waiting time on different fixed and malleable workloads with an increasing number of jobs.

Considering everything, the balance is positive for the malleability workloads, since the average job completion time (i.e. the average job waiting plus execution time) can be up to 20 times faster than that for its equivalent fixed workload, as can be seen in Figure 11.

OPEN IN VIEWER

Figure 11.

Average job completion time (waiting plus execution time) on different fixed and malleable workloads with an increasing number of jobs.

Figure 12 shows the completion time (the elapsed time between the first job was submitted and the last job is ended) of the different fixed and malleable workloads. Although execution time is increased, the workload completion time is mainly affected by the jobs waiting time (as it is shown in Figure 11). The workload completion time of the malleable version is up to 16% faster than their fixed counterparts. This result is especially significant from a cluster administrator/manager point of view, as it shows how the throughput of the system is increased when malleable workloads are used.

OPEN IN VIEWER

Figure 12.

Completion time of different fixed and malleable workloads with an increasing number of jobs.

Figure 13 details the time evolution of (i) the allocated nodes (top chart), (ii) the concurrent jobs being executed (red and blue lines on top chart), and (iii) the completed jobs (bottom chart) of the 1000-job fixed and malleable workloads.

OPEN IN VIEWER

Figure 13.

Time evolution of the allocated nodes (top chart), the concurrent jobs being executed (blue and red lines on top chart), and the completed jobs (bottom chart) on the 1000 jobs fixed and malleable workloads.

The flat shape in the top chart of Figure 13 depicts how the nodes have been allocated during the execution of the fixed workload. It reveals that almost all the resources are allocated all the time. Conversely, the malleable workload presents a sawtooth pattern that finishes earlier: More resources are generally available, which allows a greater flexibility on the scheduling of new jobs.

The red and blue lines on the top chart of Figure 13 represent the concurrent running jobs on the fixed and malleable workloads, respectively. On the one hand, it can be seen, as was expected, that for the current experimental setup, the fixed workload is almost always running 4 12-node jobs at the same time. On the other hand, the malleable workload presents a greater variability, where the most seen configuration is 8 3-node jobs.

Finally, the bottom chart of Figure 13 represents the time evolution of the completed jobs on the fixed and malleable workloads. The evolution of both is overlapped until second 1000, where more jobs per second begin to be completed on the malleable version. This behavior is consistent with the observed differences on the average job completion time of both workloads types with different number of jobs (see Figure 11 again): The differences were minimum for the 100 job workloads and became greater and greater as the number of jobs in the workloads was increased. Furthermore, due to its higher throughput, the malleable workload ends before the fixed workload.

The results presented so far can be summarized in a remarkable improvement of the individual job completion time when a malleable workload is processed. Figure 14 illustrates this fact, showing how all the malleable jobs are completed in almost a constant time, while the completion time of the fixed jobs becomes more variable: The later they are queued, the larger completion time they experience.

OPEN IN VIEWER

Figure 14.

Completion time of each job of the 1000 jobs on the fixed and malleable workloads. The greater the job id, the later the job was queued.

As for the reconfiguration time cost, it is due to the execution of the next two stages: scheduling and resizing. The scheduling time depends on the number of jobs in the queue and should be negligible when compared to the resizing stage or the tasks being executed. The resizing time is the sum of the times of the next operations: spawn of processes, data redistribution, and termination of processes. Of these operations, the most costly one is data redistribution, which is determined by the amount of data to be transferred and the bandwidth of the underlying network. While data redistribution is usually in the order of seconds, the other two operations usually are in the order of milliseconds. Table 2 displays the average reconfiguration time in the 1000-job workload. For this purpose, we have calculated the average time of each stage during the execution of the 1000-job malleable workload. The results confirm that the reconfiguration time is mainly dominated by the resize stage, being the scheduling stage almost negligible.

OPEN IN VIEWER

Table 2.

Average job reconfiguration time of the 1000-job workload.

Stage	Average time (s)
Scheduling	0.015
Resize	4.372
Total	4.387

5.3. Experiments with a larger data set

The input data set used in these experiments is larger than the currently usual production sizes. We have considered this larger size as it could become the usual production size in a few years. It consists of 80 million RNA reads of 400 nucleotides, 61 GiB in total (the production-size data set described in the previous section was 8 GiB). This data set was generated using BEERS (Grant et al., 2011).

Again, since the malleable version of HPG aligner has to inform the RMS of its minimum, maximum, and preferred number of jobs, we have first experimentally determined these for this particular experimental setup. In order to do this, we have launched the HPG aligner malleable version with different number of processes and obtained their execution times, which are shown in Table 3, alongside with their gain slope. Using the gain slope metric, the minimum, preferred, and maximum number of processes are determined as detailed on the previous section, as:

Minimum number of processes: 6.

OPEN IN VIEWER

Table 3.

Execution time and gain difference of HPG aligner malleable version when executed with an input data set of 80 million reads of 400 nucleotides for different numbers of MPI multithreaded processes.

No. of processes	No. of cores	Execution time (s)	Gain (%)
3	144	1382	—
6	288	345	75.04
12	576	155	13.75
24	1152	152	0.22
48	2304	171	−1.37

Our experiments reveal again that the malleable workloads outperform their fixed counterparts. Figure 15 shows the average job completion time as the sum of the average waiting and execution times on different fixed and malleable workloads. The waiting time (blue part of the bars) is the predominant addend in all the cases, while the execution time (orange part of the bars) is far smaller in the malleable cases and almost imperceptible in the fixed workloads. In this figure, it can also be seen that the average job completion time for the malleable workloads is around half of that of their fixed counterparts. In the interest of clarity, we have only shown the workloads up to 500 jobs, which let us appreciate the difference between both averaged times.

OPEN IN VIEWER

Figure 15.

Average job completion time (waiting plus execution time) on different fixed and malleable workloads with an increasing number of jobs.

The speedups of the job completion time, of the job execution time, and of the job waiting time on malleable workloads with an increasing number of jobs over their fixed counterparts are shown in Figure 16. The speedups of the job execution time are under 1, which indicates that jobs in the malleable workloads are running slower than jobs in the fixed cases. As already discussed, this is due to the malleable jobs being usually shrunk in order to accommodate more running jobs. Nevertheless, the job waiting time speedup widely compensates the slower execution time, leading to a job completion time speedup around 2 in all the workloads.

OPEN IN VIEWER

Figure 16.

Speedups of the job completion time, of the job execution time, and of the job waiting time on malleable workloads with an increasing number of jobs over their fixed counterparts.