Jojo

Abstract

Jojo is a distributed programming environment suitable for hierarchal Grid, implemented in Java. Jojo provides several features, including remote invocation using Globus GRAM, intuitive message passing API suitable for parallel execution and automatic user program staging. Using Jojo, users can construct parallel distributed application on the Grid.

Motivation

PC cluster system is a hopeful candidate as computation resource on the Grid. Cluster of small clusters will be the typical configuration of the Grid in the near future.

Each node of a cluster will not be able to get global IP address, because of adress exhaustion, and have to hook up to the Internet via NAT. Existing Grid middleware including MPICH-G2 assumes direct connection to each nodes and hence cannot work well in this kind of environment.

Design

Jojo is designed to fulfill the following requirements.

• System structure directly reflects the hierarchical Grid structure.
• Flexible programming model suitable for thread based code
• Bootstrap invocation which eases the burden of system and user program installation.

System Structure

Jojo targets the hiearchical Grid environment mentioned above. The whole configuration of nodes in Jojo will be a tree which takes arbitrary levels and hava the client node as a root. Each node can talk to its sibling nodes, parent node, and decendant nodes.

Programming Model

There are two welknown programming models for parallel communication libraries. One is parallel object model that is used by the Sun's RMI. Another is SPMD (single program multiple data) model that is used by the MPI.

Jojo's programming model is a hybrid of these two models. In jojo, each node executes just one representative object and it sends and receives messages to/from other nodes. Users cannot create new representative object at runtime.

In Jojo, each nodes sends and receives single object. While sending message is explicit, receiving message is implicit in Jojo, namely, user just defines receiving handler for the message. The message handler will be invoked in a seperate thread upon the arrival of the message.

Jojo provides several message passing mode including non-block, block, future-based, and context-based.

Invocation

The target environment of Jojo is widely distributed clusters. it means that we cannot assume the NFS file sharing . To ease the burden of installing user programs every-single node, Jojo automatically stages user programs to each node. Adding to it, Jojo itself also will be staged, preventing possible system version mismatch among nodes.

Implementation

Remote Invocation

To reduce the burden of installation, Jojo system stages not only user programs, but also Jojo system itself. For this purpose, we used boot strap loader rjava. The Jojo invocations steps are as follows;

1. Rjava bootstrap server which includes original class loader is transfered to the server as a jar file.
2. Invoke a bootstrap server and establish connection between the client and the bootstrap server. The client is responsible to read and transfer of class files needed by the bootstrap classl oader.
3. The bootstrap server on the remote node invokes Jojo system classes. These classes are automatically loader from the client local file system via the rjava client.
4. User program on the Jojo system also will be loader from the client local file system.

For the invocation, users can use GRAM/GASS, rsh/rcp, and ssh/scp.

Programming API

A program for Jojo is consist of several classes which extend abstract class Code. In Code class, users can uses support classes such as Node and Message.

Code Class

Abstract class Code is defined as follows. Siblings, descendants and parent points nodes in the same level, upper level and lower level, respectively.

The init method initializes the object taking a Map object which contains initialize information. The start method does the real job. The method will never be called before the init method is called.

abstract class Code{ Node [] siblings; /** Sibling nodes */ Node [] descendants; /** child nodes */ Node parent; /** parent node */ int rank; /** ordering number in the siblings */ /** initialize */ public void init(Map arg); /** do the job */ public void start(); /** message handler */ public Object handle(Message mes); }

The handle method is the handler which invoked when a message arrives to this node. The method is run in a newly created thread. Therefore, if several messages arrive continuously, several threads will execute the handle method simultaneously. It means you can do something which will take a long time in the handler without disturbing other message handlings. On the other hand, you have to perform proper mutual exclusive control to access shared resources.

Node Class

Code classe objects communicate each other invoking methods of Node class. Node class provide 4 kinds of method to enable flexible asynchronous communication.

• void send(Message msg)
  • Simply sends a message object, then return immediately.
• Object call(Message msg)
  • Sends a message object and wait for a reply object, then returns the reply.
• Future callFuture(Message msg)
  • Sends a message object and immediately returns with Future object. Future object is a kind of wrapper for the reply object. When the touch() method is invoked, it will return the reply object if it is ready. If not, it will block till the reply object arrives.
• void callWithContext(Message msg, Context context)
  • Takes a object which implemnts Context interface as the second argument. This method sends a message object and returns immediately. When the reply object arrives, it will invoke a thread and calls run method of the object with the reply object. Context interface is defined as follows.
    • public interface Context{ public void run(Object o); }

Message Class

Message class represents objects which are sent and received by the Code objects. This class have int tag field which represents message ID and Serializable contents field which represents message body itself.

class Message implements Serializable{ /** Message ID */ public int tag; /** Contents of the message */ public Serializable contents; }

Issues

• Through preliminary experiments, we found that the communication throughputs between nodes are very low. We suspect that it is due to a communication stream multiplexer which sits in between nodes and splits single stream into multiple streams. We have to investigate details and improve the throughput.
• We will prove scalability of Jojo through implementation of several combinatorial optimization problems using jPoP, which is implemented on top of Jojo.
• In current implemntation, sibling nodes do not communicate each other directly; instead, they communicate through the parent node. This is becuase, direct communication among siblings is prohibited by authentication issue, in some environment. However, in certain situation, it is possible to perform direct communication. We will implement it to improve total performance.