Efficient Scalable Computing through Flexible Applications and Adaptive Workloads

Efficient Scalable Computing through Flexible Applications and Adaptive Workloads Sergio Iserte, Rafael Mayo, Enrique S. Quintana-Ort´ ı Universitat Jaume I (UJI) Castell´ o de la Plana, Spain {siserte,mayo,quintana}@uji.es Vicenc¸ Beltran, Antonio J. Pe˜ na Barcelona Supercomputing Center (BSC) Barcelona, Spain {vbeltran,antonio.pena}@bsc.es Abstract—In this paper we introduce a methodology for dynamic job reconfiguration driven by the programming model runtime in collaboration with the global resource manager. We improve the system throughput by exploiting malleability techniques (in terms of number of MPI ranks) through the reallocation of resources assigned to a job during its execution. In our proposal, the OmpSs runtime reconfigures the number of MPI ranks during the execution of an application in cooperation with the Slurm workload manager. In addition, we take advantage of OmpSs offload semantics to allow application developers deal with data redistribution. By combining these elements a job is able to expand itself in order to exploit idle nodes or be shrunk if other queued jobs could be initiated. This novel approach adapts the system workload in order to increase the throughput as well as make a smarter use of the underlying resources. Our experiments demonstrate that this approach can reduce the total execution time of a practical workload by more than 40% while reducing the amount of resources by 30%. Keywords-Job Reconfiguration; Dynamic Reallocation; Smart Resource Utilization; Scale-Malleable workload I. INTRODUCTION In real HPC facilities, applications run on shared computers where hundreds or thousands of other applications are competing for the same resources. In this scenario, applications are submitted to the system with the shape of parallel jobs that conform the current workload of a system. Adapting the workload to the infrastructure can render considerable improvements in resource utilization and global throughput. A potential approach to obtain the desired adaptivity consists in applying dynamic job reconfiguration, which devises resource usage to be potentially changed at execution time. The adaption of the workload to the target infrastructure brings benefits to both system administrators and end users. While administrators would like to see the throughput rate increased and a smarter resource utilization by the applications, end-users are the direct beneficiaries of the scale-adaptivity as they will not experience strict resource requirements on submission. Although this may prevent the application from being executed in the shortest time, a faster completion time (waiting plus execution time) will compensate for this. In order to dinamically adapt a workload to the infrastructure we need two main tools: (i) a resource manager system (RMS) capable of modifying the resources assigned to a job; and (ii) a parallel runtime to rescale an application. In our solution, we have connected these components by developing a communication layer between the RMS and the runtime. In this work we enhance the Slurm Workload Manager [1] to achieve fair dynamic resource assignment while maximizing the cluster throughput. We select Slurm because it is one of the most widely-adopted RMSs worldwide, it is open-source, portable, and highly scalable [2], [3]. To exploit a collection of distributed resources, the vast majority of the scientific applications that run on high performance clusters use the Message Passing Interface (MPI) [4], either directly or on top of programming models or libraries leveraging MPI underneath. Reconfiguration is possible in MPI applications via the MPI spawning functionality. The direct use of MPI to handle migrations, however, requires considerable effort from skilled software developers in order to manage the whole set of data transfers among processes in different communicators. For this purpose, we benefit from the recently-incorporated offload semantics of the OmpSs programming model [5] to ease the malleability process and data redistribution. In addition, we adapt the Nanos++ OmpSs runtime to interact with Slurm. We improve the Nanos++ runtime to reconfigure MPI jobs and establish direct communication with the RMS. For that, applications will expose “reconfiguring points” where, signaled by the RMS, the runtime will assist to resize the job on-the-fly. We highlight that, although we benefit from the OmpSs infrastructure and semantics for job reconfiguration, our proposal may well be leveraged as a specific-purpose library, and applications using this solution are free to implement on-node parallelism using other programming models such as OpenMP or OmpSs. Our interest in developing an Application Programming Interface (API) for Dynamic Management of Resources (DMR API) is rooted on the lack of performance of Checkpoint/Restart (C/R) mechanisms when applied to dynamic reconfiguration. Figure 1 illustrates our point via a comparison of the time needed for the non-solving stages of the N-body simulation (see Section VII-B4) using the C/R approach versus the DMR API. In particular, the labels of the “spawing” bars reveal an important increment in the cost of spawning processes for C/R with respect to the DMR API (e.g., for 48– 24 processes by a factor 63.75×), because of the need to save data to disk to be later reloaded. © 2017 IEEE. Personal use of this material is permitted. Permission from IEEE must be obtained for all other uses, in any current or future media, including reprinting/republishing this material for advertising or promotional purposes, creating new collective works, for resale or redistribution to servers or lists, or reuse of any copyrighted component of this work in other works. 0 100 200 300 400 500 600 48-12 48-24 48-48 48-12 48-24 48-48 C/R DMR API 31.4x 63.75x 77x Execution time (s) Num. of procs. (initial - resized) Initial before solving Spawning Resized after solving Fig. 1. Execution time of non-solving stages of N-body simulation. In summary, the main contribution of this paper is a mechanism to accomplish MPI malleability, based on existing components (MPI, OmpSs, and Slurm) that enhances resource usage in order to produce higher global throughput in terms of executed jobs per unit of time. To that extent, we propose (1) an extension of the OmpSs offload mechanism to deal with dynamic reconfiguration; (2) a reconfiguration policy for the RMS to decide whether a job must be expanded or shrunk; and (3) a communication protocol for the runtime to interact with the RMS, based on application-level API calls. Finally, (4) we also provide an extensive evaluation of the framework that demonstrates the benefits of our workload-aware approach. The rest of this paper is structured as follows: Section II discusses related work. Section III presents an overview of the proposed methodology. Sections IV and V present the extensions developed in the Slurm RMS and the Nanos++ runtime in order to support our programming model proposal discussed in Section VI. The analysis and experimentation involve these three sections: Section VII describes the platform and applications; Section VIII evaluates all the features implemented; and Section IX presents and analyses a realistic use case for our framework, demonstrating the benefits of deploying adaptive workloads in a production cluster. Section X outlines the conclusions and discusses future work. II. RELATED WORK In general, a job (application) can be classified in one of the following types: rigid,moldable,malleable and evolving [6]. These classes depend in the number of concurrent processes during the execution of a job, so that we collapse them into two categories: •Fixed: the number of parallel processes remains constant during the execution (rigid and moldable applications). •Flexible: the number of processes can be reconfigured onthe-fly, allowing distinct numbers of parallel processes in different parts of the execution (malleable and evolving applications) or job malleability. This action is known as dynamic reconfiguration. The first steps toward malleability were in shared-memory systems exploiting the flexibility of applications. In [7] the authors leveraged moldability together with preemptive policies, such as equi-partitioning and folding. These policies can interrupt active jobs in order to redistribute processors among the pending jobs. Checkpointing mechanisms have been used in the past to save the application state and resume its execution with a different number of processes, or simply to migrate the execution to other processes. The work in [8] explores how malleability can be used in checkpoint/restart applications. There, a checkpoint–and–reconfigure mechanism is leveraged to restart applications with a different number of processes from data stored in checkpoint files. Storing and loading checkpoint files, however, poses a nonnegligible overhead versus runtime data redistribution. In [9], the authors combine Scalable Checkpoint/Restart for MPI (SCR) [10] with the User Level Failure Migration (ULFM) MPI standard proposal for fault-tolerance [11]. A resizing mechanism based on CHARM++ is presented in [12]. The authors of that work demonstrate the benefits of resizing a job in terms of both performance and throughput, but they do not address the data redistribution problem during the resizing. The authors of [13] rely on an MPI implementation called EasyGrid AMS in order to adjust automatically the size of a job. Another similar approach is found in [14], where a performance-aware framework based on the Flex-MPI library [15] is presented. That work leverages job reconfiguration in order to expand/shrink a job targeting execution performance. For that purpose, the framework monitors execution, predicts future performance, and balances the load. In the literature we can also find several works that combine malleability with resource management. ReSHAPE [16] integrates job reconfiguration techniques with job scheduling in a framework that also considers the current performance of the execution. Complementary research using this framework analyzes its impact on individual performance and throughput in small workloads [17], [18]. That solution, however, requires all applications in the cluster to be specifically-developed to be flexible under the ReSHAPE framework. In a more recent work, they present a more in-depth study discussing the ReSHAPE behavior with a workload of 120 jobs [19]. An additional important contribution is [20], where a batch system with adaptive scheduling is presented. The authors in this paper enable the communication between the RMS Torque/Maui and Charm++ as a parallel runtime. Charm++ applications are presented as automatically malleable thanks to checkpointing. Compared with previous work, we present a holistic throughput-oriented reconfiguration mechanism based on existing software components that is compatible with unmodified non-malleable applications. Our experiments are run on a platform with 65 nodes, for large workloads with up to 400 jobs. The numbers exceed the 50- node platforms and 230- job workloads in the current state- of-the-art. Furthermore, in contrast with previous studies, we configure our workloads not only leveraging synthetic applications. III. METHODOLOGY OVERVIEW Slurm exposes an API that may be used by external software agents. We use this API from the Nanos++ OmpSs runtime in order to design the job resize mechanism. Thus, Slurm’s API allows us to resize a job following the next steps: •Job A has to be expanded 1) Submit a new job B with a dependency on the initial job A. Job B requests the number of nodes NB to be added to job A. 2) Update job B, setting its number of nodes to 0. This produces a set of NB allocated nodes which are not attached to any job. 3) Cancel job B. 4) Update job A and set its number of nodes to NA+NB. •Job A has to be shrunk 1) Update job A, setting the new number of nodes to the final size (NA is updated). After these steps, Slurm’s environment variables for job A are updated. These commands have no effect on the status of the running job, and the user remains responsible for any malleability process and data redistribution. The framework we leverage consists of two main components: the RMS and the programming model runtime. The RMS is aware of the resource utilization and the queue of pending jobs. When an application is in execution, it periodically contacts the RMS, through the runtime, communicating its rescaling willingness (to expand or shrink the current number of allocated nodes). The RMS inspects the global status of the system to decide whether to initiate any rescaling action, and communicates this decision to the runtime. If the framework determines that a rescale action is due, the RMS, the runtime, and the application will collaborate to continue the execution of the application scaled to a different number of nodes (MPI processes). IV. SLURM RECONFIGURATION POLICY We designed and developed a resource selection plug-in responsible for reconfiguration decisions. This plug-in realizes a node selection policy featuring three modes that accommodate three degrees of scheduling freedom: 1) Request an Action: Applications are allowed to “strongly suggest” a specific action. For instance, to expand the job, the user could set the “minimum” number of requested nodes to a value that is greater than the number of allocated nodes. However, Slurm will ultimately be responsible for granting the operation according to the overall system status. 2) Preferred Number of Nodes: One of the parameters that applications can convey to the RMS is their desired number of nodes to execute a specific computational stage. If the desired size corresponds to the current size, the RMS will return “no action”. If a “preference” is requested, and there is no outstanding job in the queue, the expansion can be granted up to a specified “maximum” (line 2 of Algorithm 1). Otherwise, if the desired value is different from the current allocation, (lines 6 and 10 of Algorithm 1) the RMS will try to expand or shrink the job to the preferred number of nodes. 3) Wide Optimization: The cases not covered by the preceding methods (Algorithm 1, line 13) are handled as follows: A job is expanded if there are sufficient available resources to fulfill the new requirement of nodes and either (1) there is no job pending for execution in the queue, or (2) no pending job can be executed due to insufficient available resources. By expanding the job, we can expect it to finish its execution earlier and release the associated resources. A job is shrunk if there is any queued job that could be executed by taking this action. More jobs in execution should increase the global throughput. Moreover, if the job is going to be shrunk, the queued job that has triggered the shrinking event will be assigned the maximum priority in order to foster its execution. Algorithm 1 Slurm Reconfiguration Policy 1: if preferred then 2: if am I the only job in the queue? then 3: action ←expand. 4: processes ←jobMaxP rocs. 5: else 6: if can I expand to preferred?then 7: action ←expand. 8: processes ←max procs to(preferred). 9: else 10: if can I shrink to preferred?then 11: action ←shrink. 12: processes ←preferred. 13: else 14: if are there pending jobs in the queue? then 15: if can other job run with my resources? then 16: action ←shrink. 17: processes ←min procs run(targetJobId). 18: set max priority(targetJobId). 19: else 20: action ←expand. 21: processes ←max procs to(jobMaxP rocs). 22: else 23: action ←expand. 24: processes ←jobMaxP rocs. V. NANOS++ RUNTIME EXTENSIONS We implemented the necessary logic in Nanos++ to reconfigure jobs in tight cooperation with the RMS. In this section we discuss the extended API and the resizing mechanisms. A. The Dynamic Management of Resources (DMR) API We designed the DMR API with two main functions: dmr_check_status and its asynchronous version dmr_icheck_status. These routines instruct the runtime (Nanos++) to communicate with the RMS (Slurm) in order to determine the resizing action to perform: “expand”, “shrink”, or “no action”. The asynchronous counterpart schedules the next action for the next execution step, at the same time that the current step is executed. Hence, by skipping the action scheduling stage, the communication overhead in that step is avoided. Details of the API overhead have to be avoided due to lack of space. Thus, in case an action is to be performed, these functions spawn the new set of processes and return an opaque handler. This API is exposed by the runtime and it is intended to be used by applications. These functions present the following input arguments. •Minimum number of processes to be resized to. •Maximum number of processes. This prevents the application from growing beyond its scalability capabilities. •Resizing factor (E.g.: a factor of 2 will expand/shink the number of processes to a value multiple/divisor of 2). •Preferred number of processes. The output arguments return the new number of nodes and an opaque handler to be used in the task offloading directives. An additional mechanism implemented to reach a fair balance between performance and throughput is the “checking inhibitor”. This introduces a timeout during which the calls to the DMR API are ignored. This knob is mainly intended to be leveraged in iterative applications with short iteration intervals. The inhibition period can be tuned by means of an environment variable (NANOX_SCHED_PERIOD). B. Automatic Job Reconfiguration The runtime will perform the following actions in order to leverage the Slurm resizing mechanisms (see Section IV) by means of its external API: 1) Expand: A new resizer job (RJ) is first submitted requesting the difference between the current and total amount of desired nodes. This enables the original nodes to be reused. There is a dependency relation between the RJ and the original job (OJ). In order to follow better the RMS decisions, RJ is set to the maximum priority, facilitating its execution. The runtime waits until JR moves from “pending” to “running” status. If the waiting time reaches a threshold, RJ is canceled and the action is aborted. This situation may occur if the RMS assigns the available resources to a different job during the scheduling action. This is more likely to occur in the asynchronous mode because an action then can experience some delay during which the status of the queue may change. Once OJ is reallocated, the updated list of nodes is gathered and used in a call to MPI_Comm_spawn in order to create a new set of processes. 2) Shrink: The shrinking mechanism is slightly more complex than its expansion counterpart because Slurm will have to kill all processes executing in the released nodes. To prevent premature process termination, we need a synchronized workflow to guide the job shrinking. Hence, the RMS sets a management node in charge of receiving an acknowledgment from all other processes. These ACKs will signal that they finished their offloading tasks and the node is ready to be released. After a scheduling is complete, the DMR call returns the expand-shrink action to be performed and the resulting 1void main (i n t argc ,c h a r **argv ){ 2. . . 3int t= 0 ; 4MPI_Comm_get_parent(&parentComm ) ; 5i f (parentComm == MPI_COMM_NULL ){ 6init (data ) ; 7}else { 8MPI_Recv (parentComm ,data ,myRank ) ; 9MPI_Recv (parentComm , &t,myRank ) ; 10 } 11 compute (data ,t) ; 12 . . . 13 } 14 15 void compute (data ,t0 ){ 16 f o r (t=t0 ;t<timesteps ;t++) { 17 nodeList =get_new_nodelist_somehow ( ) ; 18 i f (nodelist != NULL ){ 19 MPI_Comm_spawn (myapp .bin ,nodeList , &newComm ) ; 20 MPI_Send (newComm ,data ,myRank ) ; 21 MPI_Send (newComm ,t,myRank ) ; 22 exit ( 0 ) ; 23 } 24 compute_iter (data ,t) ; 25 } 26 } Listing 1. Pseudo-code of job reconfiguration using bare MPI. number of nodes. The application is responsible for stating the appropriate data dependencies and triggering the tasks offloading to the new set of processes (see Section VI next). VI. PROGRAMMING MODEL In this section we review our programming model approach to address dynamic reconfiguration coordinated by the RMS. The programmability of our solution benefits from relying the OmpSs offload semantics versus directly using MPI. Next, we illustrate this via a practical example. A. Benefits of the OmpSs Offload Semantics To showcase the benefits of the OmpSs offload semantics, we review the specific case of migration. This analysis allows us to focus on the fundamental differences between programming models because it does not involve data redistribution among a different number of nodes (which is of similar complexity in both models). We show a complete example in Section VI-B. a) MPI Migration: Listing 1 contains an excerpt of pseudo-code with a program which directly uses MPI calls. In this case, we assume some mechanism is available to determine the new node list in line 17. b) OmpSs-based Migration: The same functionality is attained in Listing 2 by leveraging our proposal on top of the OmpSs offload semantics. This includes a call to our extended API in line 11. At a glance, our proposal exposes higher-level semantics, increasing code expressiveness and programming productivity. In addition, communication with the RMS is implicitly established in the call to the runtime in line 11, which pursues a potential increase in overall system resource utilization. Data transfers are managed by the runtime with the directive in line 13. Last, the “taskwait” (line 15) follows the original semantics of the clause and with it, the 1void main (v o id ){ 2. . . 3int t= 0 ; 4init (data ) ; 5compute (data ,t) ; 6. . . 7} 8 9void compute (data ,t0 ){ 10 f o r (t=t0 ;t<timesteps ;t++) { 11 action =dmr_check_status(&newNnodes , &handler ) ; 12 i f (action ){ 13 #pragma omp task inout (data )onto (newComm , myRank ) 14 compute (data ,t) 15 #pragma omp taskwait 16 }else 17 compute_iter (data ) ; 18 } 19 } Listing 2. Pseudo-code of job reconfiguration using OmpSs. #0 #1 Comm 1 (a) Expand. #0 #1 Comm 2 (b) Shrink. Fig. 2. Data transfers. initial processes terminate, letting the execution continue in the processes of the new communicator. B. A Practical Example The excerpt in Listing 3 is derived from that showcased in Section VI-A to discuss malleability. In this case the application must drive the task redistribution according to the resizing action. The mapping factor indicates the number of processes in the current set that are mapped to the processes in the new configuration (see Figure 2). This example implements homogeneous distributions, where we always resize to a multiple or divisor of the current number of processes. Our model, however, supports arbitrary distributions. For the “expand” action (line 8), the original processes must partition the dataset. For instance, in Figure 2a, the processes split the dataset into two subsets, mapping each half to a process in the new configuration. The data transfers are performed by the runtime according to the information included in the task offloading directive (line 12). The “shrink” action, on the other hand, involves preliminary explicit data movement. The processes in the original set are grouped into “senders” and “receivers”. This initial data movement is illustrated in the example in Figure 2b. VII. EXPERIMENTAL SETUP A. Platform Our evaluation was performed on the Marenostrum Supercomputer at Barcelona Supercomputing Center. Each compute node in this facility is equipped with two 8-core Intel Xeon E5-2670 processors running at 2.6 GHz with 128 GB of RAM. 1void compute (data ,t0 ){ 2f o r (t=t0 ;t<timesteps ;t++) { 3action =dmr_check_status(&newNnodes , &handler ) ; 4i f ( ! action ) 5compute_iter (data ) ; 6else { 7i f (action == ” expand ” ){ 8factor =newNnodes /worldRanks ; 9f o r (i=0; i<factor ;i++) { 10 dest =myRank *factor +i; 11 subdata =part_data (factor ,data ) ; 12 #pragma omp task inout (subdata )onto (handler ,dest ) 13 compute (subdata ,t) ; 14 #pragma omp taskwait 15 }/ / End f o r 16 }e l s e i f (action == ” s h r i n k ” ){ 17 factor =worldRanks /newNnodes ; 18 sender = (myRank %factor )<(factor −1 ) ; 19 i f (sender ){ 20 dst =factor *(myRank /factor + 1) −1 ; 21 MPI_Isend (comm ,data ,dst ) ; 22 }else {/ / R e c e i v e r 23 f o r (i=1; i<=factor ;i++) { 24 src =myRank −factor +i; 25 MPI_Irecv (comm , &alldata ,src ) ; 26 }/ / End f o r 27 }/ / End i f ( s e n d e r ) 28 MPI_Waitall ( ) ; 29 i f ( ! sender ){ 30 dest =myRank /factor ; 31 #pragma omp task inout (alldata )onto (handler ,dest ) 32 compute (alldata ,t) ; 33 #pragma omp taskwait 34 }/ / End i f ( ! s e n d e r ) 35 }/ / End i f ( a c t i o n == . . . ) 36 }/ / End i f ( a c t i o n ) 37 }/ / End f o r 38 }/ / End compute ( ) Listing 3. Pseudo-code of a malleable application. The nodes are connected via an InfiniBand Mellanox FDR10 network. For the software stack we used MPICH 3.2, OmpSs 15.06, and Slurm 15.08. Slurm was configured with the backfill job scheduling policy. Furthermore, we also enabled job priorities with the policy multifactor. Both were configured with default values. B. Flexible Applications For our experimentation we used one synthetic and three real applications, described below: 1) Flexible Sleep (FS): This iterative synthetic application performs a sleep in each step (iteration). The time of the step depends on the number of processes deployed in that iteration, so if the job is resized, the sleep time is modified assuming perfect linear scalability. Apart from the sleep that simulates the computation time, the application also manages an array of doubles, distributed among the ranks. This array conforms the data-dependency for OmpSs, redistributed in each reconfiguration. 2) Conjugate Gradient (CG): The CG method1is an iterative algorithm for the numerical solution of sparse systems of linear equations that produces a solution after a finite number of steps. For our experimentation the method will perform a specific number of iterations in order to have control of the execution time. 1https://en.wikipedia.org/wiki/Conjugate gradient method The data of the application propagated in each iteration step of CG is constrained to a matrix flat-stored and four vectors. This version is implemented using OpenMP+MPI, and each MPI process works on a block of rows of the matrix and the corresponding elements from the vectors. The local matrixvector products are parallelized with OpenMP. We have applied our OmpSs-based extensions in order to ease the creation of new processes while maintaining the data dependencies after the resizing procedure. The five data structures in CG conform the data-dependencies between iterations in the OmpSs programming model, and they are redistributed when a rescaling is necessary. During a resize, the data in the matrix and vectors must be redistributed according to the new number of MPI processes. 3) Jacobi: The Jacobi method2is an iterative and embarrassingly-parallel algorithm for the solution of a system of linear equations. Our OpenMP+MPI version of this solver is based on the implementation presented in [21]. The program layout is similar to the CG implementation. In this application, we also have a flat matrix, but only two vectors. These three structures conform the data-dependencies for OmpSs and they are all distributed among the processes. 4) N-body: The N-body problem3simulates the individual motions of a group of objects interacting with each other by means of a given force. We have used an OmpSs+MPI version of this simulator where each process stores a subset of particles, while the intranode parallelism is exploited by OmpSs. The amount of work of N-body per iteration is considerably larger than that present in the remaining two real applications described in this section. Apart from computing the position and forces of its own particles, each process exchanges its local subset of particles with the other processes. At the end of the iteration, all the processes have worked with the whole set of particles. The data-dependency in this particular case is dictated by an array of particles with information about position, velocity, mass and weight. This array is split or merged when an scaleup or down is respectively scheduled. C. Workload Configuration The workloads were generated using the statistical model proposed by Feitelson [6], which characterizes rigid jobs based on observations from logs of actual cluster workloads. These include the distribution of job sizes in terms of number of processors, the correlation of runtime with parallelism, and the number of repeated runs. Among others, we found four customizable parameters to be especially relevant: •Jobs: Number of jobs to be launched. •Job size: Number of nodes determined by a complex discrete distribution. •Runtime: Fixed following a hyperexponential distribution based on the job size. 2https://en.wikipedia.org/wiki/Jacobi method 3https://en.wikipedia.org/wiki/N-body problem TABLE I CONFIGURATION PARAMETERS FOR THE APPLICATIONS Number of Processes Application Iterations Minimum Maximum Preferred Scheduling period FS 25 1 20 - - CG 10000 2 32 8 15 seconds Jacobi 10000 2 32 8 15 seconds N-body 25 1 16 1 - 0 5 10 15 20 25 0 10000 20000 30000 40000 50000 60000 70000 80000 10 25 50 100 200 400 Gain (%) Execution time (s) Number of jobs Fixed Flexible Gain Fig. 3. Comparison between fixed and flexible workloads with different number of jobs. •Arrival: Inter-arrival times of jobs modeled using a Poisson distribution. For every job the shrink-expand factor was set to 2. VIII. PRELIMINARY STUDY In this section we present an in-depth analysis of the framework’s features using synthetic flexible jobs. For this purpose, we used the FS application configured to perform 2 steps and to transfer 1 GB of data during the reconfiguration. In order to test thoroughly the features of our solution, we performed 4 different experiments comprising: synchronous and asynchronous scheduling, heterogeneous workload, and micro-steps. A. Testbed For this study we only used the FS application; we generated several workloads of different size (number of jobs), assigning up to 20 nodes to each job (the number of available nodes in this experiment) with the parameter job size; the maximum runtime was set to 60 seconds for each step; and the average arrival time was 10 seconds. Table I details the rest of the FS parameters for the reconfigurations. The array was determined to have 1 GB of data transferred in each iteration. Furthermore, by not giving a preferred value the RMS has more freedom to reallocate resources. B. Synchronous Reconfiguration Scheduling For the first test, we launched workloads of different sizes in terms of number of jobs. Figure 3 depicts the execution time for this experiment. Each workload has both a fixed as well as a flexible version. The line “Gain” in that chart indicates the reduction of the execution time (in %) attained by the flexible workload with respect to the fixed workload. Fig. 4. Evolution in time for the 10-job workload. Fig. 5. Evolution in time for the 25-job workload. Except for the 10-job workload, we can appreciate a gain in the interval 10-15% for the execution time, though the benefit decreases as the workload grows. Nonetheless, this occurs because we are evaluating a finite workload. Under these conditions, the scheduler is able to backfill jobs and fulfill resources, but malleability cannot bring a higher resource utilization. In a more realistic scenario, involving a much larger workload, the throughput is always higher for the flexible workloads. For instance, bottom plots in Figures 4 and 5 show that the productivity attained by the flexible workload is always higher than that offered by the fixed workload. Figure 4 reports an almost-full allocation of resources during the flexible execution, exposing that the remarkable gain is due to the augment in resource usage. In contrast, Figure 5 depicts the behavior of a workload of 25 jobs, for which we observe a narrower gain. The problem here arises on the last job (LJ) just in the vertical black line in that figure. Specifically, in that instant LJ is using 4 nodes Fig. 6. Asynchronous scheduling of the 10-job workload. -7 -5 -3 -1 1 3 5 7 0 10000 20000 30000 40000 50000 60000 70000 80000 10 25 50 100 200 400 Gain (%) Execution time (s) Number of jobs Fixed Flexible Gain Fig. 7. Comparison between fixed and flexible workloads using asynchronous selection of action. while the penultimate job (PJ) before completion, owns the rest of resources. When PJ finishes, 16 nodes are released, but until the next check, LJ cannot be expanded. At that point, the scheduler decides to expand the job to its maximum, in this experiment 16 nodes. At the end of the timeline, no more jobs can use the spare resources. This is the same situation that appears in the fixed workload. The consequence is that there is no further improvement, as the flexible policy had already obtained the gain from the first reallocation of resources. C. Asynchronous Reconfiguration Scheduling In this test we evaluated the asynchronous version of the scheduling. Again, we compare a fix workload and its flexible counterpart, but now the decision is made asynchronously. Here, we remind that the asynchronous scheduling takes a decision in a specific step but the action takes place in the next step. In the meantime, the status of the system may change. Therefore, the conditions found when the action is applied in the next step might not be the same that were present when the RMS decided the future action. In this situation, enforcing outdated actions may result in an inefficient use of resources. Let us analyze the effect of adopting outdated decisions for the asynchronous scheduling in the 10-job workload. In this 24599 23875 22048 22210 21442 19000 20000 21000 22000 23000 24000 25000 025 50 75 100 Execution time (s) Rate of flexible jobs (%) Fig. 8. Execution times of workloads of 100 jobs with different rates of flexible (Showing the top of the chart, Y axis is not starting in 0). example the flexible workload performs worse than the fixed execution. We can explain the cause analyzing Figure 6. If we focus on the resource allocation evolution (top of Figure 6), two relevant gaps can be identified between seconds 2000- 3000 and 3000-4000. At the beginning of the first gap in Figure 6, there are 3 jobs in execution that occupy a total of 19 nodes: one (J1) with 16, one (J2) with 2, and a last one (J3) with 1. Moreover, the RMS has decided to expand J3 to 2 nodes in the next step. Now, J1 is finished, leaving the other 2 active jobs using 3 nodes (J3 has not expanded yet). J2 is resolved to expand and acquires 16 nodes (around second 2500). J3 experiences a longer step, carrying out its pending action of expanding to 2, yielding an overall of 18 allocated nodes. If J3 had checked the resources at that moment, it would have been expanded to 4 nodes, but the asynchronous scheduling was negotiated earlier, when the conditions were different. In addition to realizing the expansion, the scheduler decides that J3 will expand to 4 nodes in the next step. The beginning of the second gap indicates the completion of J2. A few seconds later J3 expands to 4 nodes (instead of doing it to 16 if the scheduling had been synchronous). Despite the lack of good results for small workloads, the larger workload completion times reveal a higher gain. In that type of situation the malleability overcomes the initial problem described above. As we did in the synchronous benchmark (Figure 3), if we dismiss the small executions (10-to-50-job workloads), we can observe around a 6%-gain, with the improvement decreasing as we add more jobs to the workload. This test reveals that so far, there is no need of using an asynchronous scheduling. Hence, the rest of the experiments will exclusively use the synchronous mode. D. Heterogeneous Workloads In this benchmark we mixed flexible and fixed jobs in the same workload in order to study their interaction. The workloads were composed of 100 jobs and the percentage of flexible jobs determined the probability of a job being flexible. We raised the ratio of flexible jobs between 0% and 100% in steps of 25%. Figure 8 depicts the execution times for these configurations. In general, we can appreciate that the execution -6.80% 9.71% 5.83% 5.83% -0.97% 1.96% 8.50% 10.78% 10.13% 9.48% 4.40% 11.01% 13.36% 11.79% 7.23% 4.62% 10.39% 12.17% 11.28% 8.70% 0200 400 600 800 1000 Flexible Sched 2 Sched 5 Sched 10 Sched 20 Execution TIme (s) Workload Configuration 10 Jobs 25 jobs 50 Jobs 100 Jobs Fig. 9. Execution time for the different inhibition periods (bars) and the gain respect the fixed workload (percentage on the right of the bars). time decreases as the ratio of flexible jobs grows. The results show a 10%-gain with only a 50%-rate of flexible jobs, and up to a 12%-improvement when all of them are flexible. E. Checking Period Inhibitor in Micro-steps Applications For this test we reduced the time step in the model to an average of 2 seconds in order to investigate the importance of the overhead incurred by the scheduling process. Again, we generated workloads with a different number of jobs and executed them as both fixed and flexible workloads. For the flexible workloads, we enabled the checking inhibitor (see Section V-A) to prevent that each iteration triggers a check. Figure 9 depicts the variation of malleability for a fixed execution. The group at the top (“Flexible”), represents an execution without the checking inhibitor mechanism. The rest of the groups in the chart (bellow “flexible”), show the execution time when configuring the inhibitor period to: 2, 5, 10 and 20 seconds (from top to bottom). The conclusion is that this gain is virtually negligible or even negative in some cases. Hence, enabling periods of checking inhibition can avoid a burst of communications between the runtime and the RMS, reducing the overhead. In this particular example, setting a period of 5 seconds between scheduling, not only offers better results than the fixed workload, but also outperforms a simple flexible workload. IX. EXPERIMENTAL RESULTS A. Testbed For the experiments, we have generated workloads of different sizes, configuring the job submissions to the queue at their best performance. We first evaluate the individual scalability of the three real applications. These tests identify two parallel behaviors: •High scalability (CG and Jacobi). In this case both applications have a similar behavior, with the highest speed-up attained for 32 processes. However, from 8 processes on, the difference gain between tests drops below 10%, so we consider 8 processes as a “sweet configuration spot” for these two applications. •Constant performance (N-body). In contrast, this application reaches its maximum performance for 16 processes. However, in this case, the gain does not exceed 10% with respect to the sequential run, so a single process is considered as the “sweet spot”. From the perspective of computational cost, CG and Jacobi comprise “short” iterations that complete in less than 2 seconds, while N-body executes costly iterations, in the scale of minutes. For this reason, for CG and Jacobi we enabled the scheduling period inhibitor featured by the runtime, in order to reduce the amount of communications with the RMS. Table I displays the configuration employed in the experimentation. The job submission of each application is launched with its “maximum” value, reflecting the user-preferred scenario of a fast execution. Each workload is composed of a set of randomly-sorted jobs (with a fixed seed) which instantiate one of the three real applications (33% of jobs of each application class). Furthermore, the inter-arrival time among submissions is generated using the statistical model proposed by Feitelson [6], which characterizes rigid jobs based on observations from logs of actual cluster workloads. B. Results Figure 10 depicts the execution time of each workload size comparing both configuration options: fixed and flexible. The labels at the end of the “flexible” bars report the gain compared with the fixed version. Table II details the measures extracted from the executions. In the first row, we compare the average resource utilization for fixed and flexible workloads. This rate corresponds to the average time when a node has been allocated by a job compared to the workload execution time. These results indicate that the flexible workloads reduce the allocation of nodes around 30%, offering more possibilities for queued jobs. The second row of Table II shows the average waiting time of the jobs for each workload. These times are illustrated in Figure 11, together with the gain rate for flexible workloads. The reduction around 60% makes the job waiting time a crucial measure to keep in mind from the perspective of throughput. In fact, this time is responsible for the reduction in the workload execution time. The last two rows of Table II present two more aggregated measures of all the jobs in the workload: The first one is the average execution time; the second is this execution time plus the waiting time of the job, referred to as completion time. The experiments show that jobs in the flexible workload are affected by the scale-down of their number of processes. However, this is compensated by the waiting time which benefits the completion time. In order to understand the events during a workload execution, we have chosen the smallest workload to generate detailed charts and offer an in-depth analysis. The top and the bottom plots in Figure 12 represent the evolution in time of the allocated resources and the number of completed jobs. It also shows the number of running jobs for fixed and flexible workloads (blue and red lines respectively). The figures demonstrate that the flexible workload utilizes fewer resources; furthermore, there are more jobs running 46.48% 49.04% 41.42% 41.97% 0 10000 20000 30000 40000 50000 60000 70000 80000 50 100 200 400 Execution time (s) Number of jobs Flexible Fixed Fig. 10. Workloads execution times (bars) and the gain of flexible workloads (bar labels). 66.95% 69.33% 60.74% 56.40% 0 5000 10000 15000 20000 25000 30000 35000 50 100 200 400 Avg. job waiting time (s) Number of jobs Flexible Fixed Fig. 11. Average waiting time for all the jobs of each workload (bars) and the gain of flexible workloads (bar labels). concurrently (top chart). For both configurations, jobs are launched with the “sweet spot” number of processes; the fixed jobs obviously do not vary the amount of assigned resources, while in the flexible configuration, they are scaled-down as soon as possible. This explains the reduction on the utilization resources. For instance, in the second half of the flexible shape in Figure 12 (top), there are 5 jobs in execution which allocate 40 nodes. The next eligible job pending in the queue needs 32 nodes to start. Therefore, unless one of the running jobs finishes, the pending job will not start and the allocation rate will not be higher. When a job eventually finishes and releases 8 nodes, the scheduler initiates the job requesting 32 nodes. Now, the allocated nodes are 64 (the green peaks in the chart); however, as the job prefers 8 processes, it will be scaled-down. At the beginning of the trace in the bottom of Figure 12, the throughput of the fixed workload is higher, but this occurs because the first jobs are completed earlier (they have been launched with the best-performance number of processes). Meanwhile, in the flexible workload, many jobs are initiated (blue line) and, as soon as they start to finish, the throughput experiences a boost. X. CONCLUSIONS AND FUTURE WORK We have improved the state-of-the-art in dynamic job reconfiguration by targeting the global throughput of a high performance facility. For this purpose, we have taken advantage of first-class tools to design this new approach that introduce a dynamic reconfiguration mechanism for malleable jobs, composed of two modules: the runtime and the resource