Simple Comparisons on some In-Memory Data Grids

 

Fetures	Feature Description	GemFire7	Coherence3.7	Infinispan5	Gigaspaces7	Notes
1. topologies	peer-peer; client-server	High	High	High	High
2. cross-site /   WAN replication	datacenter-datacenter; region-region	High	Low	Med	High	Coherence support by incubator "Push replication"; Infinispan basic support in V5.2
3. read-most scalability	via replication	High	High	High	High
4. write-most scalability	via partitioning and replication between primary and backups	High	High	Med	High
5. high availability / HA	via replication and persistence  into disk (DB)	High	High	High	High
6. asychronous     replication	useful for slow changing data and WAN replication	High	Med	High	High	Coherence support by incubator "Push replication";
7. partition rehashing	consistent hashing to reduce data relocation	High	High	High	High	all seem to use consistent hashing-like algorithms
8. dynamic clustering	adds or loses nodes	High	High	High	Med
9. updating among     partition backups	master-backup: less deadlock prone and IO master-master/update anywhere: more dealock prone and IO	High	High	Low	High	Infinispan only has master-master that incurs deadlock for slow network and large caches
10. cache loader and       writer	2 application interfaces: loades data from DB and saves data into Cache. With cache loader and writer, applications only need to interace with Cache!	High	High	Med	High	Infinispan writers only support JPA, not JDBC or Hibernate!
11. read-through and      read-ahead	populates cache with DB in batch mode	Med	High	Med	High	no prefetch/batch support from Gemfire and infinispan
12. write-through and      write-behind	persists cache into wherever they were loaded.write-behind persists data asynchronously	High	High	Med	High
13. event notification	Cache also works as a messaging and parallel processing bus like JMS/MDB	Med	Med	Med	High
14. continous querying	register contents-based interests with Cache and receive updates continously.	High	High	Low	High
15. cache querying	SQL like query: not only searches  based on key matching	High	High	Low	High
16. locking and tx	JTA (global tx) and ACID properties	High	High	Med	High
17. off-heap	puts cache data off Java heap so that GC pause time is reduce!	Low	High	Low	Low
18. key affinity /      colocation	colocates related cache objects on the same partition to reduce IO	High	High	High	High
19. customized      partitioning	puts cache into specific cluster node to bypass cache hashing algorithm	High	Low	Med	Low
20. synchronous      requests like RFQ	synchronou requests should go through the same partition routing as asynchrous messages.	Low	Low	Low	High
21. API	Map,Restful and appropriate interfaces	Med	Med	Med	High
22. language bindings	Java, C++ etc	Med	Med	Med	High
23. map-reduce /      scatter-gather	submits aggregation tasks to run across multiple or all cluster nodes	High	High	High	High
24. monitoring	monitor the health of clusters	High	Med	Med	High
25. J2EE (JMS,Web,      Remoting)	how much J2EE? to support?	Low	Low	Low	High
26. migration effort	how to migrate to a different cache production in a J2EE app. server	High	High	High	Low	GigaSpaces is an app. server

Locking Schemes for Replicated Data Updating

Data distribution is commonly used in high-performance computing (HPC). Basically there are two fundamental data distribution topologies. One is replication; the other is partition.

With data partition, you can achieve parallel processing of a large amount of data.
With data replication, you can achieve load balancing and high availability(HA).

Even though a data item has several replicas in a data replication environment, it should have some degree of transparency and appear to only one virtual global item to end users.
The biggest challenge using data replication is the proper trade-off between data coherence and performance based on your business requirements.

Some kind of locking scheme is usually employed in order to achieve data coherence.
I will list some replication and locking schemes I have experienced using Oracle10g advanced replication, Oracle10g RAC, Oracle10g TimesTen and Gigaspaces XAP 7.1.

Before we go to details, let's suppose you have a distributed airline ticketing system (DATS hereafter) which has two databases: one is in NY and the other is in LA. Depending your replication scheme, data can be updated either at one site only and replicated to the other or at both sites and replicated to the each other.
Further suppose the following actions take place in time sequence:

1. There is only one ticket left in the database. Because both local database replicas are synchronized at this time, the only ticket can be taken by either NY or LA;
2. A NY customer bought this ticket. This action was updated in the local NY database and will be replicated to LA somehow depending on the replication scheme;
3. Depending on your replication scheme, the LA database can show the same ticket either still available or already taken by a NY customer. If the same ticket still appears available to the LA database, it will be sold to a LA customer. This will create an oversold situation.

1. Synchronous Replication Using Distributed Locks and Local Transactions

Oracle RAC (formerly OPS in version 8i and prior) allows the same physical database to be accessed at multiple instance sites. In order to allow users to read and write any data at any time at any instance site, Oracle RAC uses "Cache Fusion" to ensure data coherence.

"Cache Fusion" basically uses synchronous replication with a distributed locking manager (DLM). DLM acts as a distributed lock (DL) coordinator among other functions so that the same resource - e.g. a table row - can only be granted to one instance site for changing at a time while other sites have to wait.

DLM also act as a global resource directory. For example, when instance site 1 updated a row, it doesn't need to actively push the new version of data to all other instance sites.  When instance site 2 later requests the same row, DLM can tell it to get the latest version from instance site 1.
Also instance site 1 doesn't need to use any distributed transaction thanks to DLM and the fact that there is still only one physical database (so far I haven't seen any synchronous replication that use both distributed locks and distributed transactions).

Benefits include very high degree of data coherence and load balance for both reads and writes.
Drawback include poor write performance and requirements of high-speed interconnect due to distributed locks.
Distributed locks usually consist of quite a few daemon processes and data structures at each site whose coordination performs poorly in a low-speed interconnect such as LAN and WAN. For Oracle "Cache Fusion", distributed locks are implemented with Global Cache Service (GCS), Global Enqueue Service (GES) and Global Resource Directory (GRD).

If we apply this scheme to DATS assuming the poor interconnect performance is tolerable, step 3 has to wait for the DL to be release by step 2. When step 3 gets the DL, the same ticket will show already taken by step 2.
(In order to have good performance, most multi-tiered applications use optimistic locking that can create lost update problem. For example if we use optimistic locking in both databases in DATS, the application tier in step3 can first read the LA database before step 2 and then sell the same ticket to a LA customer after step 2.
The application must use "optimistic locking with version checking" to fix this issue. One version checking is just a version number that increases whenever there is any corresponding data change.
Suppose the version is 0 at step 1. Step 2 updates it to 1. The version found by the application tier read at step 3 is also 0. But when the application tier tries to sell the same ticket, it will fail because it will find the version has changed to 1 from its cached value 0.
All our arguments assume "optimistic locking with version checking" .)

2. Synchronous Replication Using Local Locks and Distributed Transactions
Oracle's multimaster replication (also called peer-to-peer or n-way) has two data coherence protocols.
One of them is synchronous replication that applies any changes or executes any replicated procedures at all sites participating in the replication environment as part of a single distributed transaction. If the DML statement or procedure fails at any site, then the entire transaction rolls back.

The distributed transaction ensures data consistency at all sites in real-time. However it doesn't use any distributed locking. Instead it only uses local locks in the participant local transaction.
This is the case when an application performs a synchronous update to a replicated table. Oracle first locks the local row and then uses an after row trigger to lock the corresponding remote row. Oracle releases the locks when the transaction commits at each site.

Benefits include high degree of data coherence, simple implementation, easy administration and fit to both high-speed interconnect and low-speed LAN and WAN (implementing distributed locks in low-speed LAN and WAN is much harder than using lock locks in such environments).
Drawbacks include possible deadlock due to temporal behavior of local and remote locking, high availability requirements on network and poor write performance.

If we apply this scheme to DATS assuming the poor interconnect performance is tolerable, step 3 has to wait for step 2 to release the local and remote locks. When step 3 gets the locks, the same ticket will show already taken by step 2.

3. Synchronous Replication Using Local Locks and Local Transactions
TimesTen's unidirectional active-standby pair configuration only uses so called "return twosafe
replication". It provides fully synchronous replication between the master (the active site) and subscriber (the standby site).
No distributed transaction or distributed lock is involved. Instead only local transaction and local lock are used. Specifically the local transaction on the subscriber is first committed before on the master. If the subscriber can't commit, the master will not commit either.
At any time, only the active site can be updated that greatly simplifies data updating complexity (otherwise using local locks and local transactions will be insufficient) and ensures fast fail-over to the standby site in case of the active site failure.

This scheme has similar benefits and drawbacks to the previous scheme in section 2.
However its performance is even better thanks to the avoiding of two-phase commit (2PC) required in a distributed transaction. It also eliminates the deadlock because only the active site allows updates.
Although the standby site seems to be a waste of capacity, you can collocate it with another active site as shown in figure 1 (by another I mean it has different data from the collocated standby site).
This scheme has lower degree of data coherence due to the inconsistency resulted from master commit failure even the subscriber has successfully committed (the root cause is it doesn't use distributed transactions as you can guess. But you should also know that data inconsistency can still result from the second "commit" phase failure in a 2PC process).

TimesTen's practice in this scenario is consistent with its configurations for high performance in other areas such as asynchronous logging and data caching with write-behind policy, 

Gigaspaces IMDG has a very similar topology called primary-backup replication. The only difference is it uses distributed transactions instead of local transactions only. So it has higher degree of data coherence than TimesTen.
Another advantage is the fail-over happens in Gigaspaces IMDG transparently to end users while TimesTen customers need to resort to some third-part cluster manager or some custom software.



If we apply this scheme to DATS, either NY or LA site will be the active site and the other standby site has to connect to the active for data updating (in reality, active-standby often is used with a high-speed interconnect). The local lock in the active site prevents the oversold situation.

This scheme along with data partitioning as shown in figure 1 is strongly recommend compared to the two previous synchronous schemes if it can your business requirements.
Although the two previous synchronous schemes allow updating anywhere, updating the same resource entails costly lock coordination over network. Scalable updating is usually achieved by data partitioning.
Although the two previous synchronous schemes allow distributed and scalable reads, you can fine-tune your partitions to allow more current reads.

Figure 1: The Primary-Backup Partition in Gigaspaces

4. Asynchronous Replication with Update Anywhere
Another data coherence protocol in Oracle's multimaster replication is asynchronous replication that allows users to update data at any participant site.
This scheme is also used in Oracle's updatable materialized view replication and TimesTen's bidirectional master-subscriber replication for general distributed workloads.

With this scheme the data changes at one site will be queued for propagation to other sites and committed locally. The queued changes will be propagated in batches in a separate transaction. So it doesn't use any distributed lock or distributed transaction. Instead it only use local locks required in the corresponding local transaction.

Benefits include great read and write performance, easy implementation, simple administration and fit to low-speed interconnection such as LAN and WAN and disconnected updating.
Drawbacks include limited degree of data coherence depending on data refresh frequency, and possible data change conflicts.
Because there is no distributed lock or distributed transaction involved, a replication conflict occurs if two transactions originating from different sites update the same row at nearly the same time (when the queued changes are propagated to the other site, the other site will have two versions of data changes. So which one should take place?).

A conflict resolution method must be provided to resolve the data inconsistency. Both Oracle and TimesTen have a prebuilt "latest timestamp" resolution method that makes the change with the latest timestamp the winner. Oracle also allows you to customize a resolution method based on your business requirements.

This scheme can't be applied to DATS if oversold situations are not allowed because the changes at NY and LA sites can be committed independently in two different transactions that result in the same ticket being sold to two customers.
If occasional oversold situations are allowed, the NY and LA sites can sell tickets at different times thanks to the three hours time zone difference. If a replication conflict does occur, relevant information should be recorded in the database based on which your front-end application takes proper actions (in reality a reservation system like DATS doesn't use this scheme).

5. Asynchronous Replication with Update on Master Site only

This scheme is used in Oracle's read-only materialized view replication, TimesTen's unidirectional master-subscriber replication and Gigaspaces IMDG's master-local replication.

This scheme basically has similar benefits and drawbacks to the previous scheme in section 4. However because it only allows updates at the master, it eliminates the notorious replication conflicts, which most of the time proves to be a very sound design in an asynchronous replication environment.

If we apply this scheme to DATS and suppose NY is the master site (or a third site is the master), NY has to wait if LA first gets the local lock at the master site. The local lock in the master site prevents the oversold situation.

Life is much easier using Gigaspaces IMDG's master-local topology as shown in figure 2 because it automatically delegates your local cache updating to the master which propagates the same updating to other local caches. Gigaspaces IMDG also supports optimistic locking with versioning.
You must do both by yourself if you use Oracle's read-only materialized view replication and TimesTen's unidirectional master-subscriber replication.

Figure 2: Gigaspaces Master-Local Topology where Master can be Figure 1

TimesTen -- an Oracle Database Saviour for Performance-Critical Applications?

Performance-critical applications such as those used in trading and telecoms often need low latency in sub-milliseconds or even microseconds and high throughput.
If such an application uses a disk-based database for persistence (or the system of record), you probably will have to miss your low latency requirement because disk-based accesses incur latency in many milliseconds and often seconds.

In order to achieve low latency and high throughput, there should be at least 3 requirements to be met:

1. Uses memory-based accesses. The access rate for the commonly wildly used computer DRAM is in microseconds;
2. Collocates your application logic with your data. The traditional multi-tiered approaches incurs network IO latency.
3. A fast and predictable JIT compiler if you use Java. Oracle's JRockit is a good choice.

I know 2 products which can meet the above requirements and worth a try.
One is to adopt Gigaspaces' space-based programming paradigm.
The other is to employ an in-memory database (IMDB) such as Oracle's TimesTen along with your traditional database such as Oracle.

I will focus on TimesTen here and compare it with Gigaspaces.
As mentioned above, using Oracle database alone in a performance-critical application is not practical. Accordingly Oracle just doesn't have the same popularity in such applications as in others.
Fortunately if we use Oracle along with its cache companion - TimesTen, it is a very attractive option.

TimesTen is an in-memory relational database from Oracle. Because it caches all data in memory and has no disk access, TimesTen can achieve breakthrough performance such as 5-microsecond reads and 15-microsecond updates based on its white paper "Using Oracle In-Memory Database Cache to Accelerate the Oracle Database".
TimesTen is targeted to run in your application tier, close to applications, and optionally in process with applications. A TimesTen database may be used as the database of record, and/or as a cache to an
Oracle database.

As a cache to Oracle database, TimesTen can be embedded into your application tier and only cache the performance-critical subset of your Oracle database while still leaving your Oracle database for non-performance-critical applications. This architecture is shown in Figure 3 in the above white paper:

In the above figure, you can only cache the subset of an Oracle database that needs real-time processing such as stock trading data. The cached data can be either read-only, read-mostly or read-write. In case of data modification, the "Cache Agent" will take care of the data synchronization between TimeTen and Oracle.
You can also scale out your application tier horizontally by deploying several TimeTimes together to form a so called cache grid.

Here are my TimesTen comparisons to Gigaspaces.
Commons to Both:

• cache data along with your application logic;
• supports multiple topologies such as partitioned, replicated or both;
• supports ACID properties often provided by traditional relational databases; 
• supports asynchronous operations for better performance such as transaction write-behind and asynchronous replication;
• besides as a cache bus, they can also be used as a message bus.

Pros and Cons on TimesTen and Gigaspaces:

• Gigaspaces can cache all your data in a computing cloud such as Amazon's EC2 while TimesTen only practically caches subset of your whole enterprise data;
• Gigaspaces has more flexible deployment topologies than TimesTen. For example, Gigaspaces can adjust your deployment based on your SLA while TimesTen's deployment tends to be static;
• Gigaspaces supports application level optimistic locking while TimeTen only provides 2 traditional database's isolation levels - read-commit and serializable;
• Gigaspaces intrinsically supports parallel processing such as map-reduce paradigm through its master-worker pattern and thread executor framework while using TimesTen you have to build up such functions by yourself.  For example, if you want to do a search in your TimesTen cache grid (suppose you partitioned your data in the grid), you have to code by yourself to submit tasks to grid memebers and finally aggregate the each individual searching result;
  
• It is easier to adopt TimesTen than Gigaspaces because TimesTen intrinsically supports the popular JDBC API while Gigaspaces's intrinsic JavaSpaces API is still not that popular; 
• TimesTen is budget and deployment friendly because it only needs to cache a subset of your whole enterprise data. Most often a large amount of your enterprise data don't need real-time accesses and accordingly caching them may be wasting your valuable resources.
• TimesTen has closer integration with Oracle than Gigaspaces does. For example, TimesTen supports the same flexible locks as in Oracle and materialized views;
• TimesTen seems to have a bit better performance. For example, TimesTen can achieve 5-microsecond reads and 15-microsecond updates while Gigaspaces can achieve sub-millisecond latency at best. This may be because TimesTen was implemented by C/C++ and Gigaspaces was implemented Java (JVM has pause).

    

Master-Local -- the most common topology in IMDG

IMDG (In-Memory-Data-Grid) functions like an in-memory database to the front end users. In order to be scalable, IMDG has to provide partitioned topology besides the replicated one.

If we bundle an application locally with a data partition together and the application only processes its local data, we form what the GigaSpaces calls a PU(Processing Unit).
Because PU doesn't need to communicate with others, we can achieve linear scalability by deploying multiple PUs.

However such a tightly couple application rarely exists practically.
On one hand IMDG basically partitions data evenly without caring about too much of your application. After all IMDG, as the delegate of a system of record (the database), should be able to server all applications.
On the other hand your application should implement your specific business logic and you may have different applications for different business requirements.

So your universal data and your specific business requirements are just conflicting with each other.
To solve this dilemma, a Master-Local topology is often employed where the master cache (or Data-only Processing Unit or EDG in GigaSpaces) itself can be either replicated, partitioned or both; the local cache (or embedded space in GigaSpaces), caches data from, and synchronizes any update with, the master on demand. The communication between local and master is usually a single network hop and extreme low latency can be achieved using switches.
Here is a diagram from GigaSpaces where the master is replicated:

If your local cache is read-only, the performance is the best.
Oracle Coherent also calls the above topology as Near Cache.

Going back to GigaSpaces' "XAP Order Management Tutorial" in its XAP 7.0.0 "Quick Start Guide" section. As its comment mentions, the AccountData comes from IMDG which is simulated by AccountDataLoader in the example. In practice it should come from an EDG which connects to the local embedded space in the order runtime processing unit.