Oracle Coherence Distributed Auction Example

Introduction

The Oracle Coherence Incubator provides an example of a distributed auction application where near real-time automated bidders concurrently read and raise bids on a finite set of items for sale.

This application uses the Incubator's Push Replication facility to allow two distinct Coherence clusters to support bidding applications. The replication models used in this application are active-active and active-passive replication. The active-active model permits concurrent updates to occur in two Coherence clusters. The active-passive model restricts updates to a cache in a single cluster, and propagates updates to a passive (read-only) cache in the other cluster.

(Note: a Coherence cluster is simply a set of Coherence cache servers and applications living within a single local area network, serving a shared set of caches and applications. Typically the cluster defines the bounds of where data in a cache lives, but with Push Replication, data in caches are replicated to other defined clusters).

Application Model

The current examples supports the auction for two clusters: "Site1" and "Site2". "Site1" is special in that it is where the Auctioneer runs, and is where the auction starts and is shutdown.

There are two caches used in this demo, one that stores the bidding information ("bidding-cache") and one that holds the state of the auction itself ("control-cache"). Each uses a different Push Replication deployment model.

bidding-cache

control-cache

The application has three small applications that support the auction:

• The Auctioneer - this populates a bidding cache with items to be put up for auction. Each item is identified by a unique item id (long) and is assigned a random starting price. The initial customer id for the item is empty, which indicates that no one has bid on the item. The Auctioneer's job is to seed the cache with items, start the auction, update the time remaining in the auction, then shut down the auction after a fixed amount of time. The Auctioneer then terminates.
  By convention, the Auctioneer needs to run once somewhere at Site1, to initialize and control the auction. The Auctioneer takes two arguments:
  the number of items being put up for the online auction (e.g. 30) and the length of the auction in minutes (e.g. 5).

• The Bidder - this represents an application that bids on items for sale in the auction. It runs concurrently on both clusters. Minimally the example will want at least one Bidder to run in each cluster, but this is not a limitation. Multiple bidders can run within a single cluster. The Bidder takes a single argument: the site name (String) e.g. "Site1" or "Site2". On "Site1", customer ids for bidders are randomly chosen from a list of Warner Brother's cartoon characters (e.g. "Daffy Duck"). On "Site2" customer ids for bidders are chosen from Disney characters (e.g. "Mickey Mouse"). The Bidder works by waiting for the Auctioneer to seed and start the auction. Once the auction is started, a Bidder picks an item randomly in the auction, retrieves its price, and then raises the prices by a random percentage. This becomes a new bid, which is put back into the cache. The Bidder sleeps for a short period of time and then performs the same task: picking another item, getting its price and raising it. The Bidder continues this until it sees that the Auctioneer has shutdown the auction, at which point the Bidder stops bidding and terminates. An important semantic is that if the Bidder sees that the auction is active and proceeds to make a bid, that bid will be processed by the system.

• The BidObserver is an application whose job is to look at the bidding in a cluster and report on what it sees. It is a passive application that only observes activity. Typically only one instance needs to be run in each cluster during an auction. When the auction ends and the Auctioneer and Bidders terminate, the BidObserver is useful to watch the bidding cache in both clusters. After both sites replicate the last bids made during the auction, both sites should see the same highest bid made by a Bidder running in either Site1 or Site2. (Typically the final bids are replicated and reconciled between two clusters within few seconds after the auction is stopped.)

Running the Auction example

By necessity, this example has a lot of moving parts. To run the auction example you'll need seven command line windows. Before starting you'll want to make sure that you've set the JAVA_HOME and COHERENCE_HOME variables in the set-env.(sh|bat) script. Once done to run the example you'll want to:

1. In the each window run or source the bin/set-env(.sh|bat) script.
2. In the first window run an auction cache server at Site1 by running: bin/run-incubator.(sh|bat) resource/pushreplicationpatter/auction/auction-server1.properties.
3. In the second window run an auction cache server at Site2 by running: bin/run-incubator.(sh|bat) resource/pushreplicationpattern/auction/auction-server2.properties.
4. In the third window run a bidder at Site1 by running: bin/run-incubator.(sh|bat) resource/pushreplicationpattern/auction/bidder1.properties.
5. In the fourth window run a bidder at Site2 by running: bin/run-incubator.(sh|bat) resource/pushreplicationpattern/auction/bidder2.properties.
6. In the fifth window run a bid observer at Site1 by running: bin/run-incubator.(sh|bat) resource/pushreplicationpattern/auction/bidobserver1.properties.
7. In the sixth window run a bid observer at Site2 by running: bin/run-incubator.(sh|bat) resource/pushreplicationpattern/auction/bidobserver2.properties.
8. In the seventh and last window run the auctioneer by running: bin/run-incubator.(sh|bat) resource/pushreplicationpattern/auction/auctioneer.properties.

With the first six processes running, the servers, the bidders, and the observers will be waiting for the auction to start. When the seventh process is started,
the auctioneer will put up items for bid and start up the auction. When the auctioneer ends the auction, bidders will terminate, and the observers will
reconcile data and show the same final bids at both sites.

A example defaults with the Auctioneer posting 30 items for sale for 5 minutes. This data set is relatively small and easy to monitor, and shows a high degree of contention between competing bidding applications.

Data Model

The data model supporting this application uses two caches: a "bidding-cache" and a "control-cache".

The "bidding-cache" is a distributed homogeneous Coherence Named Cache that is actively replicated by Site1 and Site2. This means that Bidders running in either Site1 or Site2 can make updates, and these bids will be replicated to the other site every ten seconds. Conflict resolution logic is used to reconcile bids between Bidders both within and across clusters.

A "BiddingItem" object that contains a price and a CustomerReference object populate the "bidding-cache". The CustomerReference object contains a customer id number and site name. The combination of both provides a unique customer id across both clusters.

The key of a bidding item is a unique integer cast as a String. If the Auctioneer is told to seed the auction with 30 items, 30 BiddingItems will be created to populate the cache with keys numbering from 1 to 30.

The "control-cache" is populated and updated only by the Auctioneer with a singleton entry whose key is a String "control". The entry is an AuctionControl object that contains a Boolean indicating whether or not the auction is active. It also contains an item count indicating the number of items in an auction, and has member variables that track the time left in the auction. The Auctioneer updates these fields as the auction progresses.

The "control-cache" is an active-passive cache that is only updated by the Auctioneer running at Site1. The purpose of the "control-cache" is to quickly replicate to the passive site (i.e. Site2), when the auction is started and stopped. This allows Bidders to start and stop bidding at roughly the same time. The "control-cache" is modeled on both systems as Coherence "near" cache with an invalidation policy of "present". Basically this means that the Bidders get a copy of the AuctionControl item local to their JVM for locality of reference.

Conflict Resolution

Conflict resolution refers to logic needed to resolve conflicts between concurrent updates of the same data item. In the auction example, conflict resolution needs to be dealt with in two discrete contexts: when two Bidders raise a bid on the same item concurrently within a single cluster, and when a conflict occurs between Bidders running in two separate clusters.

To deal with both contexts, two different mechanisms are used to deal with the same problem. Algorithmically, conflict resolution for bidding is that the highest price always wins. For the fringe case where two Bidders concurrently bid the same price, a simple-minded tiebreaker is used using the CustomerReference. This logic is embedded within two Java classes tied to the application: the PlaceBidProcessor and BidderConflictResolver.

PlaceBidProcessor is an implementation of Coherence's AbstractEntryProcessor that allows a Bidder to change a bid atomically within a cluster. When a Bidder retrieves and item and decides to increase a bid, it creates an instance of PlaceBidProcessor, containing the Bidder's customerId (i.e. Site name), and the new price. The Bidder then invokes the PlaceBidProcessor object against the "bidding-cache". Coherence locks the data item in the cluster and then executes the "process" method in the object.

From the time it takes the Bidder to read a price, compute a higher price, and submit it as a bid, a concurrent Bidder might have submitted a new bid and changed the price. The purpose of the PlaceBidProcessor is to check the price, as it exists in the cache and do the right thing: if the existing price in the cache is higher than the bid made by the Bidder, the operation is a no-op. If the existing cache has a price lower than the Bidder, then the Bidder's price is accepted and cache entry is updated with the Bidder's price and the new customer reference.

The other context where a conflict occurs is when updates to a "bidding-cache" are replicated to the other site. To handle this, a BidderConflictResolver object must be declared by any application making updates to the "bidder-cache" (in this case this is the Auctioneer and the Bidder.)

The BidderConflictResolver defines a "perform" method that is analogous to the BidderEntryProcessor. It has access to the bid being replicated from the other site, and the current bid existing at the target site. Like the BidderEntryProcessor it compares the two prices, and the highest bid is retained. The key difference between the BidderConflictResolver and the BidderEntryProcessor is when this conflict is resolved.

Again, conflict resolution is only needed for BiddingItem objects stored in the "bidding-cache". Since the Auctioneer only updates the AuctionControl object at Site1, no conflict resolution needs to be defined for the "control-cache".

Below is an illustration of state changes. Figure 1 shows two caches living in two Coherence clusters that are synchronized. Figure 2 shows that independent bidding activity can result in a divergence in updates where a conflict exists. Figure 3 shows the process of replication utilizing conflict resolution to resolve the divergence.

Summary

The Auction application showcases the power of using Coherence clusters and push replication functionality. With just few hundred lines of application code a distributed active-active application can be written with simple Coherence NamedCache primitives. The auction also shows how a single application can use caches with different replication policies to solve various problems.

In general, active-active replication is useful when the semantics of conflict resolution are simple, algorithmic, and can be resolved by looking at the state of a replicated object and the state of the object in the target cluster. This auction example works because replication is at the data level, and because conflict resolution logic creates the desired effect of the application, to preserve the highest bid. Other classes of applications that could support this type of replication include package-tracking applications where the Coherence clusters stores tracking data and where the latest tracking observation of an object represents the best observation, and the one that is preserved.

Active-active typically works best for data level replication, not transactional level replication where global serialization must be enforced.