The Parallel Applications for Distributed Systems Essay Example

processors located in a different site, cluster, or domain.

Reconfigurable or malleable and migratable applications enhance the capabilities and adaptability of distributed computing scheduling and resource management systems.

Saving the state of parallel applications is necessary for starting and stopping them. Scholars like Elonazhy, Plank, Tannenbaum, Litzkow, Dikken, and Geist have examined various checkpointing strategies for both sequential and parallel applications. Tannenbaum and Litzkow created checkpointing systems specifically for sequential applications while Dikken et al. focused on parallel applications. Checkpointing systems can be classified based on their transparency to users and the portability of the checkpoints. Transparent and semi-transparent systems proposed by Tannenbaum and Litzkow hide checkpointing details from users but lack portability. Non-transparent systems like Geist et al.'s require program modifications but are highly portable across different systems. Additionally, checkpointing can be implemented at either the kernel level or user-level.

This paper presents a checkpointing infrastructure for malleable and migratable parallel applications in distributed systems. The infrastructure includes a semi-transparent checkpointing library called SRS (Stop Restart Software) and a Runtime Support System (RSS). The SRS library is semi-transparent as it requires the user to insert specific calls in their program to indicate the data for checkpointing and to restore the application state during a restart.

The library handles the storing of checkpoints and redistribution of data internally. Although there are a few checkpointing systems that allow changing the parallelism of parallel applications, this system is different because it enables applications to be migrated to distributed locations with different file systems without manual migration of checkpoint data. To achieve this, a distributed storage infrastructure called IBP is used to remotely access

checkpoint data.

The SRS Checkpointing Library

The SRS (Stop Restart Software) is a library for user-level checkpointing that allows iterative parallel MPI message passing applications to be reconfigured. These applications include linear solvers, heat-wave equation solvers, and partial differential equation (PDE) applications. The SRS library can be used with both C and Fortran programming languages and can be accessed from MPI programs in either language. The library comprises of 6 primary functions.

SRSJnit is an educational institution that functions as a university.

2. SRSJtestart-Value,

3. SRS-Read

The SRS-Register,

The SRS-CheckJ3top is a tool used for verification and validation.

6. SRS-Finish.

The user invokes SRS-Init after invoking MPIJnit. SRS-Init is a collective operation that initializes the internal data structures used by the library. SRSJnit also reads specific parameters from a configuration file given by the user. These parameters include the location of the Runtime Support System (RSS) and a flag indicating the need for periodic checkpointing in the application. After reading these parameters, SRSJnit connects with the RSS and transmits the current number of processes utilized by the application. If the application has been restarted from a previous checkpoint, SRSJnit also acquires the previous configuration of the application from the RSS.

To save and resume an executing application, it is important to save both the data used by the application and its execution context. When starting the application, certain data needs to be initialized. However, when resuming from a checkpoint, the data can be read from the checkpoint instead of going through the initialization phase. Some checkpointing systems restore execution context by storing and retrieving

the execution stack. However, this method compromises the portability of the checkpointing system [30].

The SRS library's primary objective is to provide heterogeneous support. So, the user needs to use SRS-Restart_Value to restore the execution context. When the user calls SRS-Restart-Value, it returns 0 if the application is starting its execution and 1 if the application is continuing from its previous checkpoint. By utilizing these values, the user can implement conditional statements in their application. These statements allow executing specific parts of the code when the application starts and different parts when it continues from its preceding checkpoint.

The SRS library uses the Internet Backplane Protocol (IBP) for storing checkpoint data. Depots are started on all the desired machines for the application's execution. The SRS-Register function is used to identify the data that will be checkpointed through periodic checkpointing or when SRS-Check_Stop is called. Only the data included in the SRS-Register call will be checkpointed, and parameters such as size, data type, and data distribution can be specified. The SRS library supports various data distributions such as block, cyclic, and block-cyclic distributions. For data local to a process or without distribution, a distribution value of 0 can be used. The parameters of the data are stored in a local data structure by the SRS-Register function, but it does not actually perform the checkpointing of the data.

The main function SRS_Read is responsible for reconfiguring the application. When the application is stopped and then continued, the checkpointed data can be retrieved by using SRS-Read. The user needs to provide the name of the checkpointed data, the memory in which the data is

stored, and the new data distribution when calling SRS-Read. The specified data distribution can be a conventional distribution or 0 for no distribution or SAME if the same data needs to be propagated to all processes. The value SAME is particularly useful when retrieving iterator values and all processes must start execution from the same iteration. SRS-Read communicates with RSS to obtain the previous data distribution and the location of the actual data. If no distribution is specified for SRS-Read, each process will retrieve its corresponding portion of data from the IBP depot used in the previous execution.

If SAME is used for data distribution, the first process reads data from the IBP depot of the first process in the previous execution and broadcasts it to the other processes. If data distribution is specified in SRS_Read, SRS-Read determines data maps for the old and new distributions of the data in the previous and current distributions. Each process retrieves its portion of data from the IBP depots based on the information in the data maps. Thus, reconfiguration of the application is achieved by using a different level of parallelism in the current execution and specifying a data distribution in SRS-Read that may be different from the distribution used previously.

SRS-Check_Stop is a collective operation called at different stages of the program to determine if the application should be halted. If the return value of SRS-Check_Stop is 1, it indicates that an external component has requested the application to stop. In this case, the executing application can run specific code to halt itself. SRS-Check_Stop communicates with the RSS to fetch a value that indicates whether

the application needs to be stopped.

To halt the application upon request from an external component, SRS-Check-Stop saves the data distributions and registered data of the application from SRS-Register to the IBP (Plank et al.) depots. Each process of the parallel application saves its specific data to the local IBP depot. By saving only the specified data from SRS-Register and requiring each process to save its data in the IBP depot on its respective machine, the overhead for checkpointing is greatly reduced. SRS-Check-Stop sends pointers to the checkpointed data to RSS and removes all local data structures managed by the library.

SRS-Finish is collectively called by all parallel application processes before invoking MPI-Finish. When invoked, SRS-Finish removes all local data structures managed by the library and communicates with RSS to request termination of execution. In addition to the six main functions, SRS also provides two extra functions: SRS-DistributeFunc-Create and SRS_DistributeMap-Create. These additional functions allow users to define their own data distributions instead of relying on the standard ones provided by the SRS library.

The Runtime Support System (RSS)

The RSS (Remote Server Service) is a sequential application that can run on any machine. It communicates with the machines used for executing the actual parallel application. The RSS remains active throughout the entire application runtime, including multiple migrations. Before starting the parallel application, the user initiates the RSS. The RSS displays a port number for listening to requests. The user then enters the machine name and port number into a configuration file called srs.config and shares it with the first process of the parallel application. When the parallel application

begins, the first process retrieves the RSS location from the configuration file and registers with it through SRS_Init. The RSS keeps track of the current and previous execution configurations of the application.

The stop-flag is an internal flag within the RSS that determines whether the application needs to be stopped. At first, the flag is cleared by the RSS. To stop the application, a utility named stop-application is available for the user. By providing the RSS's location as an input parameter when executing this utility, it communicates with the RSS and instructs it to set the stop-flag. When SRS-Check.Stop is called by the application, the SRS library contacts the RSS and retrieves the value of the stop-flag. Based on this value, either execution continues or stops.

When the SRS-Check_Stop sends the location and data distributions of the checkpoints to the RSS during the transfer of data from the application to IBP depots, it later retrieves the location of the checkpoints from the RSS when the application is restarted. Upon calling SRS-Finish, the application requests the RSS to terminate itself. The RSS then cleans the data stored in the IBP depots, deletes its internal data structures, and terminates.

In this section, we will discuss the previous research and studies that are relevant to our work.

Extensive research has been conducted on checkpointing parallel applications (Elnozahy et al.). Multiple checkpointing systems have been developed for this purpose (Dikken et al., Godard et al.). While certain systems were designed for homogeneous systems (Russ et al.), others allow applications to be checkpointed and restarted on heterogeneous systems (Naik et al.). However, Calypso and Plinda (Jeong

et al.; Baratloo et al.) require application writers to use specific constructs in their programs and are not compatible with third-party software.

The text mentions various systems such as Dynamic PVM and CUMULVS which utilize PVM mechanisms for fault detection and process spawning. However, these systems can only be used with PVM environments. Cocheck and Starfish, on the other hand, provide their own MPI implementations for fault tolerance. However, these implementations are not suitable for distributed computing and Grid systems that use the more secure MPICH-G. Additionally, the text states that CUMULVS, Dome, the research by Hofmeister and Deconick, DRMS, and DyRecT are closely related to the research in terms of their checkpointing API, migrating infrastructure, and reconfiguration capabilities.

The CUMULVS (Geist et al.) API has many similarities to our API. It follows a similar approach where application writers need to specify the data distributions used in their applications. It also offers support for commonly used data distributions such as block and cyclic. Additionally, CUMULVS allows applications to be stopped and restarted. However, there is a limitation that applications can only be stopped and continued on the same number of processors. Despite supporting MPI applications, CUMULVS is built on top of PVM infrastructure, which means that applications can only be executed on PVM.

The support for reconfiguration of executing applications, in terms of changing parallelism, is provided by Dome (Arabe et al.). However, in order to be eligible for redistribution for reconfiguration, the data needs to be declared as Dome objects. Consequently, using Dome with third-party software like ScaLAPACK, which utilizes native data for computations, becomes challenging. Additionally, Dome relies on PVM

as its underlying architecture, making it unsuitable for message passing applications.

Hofmeister's research supports the reconfiguration of software modules in an application, including dynamic module replacement, module relocation to different processors, and module addition or removal from applications. However, this approach is limited to homogeneous systems. On the other hand, Deconinck's work shares similarities with SRS in terms of the checkpointing API and infrastructure. Their checkpoint control layer is comparable to our RSS in managing distributed data and communication protocols between applications and the checkpoint control layer. Both our approach and Deconinck's utilize architecture-independent checkpoints for heterogeneous and portable checkpoints. However, Deconinck's work does not support application reconfiguration concerning varying parallelism for applications.

The DyRecT framework allows for dynamic reconfiguration of applications by adding or removing processors during their execution. It provides a user-level checkpointing library that supports data distribution and allows for dynamic reconfiguration across heterogeneous systems. However, DyRecT's implementation of the checkpointing infrastructure uses LAM MPI, which limits its suitability to workstation clusters and not distributed and Grid systems where MPICH-G is preferred. Additionally, DyRecT requires machines to have a common file system, making it impossible to migrate and reconfigure applications to distributed locations without shared file systems.

The DRMS (Naik et al.) checkpointing infrastructure utilizes the DRMS programming model to enable the checkpointing and restarting of parallel applications on varying numbers of processors. It includes robust checkpointing mechanisms for storing and retrieving checkpoint data from permanent storage. The DRMS checkpointing API allows for the reconfiguration of MPI message passing applications on heterogeneous systems, making it closely related to SRS. However, DRMS does not support migrating and restarting applications on environments without

shared file systems. A more recent work by Kale et al. achieves reconfiguration of MPI-based message passing programs by utilizing a less suitable MPI implementation called AMPI, which is not ideal for Grid systems like MPICH-G.

In conclusion, this article discusses the findings and implications of the research conducted. It also proposes potential areas of future investigation and development.

This paper presents an explanation of a checkpointing infrastructure used for developing and executing malleable and migratable parallel applications across heterogeneous sites. The SRS API offers a limited number of functions to enable seamless malleability of parallel applications. The SRS system stands out by utilizing the IBP distributed storage infrastructure. The discussion focuses on evaluating the overhead experienced by the applications, as well as the time required for storing, reading, and redistributing checkpoints. Results demonstrate that SRS enables reconfigurability of parallel applications with minimal overhead.

In the future, one of the main goals could be to utilize precompiler technologies in order to restore the execution context and reduce the need for significant program modifications by users, thus enabling greater flexibility in their applications. Further research can involve introducing checkpointing functionality for files, handling complex pointers and structures, and providing support for various application types.

The checkpointing framework's design allows for migration of heterogeneous environments, but the current implementation stores the checkpoint data as raw bytes. This can cause data misinterpretation by the application if, for instance, the data is stored on a Solaris system and read by a Linux machine. The reason for this is the varying byte orderings and floating point representations used by different systems.

In order to

enhance the fault-tolerance of the RSS daemon, it would be beneficial to regularly save its state. This would enable the RSS service to be relocated across different sites. Currently, all processes in the parallel application depend on a sole RSS daemon for communication. This can result in scalability problems when managing numerous machines. Our future plan involves implementing a distributed RSS system to tackle this scalability concern.

References

The website for LAM-MPI can be found at http://www.lam-mpi.org.The book titled The Grid: Blueprint for a New Computing Infrastructure, edited by I. Foster and C. Kesselman, was published in 1999 by Morgan Kaufmann. Its ISBN is 1-55860-475-8.
The article "A Survey of Rollback-Recovery Protocols in Message Passing Systems" was written by M. Elnozahy, L. Alvisi, Y.M. Wang, and D.B. Johnson. It was published as a Technical Report with the reference CMU-CS-96-181 by the School of Computer Science at Carnegie Mellon University in Pittsburgh, PA, USA in October 1996.

L. Dikken, F. van der Linden, J. J. J. Vesseur, and P. M. A. Sloot conducted a study titled "DynamicPVM: Dynamic Load Balancing on Parallel Systems." This study was published in the Lecture notes in computer science 797 book, specifically in the High Performance Computing and Networking section of Proceedings Volume II, Networking and Tools chapter.

G. A. Geist, J. A. Kohl, and P. M. Papadopoulos. CUMULVS:

Offering Fault-Tolerance, Visualization and Steering of Parallel Applications. International Journal of High Performance Computing Applications, 11 (3):224-236, August 1997.

The text below is a technical report titled "A Survey of Rollback-Recovery Protocols in Message Passing Systems" written by M. Elnozahy, L. Alvisi, Y.M. Wang, and D.B. Johnson. It was published in October 1996 by the School of Computer Science at Carnegie Mellon University in Pittsburgh, PA, USA.

S. H. Russ, B. K. Flachs, J. Robinson, and B. Heckel. Hector: Automated Task Allocation for MPI. In Proceedings of IPPS '96, The 10th International Parallel Processing Symposium, pages 344-348, Honolulu, Hawaii, April 1996.

The authors' names are V. K. Naik, S. P. Midkiff, and J. E. Moreira. They published a paper called "A checkpointing strategy for scalable recovery on distributed parallel systems" in the Super Computing (SC) '97 conference, which took place in San Jose in November 1997.

A. Jeong and D. Shasha present a paper titled "P. Linda 2.0: A Transactional/Checkpointing Approach to Fault Tolerant Linda" in the Proceedings of the 13th Symposium on Reliable Distributed Systems, published by IEEE in 1994, spanning from pages 96-105.

CALYPSO, a software system created by A. Baratloo, P. Dasgupta, and Z. M. Kedem, allows for fault-tolerant parallel processing on distributed platforms. It was introduced at the Fourth IEEE Intel Symposium on High Performance Distributed Computing (HPDC-4) in August 1995 and its associated paper can be found on pages 122-129.

The citation for the work "A Malleable-Job System for Timeshared Parallel Machines" by L.V. Kale, S. Kumar, and J. DeSouza, presented at the 2nd IEEE/ACM International Symposium on Cluster Computing and the Grid (CCGrid 2002) in May 2002, is

as follows:

The citation for the work "A Malleable-Job System for Timeshared Parallel Machines" by L.V. Kale, S. Kumar, and J. DeSouza, presented at the 2nd IEEE/ACM International Symposium on Cluster Computing and the Grid (CCGrid 2002) in May 2002, is as follows:

E. Godard, S. Setia, and E. White conducted research on DyRecT, a software that provides support for adaptive parallelism on NOWs. This research was presented at the IPDPS Workshop on Runtime Systems for Parallel Programming in Cancun, Mexico in May 2000.

The authors' names are I. Foster and N. Karonis. They wrote a paper titled "A Grid-Enabled MPI: Message Passing in Heterogeneous Distributed Computing Systems". This paper was presented at SuperComputing 98 (SC98) in 1998.

The Supercomputing journal published a research conducted in 1995 by J.N.C. Arabe, A.B.B. Lowekamp, E. Seligman, M. Starkey, and P. Stephan on Dome, a system for parallel programming in a multi-user environment.

The text discusses a paper written by A. Agbaria and R. Friedman titled "Starfish: Fault-Tolerant Dynamic MPI Programs on Clusters of Workstations." The paper was presented at the 8th IEEE International Symposium on High Performance Distributed Computing in August 1999. It spans across pages 167-176.

The text below, which includes and their contents, can beand unified as follows:

G. Stellner presented a paper titled "CoCheck: Checkpointing and Process Migration for MPI" at the 10th International Parallel Processing Symposium (IPPS '96) in Honolulu, Hawaii in 1996. The paper was published in the symposium's proceedings on pages 526-531.

The article titled "User-Triggered Checkpointing: System-Independent and Scalable Application Recovery" by G. Deconinck and R. Lauwereins was presented at the 2nd IEEE Symposium on

Computers and Communications (ISCC97) held in Alexandria, Egypt in July 1997. It is published in the conference proceedings on pages 418-423.

C. Hofmeister and J. M. Purtilo conducted a study on Dynamic Reconfiguration in Distributed Systems in 1993. Their research focus was on adapting software modules for replacement. The study was presented at the 13th International Conference on Distributed Computing Systems in Pittsburgh, USA.