Parallel Processing with PIT

For some years I have been interested in parallel algebraic computation using Beowulf Clusters. A Beowulf Cluster is essentially a loosely-coupled multiprocessor configuration, where the coupling is via network connections and the individual processors are most often ordinary personal computers running Linux. In order to make use of such a cluster for parallel computation, one typically builds one's parallel application on top of either PVM, the Parallel Virtual Machine software, or on an implementation of MPI, the Message Passing Interface standard. Both systems are designed to provide a communications substructure for implementing parallel algorithms to run on networked clusters.

My initial foray into parallel programming was with PTAH, the Parallel Tensor Algebra Hybrid system. PTAH was essentially a C++ program which made extensive use of Mathematica for problem preparation and analysis. The underlying cluster software used by PTAH was PVM. I chose PVM because at the time, the API made it simpler to build fault tolerance into my application: I did not want a hardware or software malfunction on a slave processor to prematurely terminate the entire computation. While debugging PTAH I found that PVM was not entirely as good at preserving message ordering as my application required, necessitating a great deal of additional programming on my part. At the time, I realized that if I only knew how to program with sockets I could lose PVM altogether. From this desire, PIT was born.

PIT stands for "Parallelize It". Many parallel applications are essentially the parallel operation of serial programs on multiple sets of data. For instance, it is often useful to run the same program many times, each time with a different parameter set. This is especially applicable when the majority of the input data is the same for each run, and when the output is relatively brief, permitting straightforward collation of the results. In such an environment, PIT allows a serial program to be run multiple times, distributed across a Beowulf Cluster, without modification to the serial program.

One of the design goals for PIT was that it be Unix-like: a small sharp tool* for a specific purpose, that interoperates well with the rest of the system. Input to PIT is a workload file consisting of the command lines necessary to process the workload. Each command line must be independent of the others. The master instance of PIT simply distributes these commands to the slave instances, which then execute the commands and report back to the master for additional work.

This implies that all software and data necessary to execute each command must be pre-loaded on every slave system. Command output must of course be stored for later retrieval. Since the master PIT program discovers new slave instances as they are started, the cluster is automatically reconfigurable while processing. If a slave process dies (or TCP Keep Alive fails), the master reassigns the parcel to the next slave available. When the slave is restarted, it will be re-included into the cluster and work will be scheduled to it as available. If the master is terminated before the workload is complete, it will recover via a progress file, rescheduling parcels that were in progress when the termination occurred (there are two progress files, updated alternately: command line option determines which file is used for recovery). Work in process when the master died may be duplicated, but work can never be lost.

This simple approach allows for rough load balancing on inhomogeneous clusters, since multiple slave instances can be started on more powerful cluster members. In addition, PIT could be relatively easily modified to run on clusters of inhomogeneous architecture, since the only information passed is in the form of Unix command lines and status messages.

PIT provides simple status reporting: SIGHUP causes the master process to dump its current status. SIGHUP also causes a slave process to cleanly exit after the current command is completed. A generic Makefile can be used to build PIT from the C source code. When constructing the workload file, string parameters should either have no spaces or be enclosed by quotes or apostrophes. PIT currently uses the parallel TCP port assigned to GAUSStm, but another port can be easily inserted in the assignment for the variable parallel_service_entry. Invoking PIT with the -h option produces the following useful information: 

usage: pit -m (-n expected_number_of_slaves) (-p < progress_file_name >) (< workload_file_name >)
       pit < master_host_name_or_ip >
       pit -h
            may also add -d for parcel debugging
defaults: n = 64, < progress_file_name > = pit.progress, < workload_file_name > = pit.workload

The first line would be used to invoke the master instance of PIT, while the second would invoke a slave instance. The parameter n is not a limit; choosing it wisely simply eliminates a little extra overhead.

PIT currently has no built-in security measures and therefore must be run on a secure LAN or VPN. It has been tested on a small cluster and all fault tolerant attributes described above have been verified. We await the results of more elaborate testing with great anticipation.

* Inspired by Eric S. Raymond in "The Art of Unix Programming" (Addison-Wesley, 2003).

©2006, Kenneth R. Koehler. All Rights Reserved. This document may be freely reproduced provided that this copyright notice is included.

Please send comments or suggestions to the author.

Visitors since 1/5/08: