The design and implementation of cluster-based distributed applications can be complex for those who are unfamiliar with a distributed data-parallel paradigm. Moreover, the growing interest in Beowulf-type supercomputers motivates a need to provide a robust framework for those who would like to experiment with the capabilities of a Beowulf quickly with minimal effort. However, the task of building a Beowulf and the subsequent design and development of distributed applications can be difficult the first time one undertakes this task. The goal of this project is to provide a solid conceptual framework and concrete platform for rapid prototyping and development of data-parallel applications by exploiting the concepts of Object-Oriented (OO) development and Design Patterns

Distributed STORMS attempts to reach this goal using the Java programming language and the Extensible Modeling Language (XML). A framework is provided for the development of pluggable, reusable Java components based on several Design Patterns. These components are then molded into XML-based instruction lists.

Distributed STORMS is part of a larger existing project at Furman University supporting the Themosphere, Ionosphere, Mesophere, Energetics and Dynamics (TIMED) satellite mission based at Goddard Space Flight Center. The larger project has been focused on the improvement and extension of several NASA Global Circulation Models. Most of these applications are written in FORTRAN for multi-processor supercomputers. Moving these NASA Global Circulation Models to Beowulf Cluster supercomputers has been found a time-consuming and error-proned task. The difficulties intrinsic to the porting process motivated the need for a general-purpose conceptual model focused on the data-parallel paradigm on Beowulf Cluster supercomputers and a corresponding software framework implemented in Java. The following extended abstract highlights the results of the effort to develop such a framework.

The larger project at Furman has had contributions from many students over the years. However, the genesis for this particular piece came from discussions and debates between the student and the directing professor. After assistance from the directing professor at the onset -- including explanations of existing work and debates about the requirements and design of Distributed STORMS -- the student was allowed the freedom to pursue the completion of the design with reviews by the professor.

 Background and Related Work

Developing a distributed platform in Java brings with it disadvantages such as Java's reputation for poor performance. Distributed STORMS does not seek to replace high-performance systems for distributed computing based in C or FORTRAN, but rather to extend this environment to Java developers. However, existing projects have demonstrated that Java does have potential in high-performance computing.

Riley et al. [1] demonstrate the development of two customized OO applications implemented in Java for problems in Computational Fluid Dynamics. Each Java application in their project had a pre-existing corresponding Fortran or C implementation. Furthermore, a procedural instance and OO instance of the Java version was produced. They demonstrated that the two applications ported to Java perform within a factor of 2.5 to 3 of their original Fortran and C counterparts -- and that between each Java version, the OO performance was close to the performance of the procedural version.

As noted by Moreira et al. [2], "there are no technical barriers to high performance computing in Java." Numerically Intensive Java (NINJA) [2] is an effort at IBM to provide robust, efficient numerical analysis for the Java Platform. They demonstrate the performance of Java (version 1.1.6) can be improved by orders of magnitude [2] by utilizing straightforward optimizations to improve the performance of the Java Virtual Machine (JVM). Their implementation solves several problems including optimizations to the Java exception handling model, Java arrays, and the utilization of complex numbers. Just as they note the economical solution of reusing existing components in their project, we plan to build on their results in our final implementation as well. NINJA supplies a library that can be reused in applications built using the Distributed STORMS framework.

Distributed STORMS provides a user-friendly general purpose platform for writing cluster-based data-parallel distributed applications using the Java programming language and the Extensible Modeling Language (XML). Distributed STORMS uses pluggable, reusable Java components based on a combination of several Design Patterns including Abstract Factory, Strategy, Command, and Observer patterns [3]. These components are then molded into XML-based instruction lists. Moreover, by their very nature these components can be formed into larger components recursively.

The architectural framework proposed has several key components. These are Workers, Tasks, Strategies, Algorithms, and Communicators. Each of these will be addressed, but first an overview is provided to demonstrate how they interact.

Workers provide an agent-based representation of the Beowulf cluster. The primary function of a Worker is to provide common services to its clients, such as access to data and dynamic loading of Strategies. There are two kinds of data – data that is local to the currently executing Strategy on a single processor, and data that is global, and thus accessible to every Worker in the cluster. Tasks are provided as an abstraction to encapsulate distributed processes. They can be loaded and queued as needed. Multiple Tasks can run on a single Worker. Tasks execute Strategies. Strategies and Algorithms encapsulate the actions that need to take place on a cluster. The distinction between Strategies and Algorithms is a differentiation in actions that can be paused and restarted and those that cannot. Strategies interact with each other on the local machine through the Local Context. The Local Context enables Strategies to initialize, locate the data they are supposed to act upon, modify that data as needed, and leave it ready for the next Strategy to execute. Because the Local Context is persistent between Strategies, Strategies can be "hooked" together using a simple XML language. When Tasks need to distribute information or garner information with the rest of the cluster, they use Communicators. A Communicator interacts with the Distributed Context, and thereby disseminates or collects data from the rest of the cluster. Because of Java’s inherent support for multi-threading applications, Communicators and Strategies can execute simultaneous forming a pipeline whereby Communicators pump data into the Local Context where Strategies can then act upon it as necessary.

One of the focuses of Distributed STORMS is to provide a framework that will enable rapid prototyping. XML is used to tie Strategies and Communicators together into a specific order. XML is used to define Tasks, providing a standard way to tie Strategies and Communicators into a single application, while leaving the flexibility for individual Strategies and Communicators to define their own XML parameters as needed. When a Strategy or Communicator is initialized, it will be passed all of the parameters specified in the XML node that defines it. More information on how the XML format is defined and how the XML translates into an application is provided later.

A Task component is a collective action that can be executed to solve a larger problem. It is built entirely from Extensible Markup Language (XML), which is platform and application independent. These XML scripts that pull together Strategies and Communicators as needed to accomplish some action. Tasks have direct access to the services provided by Workers. They provide a buffer zone and scope for the execution of Strategies the Beowulf cluster.

The Task XML format is defined using a Document-Type Definition (DTD). Parsers use these specifications to validate XML documents when they are interpreted. A DTD also serve as a rudimentary form of documentation to indicate what is or is not permitted in the XML language being developed. A partial listing of the Task DTD used in the prototype follows. Because Workers process Tasks, a different Worker implementation could use a different format.

 The Task DTD defines four primary structured elements: task, exec-control, communicator and strategy. The task element provides a wrapper and indicates how many processors should be allocated. In the prototype, the value of requestProcessors is ignored, but it is left in the format for future load balancing extensions. A task element contains one or more exec-control elements. The exec-control elements provide simple looping functionality and conditional gateways. Which attributes are present and the value of those attributes determine the function of a particular exec-control element. When no attributes are present, the contents of an exec-control are executed once and only once.

The exec-control elements can also contain other exec-control elements to allow for nested looping. In general, any complex looping structures should be moved into a Strategy in order to keep Tasks simple. The exec-control elements also contain strategy elements that define a reference to a particular Strategy class and communicator elements that define a reference to particular Communicators.

The strategy, communicator, and algorithm elements can reference their individual components (eg Strategy, Communicator, and Algorithm) either by class (a combination of the name and package attributes) or by a link to another XML file that defines the Strategy, Communicator, or Algorithm. The format of a linked XML file is the same as the specified Task DTD.

To demonstrate how the strategy and communicator XML elements define references to Strategies and Communicators, several examples follow. The first three examples depict how references are defined to Strategies. The source code for these listings may be found in Section 3.4.

In Listing 3, the AssembleData Strategy specifies a parameter "filename" that indicates from where the data should be loaded. The parameter "storeResult" specifies a location in the Local Context of the Worker to store the results loaded from the data file. When the Strategy is initialized, it will be passed an XML node that establishes Static Context, or unchanging input parameters that are necessary for its execution.

A Worker component provides the foundation for Distributed STORMS. The Worker class provides a management container that loads, executes, and destroys Tasks as needed. Workers also provide facilities to dynamically load Strategies, Communicators and Algorithms. As a Task is loaded or executed, the actions it references will be identified, located, and loaded by the Worker as needed. The Worker maintains important information such as the processor rank the total number of processors in the cluster, which is necessary for most data-parallel applications.

The Worker processes the Task XML file sequentially by examining each XML element. As described above, these elements reference Strategies, Communicators, and Algorithms that are instantiated per the XML statement and then executed.

A Worker provides context to executing Tasks by providing access to the Local Context, which is an index-accessible data structure (eg Hashtable) on the local machine. Strategies can store and retrieve information from the Local Context. Using the param elements, direction can be given to Strategies to indicate the names of input parameters and the names of results. These names are used as keys to load and store data within the Local Context. Once an Algorithm completes its execution, it can then leave information and data in the Local Context for the next executing Algorithm.

The metaphor of Distributed Context provides an interface for inter-Worker communication. The interface provided by both the Local Context and Distributed Context allow different implementations of the Distributed STORMS architecture to utilize different communication subsystems such as mpiJava, which is a Java-based binding of the Message Passing Interface (MPI) specification. A Communicator follows the a combination of the Strategy and Command patterns much like a Strategy, but its core function is to interact with the Distributed Context. The abstraction and interface between Strategies and Communicators permits different Communicators to be substituted without modifying the rest of a Distributed STORMS application.

The Worker life cycle consists of five steps described in Figure 6. In the initialization stage, the Worker must initialize the value of clusterSize and determine the rank of the current Worker in the cluster. The Worker should also seek out the Task XML file that specifies its actions. The details of the filename and location may vary based on the implementation of the Distributed STORMS framework. Once a Worker is initialized, it has the option to perform another action before it begins loading and executing Tasks. When the framework is implemented, some implementations may choose to use the preOperate method, but as of this writing it is not used.

The Worker spends most of its life in the operate() step. During the operate step, the Worker continuously checks its WorkQueue for Tasks. A WorkQueue is a simple data structure for queuing Tasks until they can be executed. Since Tasks implement the java.lang.Runnable interface, the Worker instantiates a Task and passes it as an argument to a Thread object. The Worker invokes the start() method of the Thread, and then the Task begins processing the XML instructions – loading and executing Strategies as needed.

 Results and Contribution

As mentioned in the Project Motivation, the responsibility to design the Distributed STORMS framework was yielded to the student with reviews by the professor.

Distributed STORMS has proven an effective conceptual model for data-parallel development, fulfilling the original goal of the project. The definition of Local and Distributed Contexts simplifies access to data within a data-parallel environment through a standard interface. Strategies and Algorithms within a Worker interact using a Local Context and individual Workers interact using Communicators. With a standard set of Communicators, developers may focus more on the dependencies and implementation of Strategies and Algorithms, and focus less on the details of communication on a Beowulf Cluster.

The Distributed STORMS framework has been applied to paralleled array multiplication as a proof of concept, demonstrating that it is effective. The next application of Distributed STORMS will be applying it to the remaining NASA Global Circulation Models (GCMs) discussed in the Project Motivation. Distributed STORMS also provides a vehicle for other related projects at Furman supporting the TIMED satellite mission, including the method of Recursive Substructuring, which restructures matrices of a particular form, common in the models, in order to improve the computation of the models.

[1] Riley C J, Chatterjee S, Biswas R. "High Performance Java Codes for Computational Fluid Dynamics." Proceedings of the Joint ACM Java Grande - ISCOPE 2001 Conference. Stanford, CA. June 2001

[2] Moreira Jose E, Midkiff Samuel P, Gupta Manish, Artigas Pedro, Wu Peng, Almasi George. "The NINJA Project: Making Java Work for High Performance Computing." Communications of the ACM. v.44, n.10, pgs 102-109. 2001.

[3] Gamma Erich, Johnson Ralph, Helm, Vlissdes John. Design Patterns: Elements of Object Oriented Software. Addison-Wesley. 1994.