Using a Global Event Model in Distributed Control Systems

Adam Rifkin
Caltech Department of Computer Science 256-80, Pasadena, CA 91125
adam at xent dot com

December 16, 1997

This paper is also available in PostScript.

See also the published version of this paper, written with Eve Schooler and Mani Chandy in February 1998.


We specify an abstract model for distributed control systems in which the component objects do not have access to the state of the entire system, new objects can enter and leave the system dynamically, and each object makes its own local decisions (though collectively these local decisions maintain some system-wide constraints). We present several applications of such control systems, and then focus on distributed resource management. In exploring solutions that are compositional (and hence scale), we observe that a general Java framework for constructing distributed resource management algorithms benefits both from exploiting the multicast facilities of Java and from building global events into the Java infrastructure.

1. Introduction

  This paper discusses distributed systems for managing resources in a computational grid within the larger context of distributed control systems. Our overall goal is to develop compositional ways of obtaining high confidence in dynamically-reconfigurable scalable distributed systems.

We begin with the overall research problem of investigating distributed control by proposing an abstract model. Later, we refine this model for the specific case of distributed resource management systems, discuss different kinds of resource management problems, and discuss a strategy for implementing a distributed resource management system using the Java middleware layer called Infospheres [7, 10]. The focus of this paper is on a specific announce-listen strategy based on multicast and GEM -- a Global Event Model that extends the local events of Java Beans to be distributable -- with an implementation using the Infospheres 2.0 packages planned.

1.1 Distributed Control Systems

  A distributed control system consists of interacting component objects, each of which have persistent local state and one or more threads of control [6]. As illustrated in figure 1, input messages called feedback may come from other objects; other input messages may come in an uncontrolled fashion from the system's environment. Output messages called observations can be routed as feedback to other objects. Any communication can experience arbitrary (but finite) delay.

Abstractly, a simple specification of a distributed control system consists of the descriptions of several components: state, computation, uncontrollable variables, and constraints (including an objective function).

Figure 1: A distributed control system consists of component objects (each of which may have local control) and an uncontrollable environment, interacting over a communication substrate that may permit arbitrary but finite delay. 


The state of a distributed system is defined in terms of its components. In our context, the components are the system's participating objects, as well as the communication infrastructure itself. The objects can be persistent, can communicate with other objects, and can have one or more local threads of control. Each object has access to its own local state -- the values of its variables, its pending inputs and outputs, and its threads of control. However, no object can directly access the local states of other objects, nor can it directly access the state of the distributed system itself.


A computation of a system is a mapping from time to the states of the system; the computation specifies the state of a system at a given time. In control terms, we consider the computation to be the trajectory of the system's state.

Uncontrollable Variables.

The system gets inputs from its environments, which can then be directed to one or more of the system's component objects. These inputs cannot be directly controlled by the system. The system's state responds to these inputs according to some specified system behavior.

Constraints and Objective Function.

The system's computation is determined by uncontrollable inputs from the environment, and by the system's own state and control policies which determine its state transitions and the communications between objects. As the system's designers, we are given constraints and an objective function defining the desired computations.

An example of a constraint is that a high-priority request is always serviced before a low-priority request for the same set of resources, if both requests are made at the same time and place. An example of an objective function is to minimize the average response time for a request.

The constraints and objective function may be specified informally or formally -- the degree of rigor is not the primary concern of this paper. Indeed, in many cases, the constraints and objective may be implicit rather than explicit. As illustrated in figure 2, our problem as system developers is to design the system's controls so as to maximize (or minimize) the objective function, subject to the constraints.

Figure 2: As developers, we seek to design the control with local state, other objects' observations, and external environment signal as inputs, and the next local state and new observations as outputs, so as to maintain the objective function, subject to the constraints. 

1.2 Characteristics of Distributed Control Systems

  Several aspects of distributed control systems make the problem of designing the control challenging:

  1. Because of their distributed nature, component objects do not have direct access to the state of the entire system, so they do not know the states of other objects, conditions from the environment, or the status of communications in transit.
  2. New objects can enter and leave the system dynamically, so an object may not know which other objects are in the system at any given point in a computation.
  3. The local control of each object makes its own local decisions. By the constraints and objective function, however, we require the collection of decisions made by the local controls to result in an optimal (or at least good) system trajectory.

1.3 Outline of this Paper

  In section 2, we discuss several applications suitable for the distributed control system model in section 1.1. We focus on the problem of developing one of these applications -- distributed resource management -- and outline the categories of solutions in section 3.

Section 4 presents an abstract model for distributed resource management using tokens or slidebars, and section 5 describes several token-based algorithms for the different solution categories of section 3. We conclude with a status of our Java implementation, with details about related and future work in section 6.

2. Applications

  The distributed control model fits a wide range of applications from flight control of a flock of uninhabited autonomous vehicles to determining the scope of a multicast announcement in a session-invitation application. The common problem is that each object has to make an autonomous decision based on old (and possibly partial) information about other parts of the system. In this section, we discuss a few applications; for the remainder of this paper, we restrict attention to resource management systems.

2.1 Bandwidth Allocation in Multicast Collaborative Applications

  A multicast collaborative application such as a conferencing tool may use multiple multicast ports. A collaboration with more than one sender and several receivers has to partition bandwidth during a session within each multicast channel among the contending senders; this can be done adaptively, as with the Scalable ConsensUs-based Bandwidth Allocation (SCUBA) [3]. As a distributed control system, each sender has partial and old information about receivers and other senders, and each sender has to determine dynamically what fraction of the bandwidth to use for each channel.

2.2 Infrastructure for 21st Century Organizations

  Trends today suggest that just-in-time organizations will emerge to satisfy specific needs as they arise. Today we witness several ad hoc consortia forming to bid on everything from aircraft design and production for the Department of Defense, to basic research for the National Science Foundation. Support for such consortia requires an infrastructure in which providers and consumers can find each other quickly and reliably, and in which physical goods are transported when needed. This suggests that Fedex, UPS, and the postal service will need to become information technology companies as much as companies that deliver goods. Here too, the problem is that a scalable system constrains each decision-making object to make decisions with only partial information.

2.3 Management of Resources in a Computational Grid

  Providers of computational resources may attach and detach themselves to a computational network dynamically; in addition, communication links may fail. Requestors ask to reserve certain collections of resources, satisfying certain constraints, for specified periods of time. For instance, a requestor may ask for 10 or more nodes of a certain class of machine, a visualization workbench, and a high-quality printer, all connected by channels with at least 100Mb of available bandwidth, for an hour.

In a scalable system, each producer will not have complete and up-to-date information about every requestor or other producers. Yet, the grid has to collectively deliver resources to requestors in a manner that satisfies given constraints and optimizes the given objective, as is done with the Globus [12] and Legion [15] resource management systems.

Applications outside Metacomputing.

Note that such a system has applications outside the domain of metacomputing. For example, consider the general distributed resource management problem with finite resources, providers of resources, and requestors of resources. Traditionally, a requestor queries different providers for different types of resources; for example, if you want to go on a ski trip, you have to call different airlines, car companies, and hotels, to piece together the best deal yourself. With the proper compositional structures in a distributed vacation-control system, however, you could merely announce your requirements (for example, ski trip, price, kind of slopes, and hotels) and have the pool of providers respond by giving you proposals. For example, one interesting approach involves online just-in-time middlemen: an intermediate object can search for objects that provide cars, or hotels, or planes, and create a collaboration between them, to put together a complete package for you.

The remainder of this paper focuses on distributed control system solution strategies to two categories of the distributed resource management problem: systems with collaborative resource providers and systems with competitive resource providers.

Categories of Resource Providers: Collaborative and Competitive.

In a computational grid, providers of resources may collaborate in providing services or goods to requestors acting on behalf of resource consumers. Note that the requestors will be the consumers themselves in some cases, and they will be middlemen in others. Resource providers may also compete; to keep sensitive information hidden, they may form temporary external collaborative consortia to respond to a specific request while still competing in the large.

Which solution is most appropriate for any given distributed resource management depends partly on whether providers and requestors compete or collaborate. In a collaborative environment, providers and requestors can use an announce-listen communication strategy: each object knows that all the other objects within the scope of the announcement will get the announcement [21]. The overall system state can be propagated, with delays, throughout the system, providing that scalability is not a problem. In a competitive situation, an object cannot assume that all other objects within the scope of an announcement will hear an announcement, and this leads to different kinds of algorithms, as we discuss in section 5.

Next, we discuss categories of solution strategies. These strategies are motivated by work on the Infospheres 2.0 system, which is based on Java, Web, and XML technologies, but the essential ideas of the strategy are also applicable to other middleware layers such as CORBA and DCOM.

3. Categories of Solutions to Distributed Resource Management

  Solutions to distributed resource management can be constructed along several axes:

In this paper, we investigate distributed resource management algorithms using global events and soft state with some hard information, for consumer pull, push, and hybrid algorithms. But first, we explore these categories a little further.

3.1 Soft State and Hard State

  A soft state is an approximate representation of the current system state. An object estimates the current state of a distributed system based on information sent earlier by other objects; thus the estimated soft state is based on old information. For instance, if an object B is listening on a multicast port at time t, then we may know with high probability that B will continue to listen for some more time. If at time (t + t'), object A gets a message from B, timestamped t, stating that B is listening, then A has the ``soft'' information that B is listening at time (t + t'); and the softness of the state can be quantified in terms of probability [11].

An example application using soft state is SCUBA [3], for controlling bandwidth allocation in synchronous collaborations. The amount of bandwidth taken by a sender depends on the relative interest expressed by multiple listeners. The information that a sender gets about listener interest is old information because of message delays, but since listener interest fluctuates relatively slowly compared to message delays, the old information can be used with satisfactory results.

By contrast, a ``hard'' property is a property that holds in the current system state. A hard property follows from an invariant of the system. For instance, suppose an invariant of the system is: If variable x in object A has value 1, then variable y in object B has value greater than or equal to 1. In this case, if x in object A has value 1, then object A knows that y in object B has value at least 1, and this knowledge is hard [8].

Some distributed control problems allow for the use of soft states exclusively, and algorithms based on soft state usually scale better and are more resilient in the presence of faults [21]. Whether soft states can be used depends on the nature of the problem, the duration of message delays, and the fuzziness of constraints and objective. By contrast, if in an algorithm an application is given high priority for one resource and low priority in another resource, deadlock can arise; for such applications, invariants or hard states must be used.

3.2 Communication Mechanisms

  Our Infospheres 2.0 packages described in section 6 implement three basic communication strategies derived from Java: remote method calls, messages, and global events. These mechanisms can be used for object interactions in distributed control systems.

Remote Method Calls.

Remote method calls allow an object in one virtual machine to call methods on an object in another virtual machine. Introspection can be used to implement remote calls instead of Java's Remote Method Interface (RMI) when dynamic reconfigurability is a concern [1]. Infospheres 2.0 provides three flavors of calls: regular calls, future calls in which the caller is not blocked after making the call and during which the caller can check whether a value has been returned at any point; and, oneway calls.


The message-passing layer of Infospheres 2.0 allows hosts (virtual machines) to set up inboxes and outboxes by which messages are received and sent respectively. The package allows a message to create the appropriate receiver protocol stack on-the-fly (for example, piecing together protocols such as compression or encryption as needed, like the Horus [22] system) when the message is received. Communication protocols supported include Java's TCP, UDP, and multicast facilities; developers can employ any of these depending on their quality-of-service requirements. Remote method calls (described earlier) are implemented on top of the message layer, and global events (described next) are implemented using remote methods.

Global Events.

Java Beans provide local events as a mechanism by which a component can inform other components that something interesting has happened. These events can be thought of as active messages; for example, a button is pressed at a source, and channeled through an event listener, to trigger a method in an event observer automatically. An event is immutable and does not usually change as it propogates from an event source through an event notifier (also called an event listener) to one or more event observers (also called event targets) who respond to the events as they arrive. The notifier routes the event to the observers using a control list, and observers can ask the notifier to be added or removed from this list without notifying the event source.

The global event structuring mechanism in Infospheres 2.0 is identical to the local event model of Java Beans, except that in the place of Java Beans' reference to an object within a single Java Virtual Machine, we use a global name for the object (employing the Web's URL convention). Furthermore, because the components of the event system are distributed, efficient event-broadcasting group communication protocols such as multicast can be used in the place of Java Beans' local event point-to-pointcasting.

Using global events, an event can be broadcast (announced) by a source object in one virtual machine, and notifiers for that event in other virtual machines anywhere on the Internet can receive ( listen to) the event and forward it to the appropriate observers (who themselves can be distributed in other virtual machines as well). Furthermore, unlike the group communication in virtual synchrony [4], it is not necessary for the event sources to know at any point who the event observers will be; they simply generate the events and let the event listeners handle the distribution of those events to the appropriate targets.

An Internet-scale event generation, observation, and notification facility provides a common (event-based) architectural style for distributed, loosely-coupled, heterogeneous software system development [20]. The Global Event Model (GEM) on which Infosphere 2.0's global events are based expands the rich Java local event model with mechanisms for distributing the components of an event system, for composing (distributed) event notifiers, and for filtering (using predicates) and providing security (using access control lists) at the event notifier level [19]. Although space considerations prevent us from describing GEM in detail here, we note that GEM's decentralized nature lends itself to several scalable announce-listen compositional structures for distributed object interaction, described next.

3.3 Compositional Structures

  As mentioned in section 2.3, the compositional structures for communication in distributed resource management applications differ in who announces information and who listens for it. Here, we describe three such structures: consumer pull, consumer push, and a hybrid structure with middlemen.

Consumer Pull.

In this case, the consumer asks providers in some order whether they have some or all of the desired resources. When you travel to San Jose from Washington, you may have asked a car rental agency (a provider of cars) for a car during a specified period, an airline for a plane ticket, and a hotel for a room. As the consumer, you pulled information out of different providers and changed their state (and your state as well). Some providers (such as a travel agency) may provide multiple resources such as an airline ticket and a car rental.

Similarly, a consumer program in a computational grid can poll different resource providers for processors, printers, and data storage, and then select some combination that satisfies its constraints.

Consumer Push.

In this case, the consumer merely announces a call for service. For instance, your travel-management object announces that you require certain resources for travel from Washington to San Jose, and the providers of the resources listen to such announcements and then respond. There are at least two kinds of responses: (i) a provider responds with a proposal for serving a part of the request -- say Hertz offers a car; or (ii) the response offers the total basket of services -- a travel agency offers a car-plane-hotel package.

The announcement requesting a service can be made using the Global Event Model, and it can use multicast as the message-passing protocol for efficiency. If providers are collaborative rather than competitive, multicast can be exploited to obtain efficiency. If one provider announces that it can provide the requestor with a resource, then other providers hearing that announcement first, do not in turn flood the network and the requestor with announcements that they too can provide the same resource. If providers are competitive, the requestor uses global events to announce a request but providers use point-to-point remote method calls to respond to the provider to hide interactions from other providers.

Note that for scalability, scoping can be used in announcing requests. Initially, if you want computational resources, you announce with small scope, and gradually increase the scope of the announcement until requests are fulfilled. For example, if you are in Caltech and providers in Pasadena can provide the resources then no further requests are necessary; otherwise, the request is announced with larger scope (all of Los Angeles county), and so on [21].

Hybrid Solutions with Online Middlemen.

In hybrid systems, the requestor asks a middleman to get the resources. The middleman may use global events or remote methods. The middleman may have a directory of resource providers, and thus the middleman determines the communication mechanism to use. Middlemen arranged in a grid announce requests and ability to provide resources, and other middlemen listen to these. If middlemen are cooperative, as they will be in many computational grid situations, efficiencies can be used in multicast to prevent unnecessary messages from being sent or unnecessary computation from being carried out.

In effect, middlemen provide services such as collating and negotiating between providers and consumers, for a particular region; thus, middlemen suggest a hierarchical system based on scopes.

4. Abstract Models for Distributed Resource Management

  Next, we specify distributed resource management systems abstractly. There are two kinds of abstract representations (corresponding to two types of problem) that we can employ: indivisible tokens to represent individual resources or slidebars to represent ranges of resource collections. In this section, we describe these abstractions; in section 5, we present algorithms for event-based, announce-listen multicast communication in both collaborative and competitive distributed resource management systems, using tokens to represent resources.

4.1 Resources as Indivisible Colored Tokens

  For the case when resources can be reserved by consumers to perform tasks, it is often helpful to represent each resource -- for example, a processor, a printer, a 1 Mb bandwidth chunk, a rental car, an airplane seat, or a hotel room -- by an indivisible colored token. Tokens intrinsically permit mutual exclusion, because only the current holder of a token may use its respective resource.

We can associate a token color for each respective ``type'' of resource, and the number of tokens of each color corresponds to the number of resources of that type available [18]. For instance, yellow tokens are used for supercomputers, red for visualization tools, blue for printers, and so on. Then, for a particular application, a scientist can specify that the task requires 3 red tokens, 2 blue tokens, and 1 yellow token, with possibly some preference hints (to the times preferred to reserve the resources, to specific machines preferred, and so on).

With the resources-as-tokens model, a consumer asks for a basket of tokens with constraints on the allocation of those tokens, and each such request is considered a reservation for a particular duration: once a consumer makes its request, that request does not change. The consumer eventually returns whatever tokens it has received to their respective providers, either on demand (which may be necessary for rollback to avoid deadlock) or when the consumer's task is complete.

Modeling resources as tokens, system algorithms can schedule exclusive access to an appropriate set of tokens at the appropriate time, allowing the desired computation to run uninterrupted. We note that such token allocation is a much simpler when all tokens are controlled by a single resource manager, because that manager knows all of the request bundles for token collections, and can use such knowledge to schedule efficiently. Composing multiple decentralized resource managers, each controlling separate stashes of tokens and each having incomplete knowledge of all pending requests, is challenging.

4.2 Resources as Adjustable Slidebars

  A different model is useful for systems that are more dynamic by nature. Instead of making immutable resource reservation requests, a consumer has the option of updating its requests on-the-fly, according to its most recent resource requirements. Such fluctuating requests need not be satisfied completely before new requests are made; rather, requests are fulfilled adaptively as resources become available, as occurs with SCUBA [3].

In such situations, each resource -- for example, a divisible bandwidth pipe or a fleet of cars -- is represented as an adjustable slidebar representing the utilization of that resource. Consumers request fractions of the resource capacity under allocation constraints (such as duration of intended use). The distributed control system can dynamically adjust the values in the slidebar based on consumer requests. As with the token model, each consumer eventually returns whatever fractions of the slidebar it has received to the slidebars' respective providers, either on demand (which may be necessary for rollback to avoid deadlock) or when the consumer's task is complete.

5. Algorithms for Distributed Resource Management

  Though the use of adjustable slidebars in resource request and provision is a fascinating area worthy of consideration, in this section we concentrate on algorithms for distributed resource management using the tokens model, for collaborative and competitive resource providers, using consumer push, pull, and hybrid composition.

5.1 Collaborative Providers, Consumer Push

  Consumers use global events to announce their requests for resources (including estimates of how long they will need those resources); the resource providers are global event listeners implemented using GEM. As illustrated in figure 3, the use of multicast to post events requires limiting the scope of the multicast announcements, so that only providers within the scope listen to consumer requests.

Figure 3: Only providers within the scope of the consumer can hear the consumer's announced events. 

When a listening provider can give a partial or total basket of tokens fulfilling a given request, it uses multicast to announce to the other providers what tokens it can commit to that consumer. Since providers collaborate to satisfy each consumer's request, the listening providers can ante up additional resources (or keep from doing so) based on previous information announced by providers and consumers.

Although multicast is not virtual synchrony [4] -- and hence group communications are not guaranteed to be reliable, as shown in figure 4 -- messages being in different orders at different objects is not a problem. Some listening providers will hear each consumer's request, and each consumer will hear some collection of tokens available and respond according to its needs using a two-phase commit of reserving and then locking [14, 18].

Figure 4: With multicast, group communications are not guaranteed to be reliable. Messages can be dropped or in different orders at different objects. 

This algorithm could present a problem if middlemen are used, because deadlock can occur if no consumer gets all of the resources it needs. To solve this problem, we stipulate that consumers must give up a provider's tokens, if asked by that provider to do so, even if it has not yet received all the tokens it requested.

This scheme works using global priorities: the priority of a request event is based on the local timestamp of that request, given by the requestor's local, logical clock, with ties broken by the alphabetic comparison of requestors' globally unique names. Global priorities represent the use of an invariant in an algorithm. We can assert that if a request R has a higher priority than another request R' at one global object, then R has a higher priority than R' at all global objects. Equivalently, all objects have ``common knowledge'' about the global priorities. The judicious combination of soft state with hard information allows for an algorithm that scales but is free from deadlock.

Using global priorities, providers demand tokens back from its lowest priority requestor, who must then return them. The multicast bus and soft state can be employed in these event transmissions, to reduce the ``oscillation'' of requests and returns of tokens where possible. All providers do not concurrently demand tokens back from each of their lowest priority requestors, because this can lead to oscillations -- ``here is a token'', ``give it back to me'', ``here is a token'', and so on. Each provider listens to the other providers and requestors, and each provider can make a good estimate about what other providers will do. So, each provider knows that the lowest priority requestor will return its tokens, and it can use this fact to determine whether to ask other requestors for tokens. Here, system components use soft state as an approximation for the true current state of the system, which works fine provided the system state does not change rapidly compared with multicast delays.

Introducing new providers and consumers into this system is easy -- they only have to determine the proper multicast ports on which to listen for events, and set their scopes appropriately. Leaving the system is similarly facilitated by multicast.

Scoping can be used to make the algorithm more efficient. When a consumer requests resources, listening providers can announce back their rough estimates of when their resources will be available (based on the usage estimates of the consumers currently using them). As illustrated in figure 5, a consumer that does not want to wait based on this information can then either increase its scope and re-announce its request, or it could send its request to proxying agents outside the scope who can then announce the request in their scopes and act on behalf of their respective consumers.

Figure 5: To change its area of announcement, a consumer can either increase its scope (for example, Pasadena to Los Angeles as in the left picture) or use remote procedure calls or messages to invoke an agent with the same scope in another location (for example, Pasadena to Santa Barbara as in the right picture). 

The focus of our first set of experiments has been on algorithms for collaborative providers with consumer push. For completeness, we also briefly discuss other scalable algorithms in section 5.2 through section 5.6.

5.2 Collaborative Providers, Using Middlemen

  Each consumer uses global events to announce its requests for resources to a list of middlemen specific to that consumer; each list is updated by its consumer dynamically both by asking other consumers about appropriate middlemen and by middlemen advertising their services in announcements to consumers.

Middlemen are global event listeners implemented using GEM, who then negotiate and collate token packages from the resource providers according to consumer requests. Middlemen maintain update their lists of providers the same way consumers update their lists of middlemen: both by asking other middlemen and by provider advertising. Since middlemen collaborate to fulfill consumer requests, middlemen keep each other informed using their announce lists.

The use of multicast to post events requires limiting the scope of the multicast announcements accordingly. For efficiency, middlemen can refrain from announcing events except when resources available change dramatically; if available resources change a lot, the middlemen can announce these changes periodically [21].

5.3 Collaborative Providers, Consumer Pull

  Providers use global events to announce their available resources; consumers are global event listeners implemented using GEM. The use of multicast to post events requires limiting the scope, so that only consumers within the proper scope listen to provider resource announcements. Providers make announcements when the tokens they hold change significantly, and on a periodic basis [21]. A consumer may choose to execute a low-priority task just because the resources it needs happen to be available [18].

5.4 Competitive Providers, Consumer Push

  As in section 5.1, consumers use global events to announce their requests for resources, and the resource providers are global event listeners implemented using GEM. What changes is that providers do not announce anything that competing providers could hear; rather, they use remote method calls or messages to talk with consumers directly.

Again, consumers using multicast is not a problem, because some listening providers will hear each consumer's request, and each consumer will hear some collection of tokens available and respond according to its needs using a two-phase commit of reserving and then locking [14, 18].

5.5 Competitive Providers, Using Middlemen

  As in section 5.2, each consumer uses global events to announce its requests for resources to a list of middlemen specific to that consumer; each list is updated by its consumer dynamically both by asking other consumers about appropriate middlemen and by middlemen advertising their services in announcements to consumers.

Middlemen are global event listeners implemented using GEM, who then negotiate and collate token packages from the resource providers according to consumer requests. If the middlemen themselves are not competing, the middlemen maintain update their lists of providers the same way consumers update their lists of middlemen: both by asking other middlemen and by provider advertising. Since middlemen collaborate to fulfill consumer requests, middlemen keep each other informed using their announce lists. If the middlemen themselves are competing, they do not announce anything that competing providers could hear; rather, they use remote method calls or messages to talk with consumers directly.

In either case, as in section 5.4, providers do not announce anything that competing providers could hear; rather, they use remote method calls or messages to talk with middlemen directly. The use of multicast to post events requires limiting the scope of the multicast announcements accordingly. For efficiency, middlemen can refrain from announcing events except when resources available change dramatically; if available resources change a lot, the middlemen can announce these changes periodically [21].

5.6 Competitive Providers, Consumer Pull

  As in section 5.3, providers use global events to announce their available resources; consumers are global event listeners implemented using GEM. The use of multicast to post events requires limiting the scope, so that only consumers within the proper scope listen to provider resource announcements. Since providers are competing, providers avoid announcing (as much as possible) things that competing providers can hear; rather, they use remote method calls or messages to talk with consumers directly, allowing the consumers themselves to piece together a distributed package of tokens fulfilling its needs [18].

6. Related and Future Work

  The algorithms described in section 5 have several advantages in using global events and multicast. The announce-listen paradigm is fault-resilient [21]; that is, if a resource provider goes away, the system adapts dynamically to continue to meet the requests of the consumers. Furthermore, systems constructed using global events and multicast are compositional and scale; providers and consumers can add or remove themselves at any point dynamically. Unfortunately, such systems also have the disadvantage of delay oscillation; that is, if state changes faster than the communication substrate delays, then soft state gives a bad estimation of the current system state. Nonetheless, the algorithms are compelling, and they can be implemented in Java using the Infospheres Infrastructure.

Status of Java Implementation.

The object persistence and communication Java packages of Infospheres 1.0 have been available since August 1997 [6], and we are presently working on wrapping these services in Java Beans. We have also implemented a prototype package for dynamic remote method calling using Infospheres 1.0 [1]. We have implemented a framework for constructing distributed resource managers using Infospheres 1.0, and used the framework to develop the collaborating consumer pull and competing consumer pull algorithms [18]. We plan to collect simulation and performance results of the Infospheres 2.0 (using GEM) implementation of the algorithm for collaborative providers, consumer push given in section 5.1.

The Caltech Infospheres group is currently improving the messaging mechanisms of Infospheres 1.0 (including support for Java's multicast facilities and support for on-the-fly protocol stack creation), improving the efficiency of the dynamic remote method calling prototype, and adding support for global events. These packages will collectively comprise Infospheres 2.0, which is set for an initial release in March 1998.

Related Work.

Our efforts in building the Infospheres 2.0 framework share with other ongoing metacomputing endeavors the employment of the Internet as a resource for concurrent computations.

SuperWeb has explored the technological and economic tradeoffs of using a Java-based infrastructure to harness global resources, making them available to anyone on the Internet through intermediate brokers that match client computations with suitable hosts [2]. SuperWeb's broker architecture is well-suited to the middlemen compositional structure in section 3.3 for the distributed management of computing resources, and SuperWeb's prototype implementation, Javelin [5], demonstrates that a Java infrastructure can obtain reasonable speedups in parallel computations as well.

High-performance facilities are critical to several other metacomputing frameworks, including Globus, NPAC, and Legion. Globus provides the infrastructure to create networked virtual supercomputers for running applications on machines distributed globally [12]. NPAC seeks to perform High Performance Computing and Communications (HPCC) activities using a Web-enabled concurrent virtual machine [13]. And, Legion provides an architecture and object model for giving the illusion of a single virtual machine to users for wide-area parallel processing [15]. Unlike these efforts, Infospheres 2.0 uses only pure Java standards; that is, the packages are just a collection of Java classes, without native method calls or source-to-source compilers. Also, the emphasis of our work is the exploration of event models and multicast with Java.

As a consequence, although the Infospheres 2.0 framework could be used for metacomputing applications, we provide neither seamless parallelism, nor facilities for developing high-performance appplications. Rather, we provide general mechanisms for programmers to use in developing distributed system components and in composing them dynamically, both to work synchronously on a task in short-lived sessions [9], and to coordinate work asynchronously on an archivable activity in longer, discontinuous collaborations [7].

Furthermore, the messaging layer in Infospheres 2.0 allows for arbitrarily nestable protocol stacks built on-the-fly, such as those provided by Horus [22]. However, we have not yet incorporated support in our packages for the virtually synchronous execution model for group membership and communication introduced by the Isis Toolkit [4]. With Isis' group membership primitives, participants in a distributed activity know the identities of the other group members at all times. For the dynamic distributed control systems described in this paper, allowing the component objects to have imperfect knowledge of other participants garners the scaling benefits of soft state [21]. A probabilistic model for estimating current state based on old information announcements and elapsed time may be sufficient for a distributed control system's information needs, yet efficient for such a system to scale well [11]. Nonetheless, we are investigating the construction of a virtual synchrony layer using Infospheres 2.0, because the fault-tolerance tools afforded by virtual synchrony -- such as for load-balanced request execution, for security, and for coherently replicated data -- have proven useful for many applications, as demonstrated by the CORBA-compliant interface to Horus called Electra [16] and by the Java middleware package iBus [17].


Infospheres 2.0 provides a suitable infrastructure for investigating algorithms for distributed control. By building global events into Java, and by exploiting the multicast facilities of Java, Infospheres 2.0 allows developers to investigate the tradeoffs between soft state and hard invariants, between pushing and pulling client requests, between providing a hierarchy of middlemen and having a simple flat requestor-provider structure, and so on. The overall goal of any distributed resource management system -- from programs allocating nodes in a computational grid, to concierges using just-in-time middlemen in piecing together travel arrangements -- is the appropriate matching of resource providers and requestors, and the Infospheres 2.0 framework allows system developers to explore the costs and benefits of the different potential solution strategies.


This work was supported under the Caltech Infospheres Project is sponsored by the Air Force Office of Scientific Research under grant AFOSR F49620-94-1-0244, by the CISE directorate of the National Science Foundation under Problem Solving Environments grant CCR-9527130, by the NSF Center for Research on Parallel Computation under Cooperative Agreement Number CCR-9120008, and by Parasoft and Novell. More information about Infospheres is available at on the Web.


J. Aldrich, J. Dooley, S. Mandelsohn, and A. Rifkin, `Providing Easier Access to Remote Objects in Client-Server Systems', Proceedings of the Thirty-first Hawaii International Conference on System Sciences, Hawaii, January 1998.

A. D. Alexandrov, M. Ibel, K. E. Schauser, and C. J. Scheiman, `SuperWeb: Research Issues in Java-Based Global Computing', Concurrency: Practice and Experience, June 1997.

E. Amir, S. McCanne, and R. Katz, `Receiver-driven Bandwidth Adaptation for Light-weight Sessions', Proceedings of ACM Multimedia, Seattle, Washington, November 1997.

K. P. Birman and R. van Renesse, Reliable Distributed Computing with the Isis Toolkit, IEEE Computer Society Press, Los Alamitos, California, 1994.

B. O. Christiansen, P. Cappello, M. F. Ionescu, M. O. Neary, K. E. Schauser, and D. Wu, `Javelin: Internet-Based Parallel Computing Using Java', 1997 ACM Workshop on Java for Science and Engineering Computation, June 1997.

K. M. Chandy, `Software Infrastructure for VECS', AFOSR/Caltech Workshop on Theoretical Foundations of Virtual Engineering and Complex Systems, Pasadena, California, December 1997.

K. M. Chandy, J. Kiniry, A. Rifkin, and D. Zimmerman, `Webs of Archived Distributed Computations for Asynchronous Collaboration', Journal of Supercomputing, Volume 11, Number 2, Pages 101-118, August 1997.

K. M. Chandy and J. Misra, `How Processes Learn', Journal of Distributed Computing, Volume 1, Number 1, Pages 40-52, 1986.

K. M. Chandy and A. Rifkin, `Systematic Composition of Objects in Distributed Internet Applications: Processes and Sessions', Computer Journal, Oxford University Press, October 1997.

K. M. Chandy, A. Rifkin, P. A. G. Sivilotti, J. Mandelson, M. Richardson, W. Tanaka, and L. Weisman, `A World-Wide Distributed Sytem Using Java and the Internet', Proceedings of the Fifth IEEE International Symposium on High Performance Distributed Computing, Pages 11-18, Syracuse, New York, August 1996.

K. M. Chandy and E. M. Schooler, `Designing Directories in Distributed Systems: A Systematic Framework', Proceedings of the Fifth IEEE International Symposium on High Performance Distributed Computing, Pages 318-328, Syracuse, New York, August 1996.

I. Foster and C. Kesselman, `Globus: A Metacomputing Infrastructure Toolkit', Proceedings of the Workshop on Environments and Tools for Parallel Scientific Computing, SIAM, Lyon, France, August 1996.

G. Fox and W. Furmanski, `Towards Web/Java based High Performance Distributed Computing - An Evolving Virtual Machine', Proceedings of the Fifth IEEE International Symposium on High Performance Distributed Computing, Pages 308-317, Syracuse, New York, August 1996.

J. Gray and A. Reuter, Transaction Processing: Concepts and Techniques, Morgan Kaufmann, 1993.

A. S. Grimshaw, W. A. Wulf, and the Legion team, `The Legion Vision of a Worldwide Virtual Computer', Communications of the ACM, Volume 40, Number 1, Pages 39-45, January 1997.

S. Maffeis, `Adding Group Communication and Fault-Tolerance to CORBA', Proceedings of the 1995 USENIX Conference on Object-Oriented Technologies, Monterey, California, June 1995.

S. Maffeis, `iBus: The Java Intranet Software Bus', available at, Olsen and Associates, Zurich, 1997.

R. Ramamoorthi, A. Rifkin, B. Dimitrov, and K. M. Chandy, `A General Resource Reservation Framework for Scientific Computing', Proceedings of the First International Scientific Computing in Object-Oriented Parallel Environments (ISCOPE) Conference, Volume 1343 of Springer-Verlag's Lecture Notes in Computer Science, Pages 283-290, Marina del Rey, California, December 1997.

A. Rifkin, `GEM: A Global Event Model for Using Events in Distributed Systems', Caltech Infospheres Group Internal Report, California Institute of Technology, 1998.

D. S. Rosenblum and A. L. Wolf, `A Design Framework for Internet-Scale Event Observation and Notification', Proceedings of the Fifth ACM SIGSOFT Symposium on the Foundations of Software Engineering, Pages 344-360, Zurich, Switzerland, 1997.

E. M. Schooler, `A Multicast User Directory Service for Synchronous Rendezvous', Technical Report CS-TR-96-18, Department of Computer Science, California Institute of Technology, Pasadena, California, 1996.

R. van Renesse, K. P. Birman, and S. Maffeis, `Horus: a Flexible Group Communication System', Communications of the ACM, Volume 39, Number 4, April 1996.

About this document ...

This document was generated using the LaTeX2HTML translator Version 96.1 (Feb 5, 1996) Copyright © 1993, 1994, 1995, 1996, Nikos Drakos, Computer Based Learning Unit, University of Leeds.

The command line arguments were:
latex2html -split 0 gem.

Adam Rifkin, Dec 16, 1997