Philip develops systems and network management software for Tivoli Systems in Austin, Texas. He can be reached at [email protected].
Powerful tools that let you construct distributed applications using models considerably more powerful than traditional client/server systems are beginning to mature. Java's ability to dynamically load, serialize, and reload objects makes it possible for you to construct systems that can change their own configurations and adapt to observed run-time conditions.
In this article, I'll examine dynamic distributed Java programming techniques by describing the implementation of a prototype load-balancing system called "Sojourner." This system (available electronically, see "Resource Center," page 3) monitors the load conditions on a set of workstations and migrates server objects to underutilized hosts. This improves the performance of the servers and reduces competition for resources on busier machines.
Sojourner uses the Voyager distributed computing infrastructure from Objectspace (http://www.objectspace.com/). Voyager, which is free for noncommercial and most commercial uses, provides facilities for remote object instantiation, invocation, and migration. Voyager is intended to support mobile agent systems and, therefore, supports dynamic relocation of running objects. The Sojourner load balancer demonstrates how these facilities can be useful for many dynamic distributed systems -- not just agent-based ones.
Basics of Distributed Java
The MonteCarloTest class in Listing One is a Java application that estimates the area under the curve of the square-root function on the interval from zero to 100. The class computes the area by averaging the value of the function at 1000 random points in the interval. The MonteCarlo class in Listing Two performs the actual calculations. MonteCarlo's estimateIntegral() method receives the interval boundaries, desired number of iterations, random-number generator, and function evaluation method from its caller. The class then uses these parameters to produce an estimate of a definite integral.
If estimateIntegral() is called with a complex function and a large number of iterations, its run time can be significant. In such a situation, it may be desirable to implement estimateIntegral() in a remote (server) object. This would permit client applications to continue doing useful work while computations are being performed.
Using Voyager, the only necessary modification to MonteCarlo is that it must implement java.io.Serializable in its declaration. This lets the Voyager run-time system dynamically relocate its instances. Once this modification is made, the Voyager class compiler utility (vcc) can be run against MonteCarlo.class. vcc's output is a "virtualized" version of the MonteCarlo class called VMonteCarlo.java.
Instances of VMonteCarlo serve as proxy objects for MonteCarlo objects running on remote hosts. A Voyager virtual class has all the same methods as the class it is derived from. The only difference is its constructors. The last parameter of every virtual class constructor is a String that indicates the location of the Voyager server that is to host the remote object and a handle that can be used to identify it. Virtual classes also inherit a set of methods from Voyager's VObject that are used to control (destroy, migrate, and so on) remote objects.
Listing Three demonstrates how a Voyager virtual class can be instantiated, accessed, and invoked. Example 1(a) creates an instance of the MonteCarlo class in the Voyager server running on the machine tme10.tivoli.com and listening on port 7777. The new object is given the handle MCServer. VMonteCarlo inherits the method liveForever() from VObject. Invoking it allows the remote MonteCarlo instance to persist even when no references to it exist. Example 1(b), excerpted from Listing Three, returns a reference that can be used to communicate with the remote MonteCarlo object created by the aforementioned new operation. Listing Three shows all the logic needed to create a remote MonteCarlo server or access it if it is already available. This code, plus a Voyager server, is enough to allow clients anywhere on the Internet to offload computations. Figure 1 illustrates the interactions between the MonteCarlo client, MonteCarlo server, VMonteCarlo proxy, Voyager server, and the Java environment.
Basics of Dynamic Load Balancing
Most distributed-computing systems simply allow clients on remote machines to locate and exploit resources (file systems, databases, processors, printers, and the like) on central servers. To improve scalability and lessen demands on client systems, many client/server architectures are enhanced using a multitier approach. In multitier systems, lightweight clients make requests of nearby gateway machines which, after performing validation or batching of requests from multiple clients, forward them to an actual server for processing.
Both client/server and multitier systems are static. Certain machines are configured as servers or gateways, and they perform this task until an administrator decides to change the configuration. The efficiency of such systems can be improved by adding the capability to automatically relocate components during operation based on the observed conditions of the host systems.
Voyager provides the basic migration facilities for this dynamic architecture. The mechanism for relocating objects is the moveTo() method, which all virtual classes (like VMonteCarlo in Listing Three) inherit from VObject. The moveTo() method can take a String indicating a new host Voyager server or a reference to a VObject running on the intended new host server. Invoking moveTo() causes the Voyager server hosting the remote server object to:
- Start queueing any remote method calls made to the object being moved.
- Allow threads executing in the server object being moved to complete.
- Serialize the server object and send it to the new host.
- Inform any VObject trying to contact the relocated object (including those whose calls were queued in step 1) of its new location.
After such a relocation, a VObject.forObjectAt() call using the old server location and handle still succeeds. The original Voyager server informs the proxy object of the server object's current location. Similarly, existing references to a migrated object will be updated transparently to the client when a method on the corresponding VObject is invoked.
Given the ability to dynamically migrate objects, you need to specify the criteria for determining when and where objects should be moved. The migration criteria for the Sojourner load-balancing system is simple. When a server object is located on a machine with high CPU utilization, it is moved to a host with a low CPU load, if such a host exists.
Defining and Measuring System Load
At some point, as CPU load increases, performance decreases to a point where the user's ability to get work done is noticeably hindered. This threshold is subjective, but since the purpose of a computer is to get work done, the load point where users find system performance unacceptable is important. The amount of CPU load where a machine becomes noticeably slow to its user is that computer's "nuisance threshold" (Figure 2).
Conversely, a machine might be so underutilized that a small increase in load would be undetectable to users. A computer at or below this load measure is below the "adequate utilization threshold."
Given these definitions, the objective of dynamic-load balancing can be stated as increasing system utilization and performance by moving processes from machines that are over their nuisance threshold to ones that are below their adequate utilization threshold.
Because Java does not provide a means for interrogating a host's CPU load, a native method implementation must be used. Listing Four shows a native method for measuring the system load on an IBM RS/6000 workstation running AIX (UNIX). The code uses the rstat() function, which returns an indication of the CPU's average run queue length for the preceding 60 seconds. This value is sampled every six seconds and compared to the machine's nuisance and adequate utilization thresholds.
To avoid inadvisable migrations, a smoothed load measure is computed. This measure is changed only if several consecutive samples indicate a new load state. Because it is better to overestimate a machine's load than to underestimate it, it takes fewer consecutive high observations to increase the load measure than consecutive low observations to decrease it.
The values for these thresholds were determined experimentally. Processes were repeatedly added to a machine and the user was asked to rate the system's performance. When the user's response was "poor," the rstat() value corresponding to the number of processes running was recorded as the machine's nuisance threshold. The adequate utilization threshold was set to half the nuisance threshold. This experiment was repeated for several users of each type of machine present in the distributed system.
On nonUNIX machines, a similar methodology can be used to map the load measure exposed by the operating system to the thresholds. Each class of machine to be supported by Sojourner needs a different implementation of the native method for interrogating the host's CPU load. Each of these native methods is encapsulated by the SojournerNode Java class (available electronically).
Sojourner Implementation
Sojourner is a prototype dynamic distributed system that demonstrates techniques that can be used to balance the load of Java server processors on small or medium-sized computer clusters. Distributed systems can be used not only to provide remote access to centralized or shared resources, they can also identify idle resources and make them available to constrained or poorly performing applications.
In most workstation clusters a large number of computers will be nearly idle at any given time. This may be because users are away from their computers (vacation, meetings, and so on) or the systems are being used for nonintensive applications (text editing, for instance). At the same time other users may be running resource intensive applications (statistical or numerical applications, simulations, or CAD packages). Performance of these applications could be improved if resources on nearby idle machines could be exploited. Sojourner demonstrates how, in a Java environment, this can be done.
In a cluster running Sojourner, one machine is designated as a "master" node, which serves as a clearing house for remotely enabled Voyager server objects. All other participating machines run a program that monitors the local load condition, registers with the master node and accepts guest objects when the local load is low, and migrates guest objects to other machines when the load becomes high.
Figure 3 shows a Sojourner cluster. The master node runs an object called SojournerMaster. Other nodes run SojournerNodeControl. In Figure 3, the three shaded nodes are busy with local processes and are not hosting guest objects. The two unshaded nodes are lightly loaded and are hosting two or three guest server objects (represented by the shapes attached to the SojournerNodeControl object). Two other server objects are running as guests of SojournerMaster until they can be migrated to other hosts.
The Sojourner components used to maintain a central list of lightly loaded nodes are available electronically. The SojournerCentralRegistry is simply a wrapper class around a Hashtable. The Voyager vcc utility is used to create its VSojournerCentralRegistry proxy object class. One instance of SojournerCentralRegistry resides on the master node and is accessed by nonmaster nodes using the SojournerLocalRegistry class.
The essential logic of Sojourner is contained in the SojournerNodeControl class from which the VSojournerNodeControl proxy class is derived. SojournerNodeControl contains an instance of SojournerLocalRegistry, which is used to communicate load information among participating workstations. It also contains a Vector used to save references to currently running local guest server objects.
The SojournerNodeControl constructor starts a Thread that is used to monitor local processor load conditions and take action when they change. When the observed load becomes low, the machine registers with the central registry as being available to host guest objects. When the load is higher, this registration is rescinded.
If the load on the local processor becomes high, the evictGuests() method is called. It runs through the Vector of guest objects and uses the Voyager moveTo() method to send them directly to available hosts. A round-robin selection method is used so that the guests are dispersed to as many other hosts as possible (rather than potentially overwhelming a single node). If no lightly loaded hosts are registered, then guests are migrated to the master node.
The receiveGuest() method, which is declared in the SojournerHost interface (available electronically), is called by other machines to offload guests. It adds the incoming objects to the Vector of guests. The receiveGuest() method may refuse a migration if the local load is not low. This can happen if the load on a machine changes between the time the Enumeration of available machines is obtained from the registry and an actual migration is attempted.
SojournerMaster, the basis of VSojournerMaster, is simpler than SojournerNodeControl. Its constructor establishes the SojournerCentralRegistry. The master node, being the host of last resort when all other machines are busy, cannot refuse to accept a guest object. Therefore, it does not need to monitor local load conditions or react to load changes. Its run() method simply checks the local guest list once per minute and tries to offload one local guest object to each registered node.
Any Serializable virtual object (such as VMonteCarlo, derived by applying Voyager's vcc utility to the MonteCarlo class) can be managed by the Sojourner system without further modification. Such server objects can simply be instantiated on the Sojourner master node and a reference passed to SojournerMaster's receiveGuest() method. Listing Five shows how the VMonteCarlo client of Listing Three can be modified to pass a newly created VMonteCarlo object to Sojourner. The client has also been changed so that the name of the master machine can be passed as a command line argument. The Voyager moveTo() method used by Sojourner to relocate objects is transparent to clients of those objects. That is, the forObjectAt() method call in Listing Five will return a reference to the correct object even if it has been moved off the master node.
Conclusion
The Sojourner system presented here is a working prototype. A production system would require capabilities to restrict the number of server objects depending on the number of nodes participating, to display the location of server objects, to selectively shut down server objects, and to authenticate client applications. The Voyager infrastructure provides mechanisms for implementing these functions, but they are beyond the scope of this discussion. The prototype does demonstrate how a simple construct like Voyager's moveTo() method can be used to build flexible dynamic distributed systems.
DDJ
Listing One
import java.lang.Math;import java.util.Random;
public class MonteCarloTest implements FofX {
public double computeFof(double x)
{
return Math.sqrt(x);
}
public static void main(String[] args)
{
MonteCarlo estimator = new MonteCarlo();
MonteCarloTest function = new MonteCarloTest();
double answer;
System.out.print("The area under the curve of the square ");
System.out.print("root of x from 0 to 100 is ");
answer = estimator.estimateIntegral(0.0, 100.0, 1000,
new Random(), function);
System.out.println(answer);
}
}
Listing Two
import java.util.*;public class MonteCarlo {
public double estimateIntegral(double intervalStart, double intervalEnd,
long iterations, Random random, FofX function)
{
double intermediate = 0.0;
double intervalLength = intervalEnd - intervalStart;
double x;
for (int i = 0; i < iterations; i++)
{
x = intervalStart + (random.nextDouble() * intervalLength);
intermediate += function.computeFof(x);
}
return (intermediate / iterations) * intervalLength;
}
}
Listing Three
import java.lang.Math;import java.util.Random;
import java.io.Serializable;
import COM.objectspace.voyager.*;
</p>
public class RemoteMonteCarloTest implements FofX, Serializable {
public double computeFof(double x)
{
return Math.sqrt(x);
}
public static void main(String[] args)
throws VoyagerException
{
VMonteCarlo estimator = null;
try
{
estimator = (VMonteCarlo)
VObject.forObjectAt("tme10.tivoli.com:7777/MCServer");
}
catch(ObjectNotFoundException e) { }
if (estimator == null)
{
estimator = new VMonteCarlo("tme10.tivoli.com:7777/MCServer");
estimator.liveForever();
}
RemoteMonteCarloTest function = new RemoteMonteCarloTest();
double answer;
System.out.print("The area under the curve of the square ");
System.out.print("root of x from 0 to 100 is ");
answer = estimator.estimateIntegral(0.0, 100.0, 1000,
new Random(), function);
System.out.println(answer);
Voyager.shutdown();
}
}
Listing Four
/* C implementation of a Java native method to determine if * the load on an IBM AIX workstation is LOW, MEDIUM or high
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <rpcsvc/rstat.h>
#include <sys/param.h>
#include "SojournerNode.h"
</p>
#define LOW 0
#define MEDIUM 1
#define HIGH 2
</p>
int determine_instant_threshold(char *);
</p>
int determine_smooth_threshold(int, int, int, int *);
/* instant_threshold, old_instant_threshold,
old_threshold, consecutive_change_cnt */
void call_rstat(char *, float *);
/* hostname, cpu_level */
</p>
</p>
/* Global data */
float cpu_limit = 200.0;
int current_threshold;
JNIEXPORT void JNICALL Java_SojournerNode_runNode
(JNIEnv *env, jobject obj)
{
char hostname[256];
int old_threshold, instant_threshold,
old_instant_threshold, threshold_count;
int loop_cnt;
jclass clazz;
jfieldID fid;
threshold_count = 0;
gethostname(hostname, sizeof(hostname));
clazz = (*env)->GetObjectClass(env, obj);
fid = (*env)->GetFieldID(env, clazz, "machineLoadLevel", "I");
current_threshold = (*env)->GetIntField(env, obj, fid);
while(1)
{
sleep(6);
old_threshold = current_threshold;
old_instant_threshold = instant_threshold;
instant_threshold = determine_instant_threshold(hostname);
current_threshold = determine_smooth_threshold(instant_threshold,
old_instant_threshold, old_threshold, &threshold_count);
if (current_threshold != old_threshold)
(*env)->SetIntField(env, obj, fid, current_threshold);
}
}
int determine_instant_threshold(char *hostname)
{
float new_cpu_level;
call_rstat(hostname, &new_cpu_level);
if ((new_cpu_level < (cpu_limit / 2.0))) return LOW;
if (new_cpu_level < cpu_limit) return MEDIUM;
return HIGH;
}
int determine_smooth_threshold(int instant_threshold,
int old_instant_threshold, int old_smooth_threshold,
int *threshold_change_count)
{
if (instant_threshold == old_smooth_threshold)
{
*threshold_change_count = 0;
</p>
return instant_threshold;
}
if (instant_threshold == old_instant_threshold)
(*threshold_change_count)++;
else {
if (old_smooth_threshold == MEDIUM &&
(((instant_threshold == LOW) &&
(old_instant_threshold == HIGH)) ||
((instant_threshold == HIGH) &&
(old_instant_threshold == LOW))))
{
*threshold_change_count = 1;
return MEDIUM;
}
else *threshold_change_count++;
}
</p>
if (instant_threshold > old_smooth_threshold)
{
if (*threshold_change_count > 1)
{
*threshold_change_count = 0;
return ++old_smooth_threshold;
}
else return old_smooth_threshold;
}
</p>
if (*threshold_change_count > 10)
{
*threshold_change_count = 0;
return --old_smooth_threshold;
}
return old_smooth_threshold;
}
void call_rstat(char *hostname, float * cpu_queue_length)
{
struct statstime rstat_output;
rstat(hostname, &rstat_output);
*cpu_queue_length = rstat_output.avenrun[0];
}
Listing Five
import java.lang.Math;import java.util.Random;
import java.io.Serializable;
import COM.objectspace.voyager.*;
</p>
public class RemoteMonteCarloTest1 implements FofX, Serializable {
public double computeFof(double x)
{
return Math.sqrt(x);
}
public static void main(String[] args)
throws VoyagerException
{
VMonteCarlo estimator = null;
try
{
estimator = (VMonteCarlo)
VObject.forObjectAt(args[0] + ":7777/MCServer1");
}
catch(ObjectNotFoundException e) { }
if (estimator == null)
{
VSojournerMaster master = (VSojournerMaster)
VObject.forObjectAt(args[0] + ":7777/sojournerNodeControl");
estimator = new VMonteCarlo(args[0] + ":7777/MCServer1");
estimator.liveForever();
master.receiveGuest(estimator);
}
RemoteMonteCarloTest function = new RemoteMonteCarloTest();
double answer;
System.out.print("The area under the curve of the square ");
System.out.print("root of x from 0 to 100 is ");
answer = estimator.estimateIntegral(0.0, 100.0, 1000,
new Random(), function);
System.out.println(answer);
Voyager.shutdown();
}
}
Copyright © 1998, Dr. Dobb's Journal
