Abstract
This paper presents a parallel le object environment to support distributed array store on sharednothing distributed computing environments. Our environment enables programmers to extend the concept of array distributions from memory levels to le levels. It allows parallel I/O that facilitates the distribution of objects in an application. When objects are read and/or written by m ultiple applications using di erent distributions, we present a n o vel scheme to help programmers to select the best data distribution pattern according to minimum amount of remote data movements for the store of array objects on distributed le systems. Our selection scheme, to our best knowledge, is the rst work to attempt to optimize the distribution patterns in the secondary storage for HPF-like programs with inter-application cases. This is especially important for a class of problems called multiple disciplinary optimization (MDO) problems. Our testbed is built on an 8-node DEC Farm connected with an ethernet, FDDI, or ATM switch. Our experimental results with scienti c applications show that not only our parallel le system can provide aggregate bandwidths, but also our selection scheme e ectively reduce the communication tra cs for the system.
Introduction
High-performance distributed computing environments, which consist of a collection of high-performance machines connected via a high-speed network, can provide the aggregate computing powers necessary for large-scale scienti c applications. One of the critical issues in e ectively using these systems is the e cient transfer of data to and from secondary storage. In this paper, we i n vestigate the parallel I/O issues on a shared-nothing architecture from the viewpoints of languages and programming environments. Figure 1 shows an architecture of a shared-nothing high-performance distributed computing environment that we are dealing with in this paper. In the architecture, each processor is associated with a local disk, and the information exchange between disks of di erent processors has to go through an interconnection network. Recent research e orts in parallel programming languages have been concentrated on specifying regular dimension-wise data distribution patterns for arrays among parallel machines with distributed memories. Languages supporting the distributed array and data distribution concepts include F90-D 10], HPF 13], PC++ 14] 3], and up-coming HPC++ 22] . The data distribution concepts and SPMD programming model deliver scalable performance on parallel machines for the computational part. It however does not solve the problem with I/O part. The lack of parallel I/O supports creates two problems. First, I/O is executed serially, which results in performance bottlenecks according to Amdahl's law 1]. Second, in parallel languages such as HPF and PC++, the array is distributed among di erent processors. Due to the lack of support of distribution in the le level, each processor has to read the whole set of data from disk to memory and store them in a temporary bu er, and then assign the data into the distributed array i t o wns. This creates extra burdens on programmers to keep two set of data structures, and results in extra storage uses and program codes.
Introduction
In this paper, we extend the concept of array distributions in HPF 13] and pC++ 14] 21] 3] from memory levels to le levels. There are three key elements in this work. First, we support parallel le objects with random access. There is a unique name for each array object in a parallel I/O unit. A parallel I/O unit can be either a le or a pipe in the conventional Unix sense, but now it is a parallel le or pipe and has to be accessed by parallel le operators provided by our libraries. (For example, we n o w need to use pcat to cat names and contents of all array objects in a parallel I/O unit.) The access of array objects in a parallel I/O unit is no longer by sequential order but instead by the name of objects. The use of a unique name for object in the secondary storage environment helps us to track d o wn the access patterns to a particular object and allows us to do e cient implementation of parallel array object I/O. Second, when objects are read and/or written by multiple applications using di erent distributions, we p r o vide an interactive e n vironment for programmers to specify the inter-application I/O dependence, represented as a graph. We further provide a novel scheme to select the best data distribution pattern for array store in disk according to the minimum amount of remote data movements calculated from the I/O dependence graphs. This optimization is particularly important for a class of problems called multiple disciplinary optimization problem (MDO) in which v arious discipline codes interacting with one another to analyze the data 24]. For example, a realistic multidisciplinary optimization of a full aircraft con guration would require a number of discipline codes including aerodynamic analysis, structural design analysis, controls, performance analysis, etc, to interact with each other. Various discipline codes can be executed as a pipeline or executed asynchronously with data le being exchanged at various points in the code. The data distribution pattern of grid data that various discipline codes are using might b e di erent from each other, so it's crucial to select a good disk distribution pattern to optimize total execution time for a full set of applications. Our selection scheme, to our best knowledge, is the rst work to attempt to optimize the distribution patterns in the secondary storage for HPF-like programs with multiple disciplinary optimization problems. The third key element of our work is to support an implementation of parallel I/O libraries which can be used e ectively on shared nothing storage environments. This set of libraries can support array object with described distribution patterns for both disks and memories. When programs want to read/write array object with di erent distribution patterns between disk and memory, the system provides a collective c o m m unication library to e ciently support the functionalities. In addition, our system is capable of supporting a situation we call \out of con guration" execution, which can be described by an example as follows. Suppose we h a ve an application running with 4 processors. Let the processor set be P = fP 0 P 1 P 2 P 3 g. After the execution, the application uses our parallel write operator to write an data X :] into the disks of processor set P. A t a later time, another application is again running but with processor set Q = fP 0 P 1 g, and is reading the data X :] from disks of processor set P. W e call the parallel I/O operations beyond the processor set of the current SPMD programs as \out of con guration" operations. In such a n e n vironment, the set of processors running in a traditional SPMD mode have di culties in accessing the disks of the processor set which are not in the current running processor set. Our system solves this problem by installing a disk server on each processor, and providing a three phase protocols. This kind of \out of con guration" support is crucial for parallel le systems on shared nothing environments.
Introduction
Our parallel I/O system is currently incorporated into experimental HPF 19] 11] and parallel C++ programming environments 14] 15] based on a 8-node DEC Farm system. The parallel I/O system and programming environment is currently being used in a joint w ork with Power Mechanics Department, Tsing-Hua university to develop scalable methods to model three-dimensional gas turbine combustor model 17] 27]. Our experimental results with these applications show that not only our parallel le system can provide aggregate bandwidths, but also our selection scheme e ectively reduces the communication tra cs for the system.
Introduction
The remainder of the paper is organized as follows. Section 2 describes the related work. Section 3 presents the parallel I/O operations of our le system for distributed arrays. Section 4 describes a framework to select distribution patterns for arrays stores in the secondary storage. Next, Section 5 gives the design and implementation of parallel I/O libraries. Finally, Section 6 discusses experimental results, and Section 7 concludes this paper.
Related Work
The concept of extending the data distribution patterns from memory levels to le levels is rst pioneered by Brezany e t a l . 4 ]. Their work supports concurrent le operations on Vienna FORTRAN. The system in supporting data distribution for les can be basically classi ed into two categories: shared-storage system and shared-nothing environment. Brezany's work is based on a shared storage system, and is supported by concurrent le System (CFS) on Intel Paragon machines, while our work in this paper supports a parallel le system on shared nothing environments. The support of data distribution of a parallel le in a shared nothing environment is generally considered more di cult than that in a shared storage system, as more e orts are needed to enforce data sharing on shared-nothing storages. Follow-up work in supporting data distributions on shared-storage systems can also be seen in the work 2] 7] 8] 26] b y Choudhary et al. Choudhary's work provides the store of distributed array in the le system with a two phase access strategy scheme on CFS and has an extensive performance analysis of the le system. Both Brezany and Choudhary work are based on Fortran languages. The extension of parallel I/O with a C++ language on shared storage environment i s proposed by G o t wals 9]. Gotwals's work provides a parallel stream for a parallel extension of C++ language on Intel Paragons. Gotwals's work is done in parallel with our mechanism 16] which can be used in both HPF and parallel C++ languages. However, in contrast to Brezany, Choudhary, and Gotwals work, the access of array objects in our parallel I/O unit is not by s e q u e n tial order but instead by the name of objects. The use of a unique name for an object in the secondary storage environment helps us to track d o wn the access patterns to a particular object among a set of applications and allow us to do optimizations of parallel array object I/O.
Related Work
Related work in shared-nothing storage system includes IBM's Vesta 5], a scalable parallel I/O system at Argonne National Laboratory 23], and PIOUS 25] systems. Our work also falls into the category of shared nothing environments. IBM Vesta puts all the disks on the IO nodes so that their system is nonsymmetrical compared to our system. Vesta has two l a yer partitions. Programmers not only have the global view of the le but also specify these two l a yer partition patterns. If programmer uses these two l a yer partition properly, system can get better performance. Argonne's system is also built based on IBM SP2 system. This system provides a communication library helps applications to manage hundreds of I/O streams and to checkpoint and restart with di erent n umber of processors. PIOUS is a parallel le system provide by S t e v en A. Moyer and V. S. Sunderam. This system consists of a set of data servers, a service coordinator and library routines. PIOUS directly works compatibly with PVM environments, and does not support to the higher level of le distributions of HPF programs.
Related Work
Our system supports the data distribution of objects in the le level on shared nothing environment, and work compatible with experimental HPF and parallel C++ compilers. In addition, we address the issues of multiple disciplinary optimization problem. When objects are read and/or written by m ultiple applications using di erent distributions, we p r o vide a novel scheme to select the best data distribution pattern for the array store. The pattern is chosen with the minimum amount of remote data movements calculated from the I/O dependence graphs. This optimization is particularly important for a class of problems called multiple disciplinary optimization problem (MDO) in which v arious discipline codes interacting with each other to analyze the data les 24]. None of the existing work attempts to optimize the distribution patterns in the secondary storage for HPF-like programs with multiple disciplinary optimization problems.
Parallel Object I/O for Distributed Array
In this section, we present parallel I/O operations for distributed array objects. We will deal with the selection of I/O distributions for inter-application cases in the next section. If it is a pipe, the reader of the pipe has to use the unique name to access the object. If the named object in the pipe is not stored yet, the reader will be pending until the object is stored. Similarly, the access of array objects in a parallel I/O le is no longer by sequential order but instead by the name of objects. Only those arrays declared as "PARALLEL-FILE-OBJECT" can be used in operations of a parallel I/O unit. Ta b l e 1 g i v es a complete list of operators for a parallel le. These parallel I/O operators can be extended to parallel C++ language constructs in a similar way. The code segments shown later in Section 6 with our experiments will demonstrate the use of parallel I/O operators in a parallel C++ program.
Select Distributions on Inter-Applications Environments

I/O Models for Inter-Application Environments
Suppose we h a ve three applications, F1, F2, F3, and each application accesses a distributed array A and is reading or writing A from/to the same I/O unit using the parallel I/O operators. Suppose A is of size 16 16 , and is distributed as (BLOCK, *), (*, BLOCK), and (BLOCK, BLOCK) respectively in the memory in applications F1, F2, and F3. Figure 2 shows these three di erent distributions with 4 processors. The question is with what distribution should the array A reside in secondary storage. Our system helps programmers to select the best data distribution pattern for the store of array objects on distributed le systems according to the minimum amount of remote data movements. If (BLOCK, *) is chosen as the distribution scheme, application F1 incurs no remote data movements. However, there will be 192 remote data movements for (*,BLOCK) distribution in application F2, and 128 for (BLOCK,BLOCK) distribution in F3, and totally there are 320 data movements from remote disks. Similarly, there will be 384 remote data movements for the store of A in secondary storage if we select (*,BLOCK) as the distribution pattern, and 320 remote data movements for (BLOCK, BLOCK) distribution. Figure 3 shows the programming environment t h a t w e h a ve for the selection of I/O distribution patterns for multiple applications. When a set of applications that access an array object are ready to be executed, a programmer rst goes through an interactive I/O control panel (ICP), where they can describe the interapplication I/O dependence graphs. These dependence graphs are then solved by a solver to select the best data distribution schemes for the store of the array object on secondary storage. These distributions schemes are then fed to a compiler to generate e cient I/O codes. After the execution, pro ling information can be sent back to ICP to update the inter-application I/O dependence graphs if necessary.
I/O Models for Inter-Application Environments
Currently, the solver is able to select the best data distribution schemes for two t ypes of inter-application I/O dependence graphs:
I/O Models for Inter-Application Environments
Each application is annotated with a frequency to denote how frequently the application is invoked. Also each application is associated with a distribution pattern in which the distributed array o b j e c t is distributed among processor memories. The inter-application I/O dependence graph in Figure 4 is an example of the Independent E v ent Model. The read/write access patterns to an le object are independent e v ents in this model.
Multiple-Stage Precedence M o del
The inter-application I/O dependence graph in Figure 5 is an example of the multiple-stage precedence 
Solvers
A solver selects the best data distribution scheme for the store of the array object on secondary storage according to the speci ed I/O graphs. Here, we rst introduce the solver for the Independent E v ent M o d e l . For simplicity, w e rst assume that there is only one parallel I/O unit in the system. Our scheme works under the assumption that the processor allocation scheme is statically decided, and our goal is to nd a distribution such that
Solvers
where f i is the execution frequency of ith Application, D i is the memory distribution pattern of the object in the ith application, m is the number of applications, and
Solvers
where D is the domain of all possible distribution schemes, d is the number of processor dimension, and ( D i ) is the amount of remote data movements for a given distribution under the condition that the application is using D i at the memory distribution scheme, and it can be calculated by
Solvers
The function NumOfRemoteRef p calculates the number of remote references for any g i v en processor p. The function NumOfRemoteRef p can be calculated as follows. Suppose the distribution of an array o n a dimension is BLOCK or CYCLIC, and then the element in processor p can be described as begin 1 : end 1 : stride 1 and similarly the elements resided in the disk of node p can also be described as begin 2 : end 2 : stride 2 . The number of local references is equivalent t o t h e n umber of non-negative i n teger pairs (x, y) such t h a t MIN(a,b) ). a and b are strides and will be 1 in the case of BLOCK distribution and equal to the number of processor in the case of CYCLIC distribution.
Solvers
Thus ( D i ) can be computed in O(p Log(p)) time, where p is the number of processors in the system. The above complexity is certainly true for the one-dimension case. For multiple dimensional array, w e n e e d further explanation below.
Solvers
Suppose we h a ve a d dimensional array, and the processor array in the memory and disk is (p 1 p 2 888a:dbb7:2d49:b3f6:8605:bf4e:caed:e5a4: p d ) and (p In the case that the distribution is a Block-Cyclic distributions, we need extra e orts to calculate the function NumOfRemoteRef p . Suppose that we h a ve memory distribution B(N 1 ) and disk partition as B(N 2 ), we will try to nd x, y, i, j, such that begin1 + stride1 x + i = begin2 + stride2 y + j where 0 i < N 1 0 j < N 2 Therefore, begin1 + stride1 x = begin2 + stride2 y + ( j ; i) w h e r e 0 i < N 1 0 j < N 2 We need to solve the diophantine equation MA X(N 1 N 2 ) times. Assume N 1 is bigger than N 2 , ( D i ) can be computed in O(p N 1 Log(p) ) time, where p is the number of processors in the system.
Solver for Multiple-Stage Precedence Model
Now let's develop the solver for Multiple-Stage Precedence Model. In this model, we h a ve an adaptive distribution scheme for then array object in di erent stages. Our goal is to nd an arbitrary sequence of (x 1 x 2 888a:dbb7:2d49:b3f6:8605:bf4e:caed:e5a4: x k ), to Minimize k j=1 (j x j ) + (x j x j+1 ) where (j x j ) is the amount of remote data movement at stage j, g i v en the distribution pattern x j , k is the number of stages, x i 2 D d 1 i k, a n d is a reshape cost from one shape to another shape.
Solver for Multiple-Stage Precedence Model
where f i j is the execution frequency of ith application in the jth stage, D i j is the distribution pattern of the object in the ith application of the jthstage, and and can be calculated in the same way a s t h o s e in the Independent Model. Figure 6 shows a standard three level data mapping models in the processing of an array o b j e c t i n a H P F language. Our work in the previous section can only work with distribution directives, and here we extend our framework to work with alignments.
Extended Frameworks for Solvers
Assume that array A is aligned with B, and B is distributed by \BLOCK" distribution as shown in the program code segment below. Similarly, when array B is distributed by \Cyclic" distribution the Index A (pid) can be described by a triplet, begin : end : stride]. We can get the triplet by solving the diophantine equation below a i + b = pid + j NPROC 0 i < N 0 j < N = NPR O C Since the element index in the processor pid can be described as a triplet, we can calculate the function NumOfRemoteRef p by the method used in Section 4.2. Suppose the distribution of an array on a dimension is BLOCK or CYCLIC, and the element in processor p can be described as begin 1 : end 1 : stride 1 and similarly the elements resided in the disk of node p can also be described as begin 2 : end 2 : stride 2 . The number of local references is equivalent to the number of non-negative i n teger pairs (x, y) s u c h that begin 1 + stride 1 x = begin 2 + stride 2 y where begin 1 + stride 1 x MIN(end 1 end 2 ):
Extended Frameworks for Solvers
In the case that the distribution is a Block-Cyclic distribution, we need extra e orts to calculate the function NumOfRemoteRef p . Suppose that array B is distributed by block-cyclic distribution, say B ( k 1 ). Then Index A (pid) can no longer be described a a triplet, but instead it can be described by k 1 triplets, c i + d = pid + j NPROC + offset 0 offset k 1 Suppose that we h a ve memory distribution B(k 1 ) and disk partition as B(k 2 ). Then the element i n processor p can be described as k 1 triplet, begin i : end i : stride i 0 i < k 1 and similarly the elements resided in the disk of node p can also be described by k 2 triplet, begin 0 j : end The complexity in nding ( D i ) i n t h i s c a s e i s O ( p k 1 k 2 Log p ). Finally, our scheme works well with or without the optional mapping between abstract processor array and physical array described in Figure 6 . If the abstract processor array i s t h e p h ysical array, our scheme can be applied immediately. If the optional mapping exists, we only need to do a mapping between the physical processor number and the abstract processor number when calculating the remote reference numbers. All of our frameworks can be applied directly.
Design of Parallel I/O Libraries

File Structure Design
In our parallel le system, each parallel le is distributed among disks of machines. In the disk of each machine, we h a ve a parallel le structure and a data set of distributed array objects for each le. The parallel le structure records the information of a parallel le. The information includes the type of the le, the number of objects in the le, the distribution con guration for each object in the le, the disk address to denote where the object is located, the name tag for each object, the processor con guration, and the size of each object. These information can be used to locate the positions of the data set of objects among processors and disks.
File Structure Design
When users want to create a parallel le in our system, the system creates an ordinary unix le on each node named as user speci ed parallel le name and appended node numb e r a s p o s t x . Then each corresponding le of each processor will have only partial data set of distributed array objects. This le structure is used as the basis of our parallel le system.
Collective Communication Library
In the case that we w ant to read an array object from disks, the system rst pack data from local disks into memory bu ers. If the distribution of the object in the memory is the same as that in the disk, each processor read the data from the memory bu ers with SPMD mode. On the other hand, if the distribution of the object in the memory is di erent from that in the disk, our system provides a collective c o m m unication library to exchange data into its correct positions. The collective c o m m unication library will compute the sending set and receiving set. Sending set is the set of data that that each local processor will send to other processors, and receiving set is the set of data that each local processor will receive from other processors. When the collective communication is done, data is then placed into the correct memory addresses from the memory bu ers. A parallel write operation can be done similarly.
Collective Communication Library
The sending set and receiving set in our collective c o m m unication library for parallel I/O operations can be modeled by extending the concepts in the data movements in the memory level on distributed memory environments 12]. We will explain our revision as follows. Suppose A is an array, a n d p and q denote processor numbers. The essential set operations involved in a collective c o m m unication library implementation a r e described as follows.
Collective Communication Library
local A (p) w h i c h represents the set of data owned by processor p is de ned as follows.
Collective Communication Library
local A (p) = fa j a is a member ofA and a is stored on pg send set(p q) that represents the set of data processor p will send to processor q is de ned as follows.
Collective Communication Library
send set A (p q) = fa j a is a member of A and must be sent from p to qg recv set A (p q) w h i c h represents the set of data processor p will receive from processor q is de ned as follows.
Collective Communication Library
recv set(p q) = fa j a is a member of A and p must receive a from qg We then have the following two basic properties.
Collective Communication Library
Lemma 1 Assume that the owner computing rule is used, and let A be the array object to be written from memory to a le, and denoted by Disk Similarly, if A(N) is with Cyclic Distribution and distributed on #P processors, the elements on processor pid can also be described by a triplet.
Collective Communication Library
We will give an example in calculating send set and recv set below:
Collective Communication Library
Example 1 Assume that the owner computing rule is used, and let A(10) be the array object to be read from 
Out of Con guration Support
Suppose we h a ve application A running with 4 processors. Let the processor set be P = fP 0 P 1 P 2 P 3 g.
Out of Con guration Support
After the execution, application A uses our parallel write operator to write an data X :] into the disks of processor set P. Later, application B is running but with processor set Q = fP 0 P 1 g, and is reading the data X :] from disks of processor set P. Figure 7 illustrates the data movements of such a n e n vironment. We call the parallel I/O operations beyond the processor set of the current SPMD programs as \out of con guration support". In such a n e n vironment, the processors running in traditional SPMD mode have di culties in accessing the disks of the processor set outside the current running processor set. Figure 8 describes our three phase protocols, OCP (Out of Con guration Protocols) between the program and the server when a parallel read operation is issued. The program rst check if a \out of con guration" parallel read operation is invoked. If an out of con guration read operation does happen, processor 0 then sends a parallel read request to the server of each processor which is not in the current con guration, but possesses data. This daemon in the server site rst waits and listens to request. Upon receiving the parallel read request, it does handshake with processor 0 to get the con guration and object information. In the second phase, the server reads the local collection of the array object from the disk, and waits to receive the sending set and receiving set information from each process running SPMD program. Finally after the server receives the sending set, it then sends data to each of the SPMD program according to the sending set it receives.
Out of Con guration Support
Similarly, when a parallel write operation is issued, a program rst checks if a \out of con guration" parallel write operation is invoked. If an out of con guration write operation does get invoked, processor 0 then sends a parallel write request to the server of each processor which is outside the current con guration, but the data will be written to. This daemon in the server site rst waits and listens to request. Upon it receives the parallel write request, one of the SPMD processor continues to send the con guration and object information to each of the servers. In the second phase, the server receives receiving set and the data from each of the SPMD processors. Finally in the third phase, it stores the data into local disk according to the receiving set.
Experimental Results
Our parallel le system is currently being incorporated with an experimental parallel C++ and HighPerformance Fortran compiler on a 8-node DEC Farm. Figure 3 in Section 4.1 showed the full system view for the parallel programming environments. This system is also used for the development o f c o mputational uid dynamic applications in a joint e ort with scientists in the Power Mechanics Department, Tsing-Hua university 17] 27]. In the remainder of this section, we will present three sets of experiments done on our parallel le systems.
Experiments with Primitive Operations
The rst set of our experiments is conducted to evaluate the basic functionalities of our parallel I/O operators. Table 2 shows the time spent and the speedup gures for parallel write operations. The experiment is done with the memory and disk having the same distribution patterns. The size of the array in the experiment i s 128 by 128 with each element o f t h e a r r a y h a ving 64 bytes. We experiment with basic distribution patterns including (Block,Block), (Block,Cyclic), (Cyclic,Cyclic), (Cyclic,Block), and (Block,*). All of them exhibit close to linearly speedup and deliver aggregate bandwidth as processor number grows. The time listed is measured by running the parallel read and write operations over 100 times. We observe their behavior similar to each other in spite of di erent distribution patterns used. Similarly, parallel-read operations also exhibits similar performance speedup in our experiment.
Experiments with Primitive Operations
The experiment listed above is done when the data distribution pattern of the object in the memory is the same as the distribution in the disk. Our system also supports the parallel I/O operator when the memory object has di erent distribution from the disk object. In the following, we conduct an experiment in our system with such cases. In our experiment, we h a ve a n a r r a y object stored in the disk with (Block, Block) distribution, and has to be read from the disk into the memory with (Block, Cyclic) distribution pattern. The array sizes listed used are 256 by 1 2 8 , 1 2 8 b y 128, and 128 by 64, respectively. The size of each element is again 64 bytes. The disk data is rst fetched into the local processor, and then our collective communicationlibrary described in Section 5 is invoked to exchange data elements between di erent processors. The collective communication time is bound by the total number of remote references and the network bandwidths. Table 3 In these experiments, our parallel le system can not only provide aggregate bandwidths when the data distribution pattern of an object in the memory is the same as the distribution in the disk, but also provide e cient supports of a collective c o m m unication library on high-speed networks when the memory object is associated with a di erent distribution from that of the disk object.
Scienti c Applications Using Parallel I/O
In our second set of experiments, our parallel le system is being used for actual software development with parallel computational uid dynamic applications to reduce I/O bottlenecks. This is a joint w ork with the Power Mechanics Department, Tsing-Hua university. Currently, t wo uid dynamic codes are employing the parallel I/O operations to parallelize the original serial part of I/O statements. These codes are being developed by using a parallel C++ language 14] 15] with our parallel I/O library. The rst application in our experiment is a gas-turbine-combustor model for simulating dilution jets 17]. The 3D grid is partitioned among processors and the data distribution scheme is illustrated in Figure 9 . Two of the three dimensions are distributed among processors while the dimension along the k direction is not distributed. The data distribution is Block,Block,*]. The data structure represented in a parallel C++ programs is shown below. The above is an extended syntax of an experimental parallel C++ language 14]. We declare a distributed array of \SubGrid" with the array being distributed by Block,Block] distribution. The element o f a n a r r a y The algorithm is basically divided into three steps. In the rst step, a nite di erence scheme (QUICK scheme) in a 3D plane with 32 points in the stencil pattern is used. After the nite di erence scheme is completed, we n e e d t o s o l v e a group of linear systems using ADI methods. At this point in each iteration, the uid data of the three-dimensional grids are written out by our parallel I/O operators. The periodical output of the uid data can make the checkpoints to restart the program in case the machine is down and can be used for programmers to observe (or visualize) the progress of a program. Using the parallel I/O library we can parallelize both the read of the initial grid data and the output of the grid data in each iteration. The parallelization of I/O statements can reduce the time with I/O statements, simplify program coding e orts, and reduce the storage consumptions. Without using a parallel write operations, the program originally had to collect all the data of the distributed array i n to one processor and then write the data to a disk from one processor. Table 4 compares the time spent b e t ween a parallel and sequential write operation with the gas-turbine-combustor model. In this experiment, the size of the distributed array i s 3 2 b y 16, and the size of each SubGrid is 3296 bytes, as it contains the third dimension of the structures and many essential information for uid computations. The total size of the distributed array is around 1.68MB. With parallel write operations, it only takes 0.10 seconds on a 8 processor DEC ALPHA Farm with Ethernet connections to write the data into the disk in parallel. With sequential write operations, all of the processors has to rst move the data through network into processor 0, which t a k es 3.40 seconds. And then it takes 0.62 seconds to nish the disk write operations. A parallel write operation signi cantly out-perform a sequential write operation.
Scienti c Applications Using Parallel I/O
The second application in using our parallel I/O libraries is a parallel vortex method to simulate the turbulence of a three-dimensionally evolving jet. Vortex method 20] is widely used to simulate vortex induced ow problems, and is adopted to simulate the evolution of three-dimensionally periodical jet under axial perturbations. The global interaction of all point v ortexes is a typical N-body problem with an operation Similar to the rst application, this application is using parallel I/O to read the initial vortex data and output the vortex data in each iteration. In this experiment, there are 16 rings and each ring can hold up to 1000 vortex elements. The total size of the structure is around 1.1MB. With parallel write operations, it only takes 0.07 seconds on a 8 processor DEC ALPHA Farm with Ethernet connections. With sequential write operations, it rst moves the data into one processor which t a k es 2.13 seconds, and then it takes 0.42 seconds to nish the disk write operations. Totally, a parallel write operation is almost 36 times faster than a sequential write operation after adding the factor of network tra cs. This is calculated from the formula below. The selection of a data distribution pattern for the store of array object in the le system is particularly useful in a multiple disciplinary optimization problem (MDO) in which v arious discipline codes interacting with each other to analyze the data. Figure 11 shows the I/O precedence graphs in our model for a simpli ed sample of MDO applications 24] with an aircraft design. A realistic multidisciplinary optimization of a full aircraft con guration would require a number of discipline codes including aerodynamic analysis, structural design analysis, control system analysis, performance analysis, etc, to interact with each other. Various discipline codes can be executed as a pipeline or executed asynchronously with data le being exchanged at various points in the code. In the sample application in Figure 11 , optimizer, nite element s o l v er, and ow solver all access the surface geometry information object which is stored in a parallel le. The data distribution pattern of the surface geometry information used in the optimizer, nite element solver, and ow solver might be di erent. In this case, selecting a good disk distribution pattern by using our proposed scheme in Section 4 can decrease total execution time of a full system. Figure 12 plugs in a template data set to model MDO problem. Suppose the optimizer, ow s o l v er, and nite element solver uses the grid object with (Block,Block), (Block,Cyclic), and (Cyclic,Cyclic) distribution patterns, respectively and there are totally four processors in use. Figure 12 shows the basic data movement ratio between di erent distributions. The number associated with the edge represents the data movement ratio from memory to disk for the two di erent distributions. The numbersuch as 1/2 represents the amount of remote data movement o ver the size of the array object. Let's assume that each application has the same frequency in accessing the object in disk. By employing our dynamic programming algorithm in selecting the best data distribution pattern, a selection of (Block,Cyclic) as the disk distribution will minimize the where f i is the execution frequency of ith application, and ( D i ) is the amount of remote data movements for a given distribution under the condition that the application is using D i at the memory distribution scheme.
Scienti c Applications Using Parallel I/O
Similarly, Figure 13 illustrates the same idea but with 8 processors. The diagram shows the remote data movement from memory to disk with di erent distributions. The (Block, *) distribution is to distribute the rst dimension of the array b y b l o c k, but there is no distribution for the second dimension of the array. It's used as one extra reference point t o s h o w the data movement rate with di erent distribution pattern. Suppose we still use the same assumption with the four processor case, and apply the selection algorithm in Section 4, we will nd the (Block,Cyclic) distribution is the best distribution.
Scienti c Applications Using Parallel I/O
Focus is now directed to Table 7 : Predicted Data Movement Ratio vs. Experimented Performance on 8 Nodes optimizer, ow s o l v er, and nite element solver uses the grid object with (Block,Block), (Block,Cyclic), and (Cyclic,Cyclic) distribution patterns in the memory respectively, and let N be the size of the array object. The predicted smallest remote movement ratio is N if we c hoose (Block,Cyclic) as the disk distribution. The actual execution time is more than 30% improvement o ver the next best selection. The array size in this experiment i s 6 4 b y 64, and each element o f t h e a r r a y is again of size 64 bytes. The time measured in the table is in the unit of 10 ;2 second. Note that due to CSMA/CD behavior in the ethernet, the movement ratio is not linearly proportional to the actual execution time. In one instance in our experiment, when the data movement ratio increases from one 1/2*N to 3/4*N, the actual communication time increases nearly 100 percents. This observation further underlines the importance of pattern selections. Table 7 shows a similar result, but experimented on an 8-node DEC Farm. Finally, our solver for selecting proper distributions for parallel le system is very e cient i n o u r e xperiments. For example, in one of the test run, with array size 4000 4000, processor array size 4 4, array object distribution scheme B(10), B(10)], and le array distribution Cyclic,Cyclic], it only takes 0.034 seconds at a SUN Sparc 10/30 machine to decide the number of remote references needed.
Conclusion
This paper describes a parallel le object environment to support distributed array store on shared-nothing distributed computing environments. Our environment enables programmers to extend the concept of array distribution from memory levels to le levels. It allows parallel I/O according to the distribution of objects in an application. Currently, our parallel I/O library is incorporated into an experimental parallel C++ and a subet set of HPF compiler on a 8-node DEC Farm with Ethernet or FDDI connections. In addition, this paper proposes a selection scheme to help programmers to select the best data distribution pattern according to the minimum amount of remote data movements for the store of array objects on distributed le systems, when objects are read and/or written by m ultiple applications using di erent distributions. This selection scheme, to our best knowledge, is the rst work to optimize the distribution patterns in the secondary storage with HPF-like programs for multiple disciplinary applications. Our experimental result shows that not only our parallel le system can provide aggregate bandwidths, but also our selection scheme e ectively reduce the communication tra cs for the system.
22


{'paper_id': '11804434', 'title': 'Parallel Array Object I/O Support on Distributed Environments', 's2_url': 'https://api.semanticscholar.org/CorpusID:11804434', 'source': 'S2ORC'}