Distributed Thoughts

What is Apache Ignite?

2017-09-13T20:03:00.000-07:00

If you ever looked at Apache Ignite, you have probably noticed that it is a fairly rich platform with lots of components. However, despite the extensive feature set, Ignite community aims to make the platform easy to use and understand. Here is how the Ignite community defines their project:

Apache Ignite is

the in-memory computing platform

that is durable, strongly consistent, and highly available
with powerful SQL, key-value and processing APIs

So, in summary, Ignite looks like a distributed data storage that can work both, in-memory and on-disk, and provides SQL, key-value and processing APIs to the data. Sounds simple enough. However, to get a complete picture, perhaps it is better to define Ignite by answering several "Is Ignite a ...?" questions:

Is Ignite a persistent or pure in-memory storage?

Both. Native persistence in Ignite can be turned on and off. This allows Ignite to store data sets bigger than can fit in the available memory. Essentially, the smaller operational data sets can be stored in-memory only, and larger data sets that do not fit in memory can be stored on disk, using memory as a caching layer for better performance.

Is Ignite an in-memory database (IMDB)?

Yes. Even though Ignite durable memory works well in-memory and on-disk, the disk persistence can be disabled and Ignite can act as a pure distributed in-memory database, with support for SQL and distributed joins.

Is Ignite an in-memory data grid (IMDG)?

Yes. Ignite is a full-featured data grid, which can be used either in pure in-memory mode or with Ignite native persistence. It can also automatically integrate with any 3rd party databases, including any RDBMS or NoSQL stores.

Is Ignite a distributed database?

Yes. Data in Ignite is either partitioned or replicated across a cluster of multiple nodes. This provides scalability and adds resiliency to the system. Ignite automatically controls how data is partitioned, however, users can plugin their own distribution (affinity) functions and collocate various pieces of data together for efficiency.

Is Ignite an SQL database?

Not fully. Although Ignite aims to behave like any other relational SQL database, there are differences in how Ignite handles constraints and indexes. Ignite supports primary and secondary indexes, however, the uniqueness can only be enforced for the primary indexes. Ignite also does not support foreign key constraints.

Essentially, Ignite purposely does not support any constraints that would entail a cluster broadcast message for each update and significantly hurt performance and scalability of the system.

Is Ignite a transactional database?

Not fully. ACID Transactions are supported, but only at key-value API level. Ignite also supports cross-partition transactions, which means that transactions can span keys residing in different partitions on different servers. At SQL level Ignite supports atomic, but not yet transactional consistency. Ignite community plans to implement SQL transactions in version 2.4.

Is Ignite a key-value store?

Yes. Ignite provides a feature rich key-value API, that is JCache (JSR-107) compliant and supports Java, C++, and .NET.

You can find out more about Ignite by visiting the freshly redesigned Ignite website.

Apache Ignite - In Memory Performance with Durability of Disk

2017-09-07T17:39:00.000-07:00

Ignite is the in-memory computing platform
that is durable, strongly consistent, and highly available
with powerful SQL, key-value and processing APIs

Starting with 2.1 release, Apache Ignite has become one of a very few in-memory computing systems that provides its own distributed persistence layer. Essentially, users do not have to integrate Ignite with any type of 3^rd party databases (although such integration is supported), and start using Ignite as a primary storage of their data on disk and in memory.

So, what makes Ignite data storage unique? Let us look at a few important features provided by Ignite. You will probably notice that some of these features can also be seen in other data storage systems. However, it is the combination of these features in one cohesive platform that makes Ignite stand out among others.

1. Durable Memory

Ignite durable memory component treats RAM not just as a caching layer, but as a complete fully-functional storage layer. This means that users can turn the persistence on and off as needed. If the persistence is off, then Ignite, just like always, can act as a distributed In-Memory Database or as an In-Memory Data Grid, depending whether you prefer to use SQL or key-value APIs. If the persistence is turned on, then Ignite becomes a distributed, horizontally scalable database that guarantees full data consistency and is resilient to full cluster failures. On top of that, the data is stored in off-heap memory so there are no GC pauses even on large data sets.

2. Complete SQL support

With the latest release, in addition to SQL querying, Ignite added support for DDL and DML, allowing users to interact with Ignite using pure SQL without writing any code. This means that users can create tables and indexes, insert, update, and query data using only SQL. Having such complete SQL support makes Ignite a one-of-a-kind distributed SQL database.

3. ACID compliance

Data stored in Ignite is ACID-compliant both in memory and on disk, making Ignite a strongly consistent system. Ignite transactions work across the network and can span multiple servers. This makes Ignite stand out from the eventually consistent NoSQL systems that hardly support any type of transactions.

4. Collocated Processing

Most traditional SQL and NoSQL databases work in a client-server fashion, meaning that data must be brought to the client side for processing. This approach requires lots of data movement from servers to clients and generally does not scale. Ignite, on the other hand, allows for sending computations to the data, moving only the light weight compute functions across the network. As a result, Ignite scales better and minimizes data movement. When collocated, all the data processing happens locally on the node that stores the data, and only the result is brought back to the user.

5. Scalability and Durability

Ignite is an elastic, horizontally scalable distributed system that supports adding and removing cluster nodes on demand. Ignite also allows for storing multiple copies of the data, making it resilient to partial cluster failures. If the persistence is enabled, then data stored in Ignite will also survive full cluster failures. Cluster restarts in Ignite can be very fast, as the data becomes operational instantaneously directly from disk. As a result the data does not need to be preloaded in memory to begin processing, and Ignite caches will lazily warm up resuming the in memory performance.

GridGain In-Memory Data Fabric 7.5 is Released

2016-02-11T02:39:00.001-08:00

Today the GridGain team has announced the release of enterprise-grade GridGain In-Memory Data Fabric v. 7.5, based on Apache Ignite^tm v. 1.5. For those not familiar with GridGain or Apache Ignite, it provides the ability to distribute, cache, and compute on data in memory, including such features as in-memory data grid, compute grid, ANSI-99 in-memory SQL, real-time streaming, in-memory file system, and many more.

Some of the most important features of this release, among others, include deadlock-free in-memory transactions, significant improvements to the zero-deployment model, and major performance improvements. All these features have been available in Apache Ignite 1.5 for a while, but now, after many rounds of load testing and bug fixes, have finally received GridGain's ready-for-production stamp of approval.

Deadlock-Free Transactions

Deadlocks usually happen when the same objects are concurrently updated in a different order by different transactions. Transactions begin to indefinitely wait on each other to complete, causing deadlock scenarios. Such problems are very difficult to spot and are even harder to debug. In production, they would require a full cluster restart, leading to costly system down times.

The traditional solution to the deadlock problem is to ensure that applications acquire locks in the same order. However, this is easier said than done, especially in large distributed teams. Just imagine how many objects may be updated by a simple "transfer(...)" method on some bank's API. Grouping such calls in a common transaction is almost certain to generate a deadlock.

A much better solution is to drop the locks altogether, which is what Apache Ignite community did. Essentially, all the transactions are given a chance to succeed until it is impossible to logically order some transaction. When this happens, the transaction optimistically fails with an exception and is allowed to be retried. Turns out that this optimistic-serializable consistency model is about 50% faster than its pessimistic counterpart. However, the biggest benefit is that the deadlocks are now impossible.

Zero-Deployment (revisited)

With this release user objects are always kept in the binary form and are never deserialized on the server side. This makes GridGain servers agnostic to user domain models, allowing users to dynamically add or remove fields to their data types, or create and deploy new data types on the fly. When it comes to executing computations, Ignite provides a distributed class-loader which will automatically undeploy the old computation logic and deploy the new one.

Such combination of the binary protocol together with the distributed class-loader creates a deployment-free cluster environment, where both data model and computation logic can be dynamically updated without any explicit deployment steps or down times.

Pushing Performance Boundaries

With the introduction of the revised compact binary protocol, GridGain and Ignite became a lot faster, beating its nearest data grid competitor, Hazelcast, by over 100% in throughput and latencies. The performance benchmarks are public and can be viewed and downloaded on the Ignite website.

Other features of 7.5 release include OSGI compliance as well as new data streamers, including support for Twitter, MQTT, and Flume real time streams.

The latest release can be downloaded here.

Benchmarking Data Grids: Apache Ignite vs Hazelcast, Part II

2015-05-19T14:56:00.001-07:00

In my previous post I have demonstrated benchmarks for atomic JCache (JSR 107) operations and optimistic transactions between Apache Ignite^tm data grid and Hazelcast. In this blog I will focus on benchmarking the pessimistic transactions.

The difference between optimistic and pessimistic modes is in the lock acquisition. In pessimistic mode locks are acquired on first access, while in optimistic mode locking happens during the commit phase. Pessimistic transactions provide a more consistent view on the data, given that, since locks are acquired early, you are guaranteed that no changes will happen to the data between transaction start and commit steps.

Yardstick Framework

Just like before, I will be using Yardstick Framework for the benchmarks, specifically Yardstick-Docker extension.

Transparency

One of the most important characteristics of any benchmark is full transparency. The code for both, Apache Ignite and Hazelcast benchmarks is provided in the corresponding GIT repos:

Both, Apache Ignite and Hazelcast teams were given the opportunity to review the configuration and provide feedback.

Hardware

Both benchmarks were executed on 4 AWS c4.2xlarge instances used as servers and 1 AWS c4.2xlarge instance used as the client and the driver for the benchmark.

Benchmarks

In this benchmark we attempt to compare pessimistic cache transactions only. Both, Ignite and Hazelcast have many other features that you can learn more about on their respective websites.

The benchmarks were run in 2 modes, synchronous backups and asynchronous backups. In case of synchronous backups, the client waited until both, primary and backup copies were updated. In case of asynchronous backups, the client waited only for the primary copies to be updated and the backups were updated asynchronously. This was controlled with configuration properties of both products.

Also, in both benchmarks clients were allowed to read data from backups whenever necessary.

The code used for the benchmark execution is very simple and can be found on GitHub:

Apache Ignite: IgnitePutGetTxBenchmark
Hazelcast: HazelcastPutGetTxPessimisticBenchmark

Apache Ignite:

try (Transaction tx = ignite().transactions().txStart()) {
    Object val = cache.get(key);

    if (val != null)
        key = nextRandom(args.range() / 2, args.range());

    cache.put(key, new SampleValue(key));

    tx.commit();
}

Hazelcast:

TransactionContext tCtx = hazelcast().newTransactionContext(txOpts);

tCtx.beginTransaction();

TransactionalMap<Object, Object> txMap = tCtx.getMap("map");

Object val = txMap.getForUpdate(key);

if (val != null)
    key = nextRandom(args.range() / 2, args.range());

txMap.put(key, new SampleValue(key));

tCtx.commitTransaction();

Result:

Just like with optimistic transactions, we found that in pessimistic mode, Apache Ignite data grid is about 44% faster than Hazelcast. Apache Ignite averaged approximately 16,500 transactions per second, while Hazelcast came in at about 11,000 transactions per second.

Here is a sample graph produced by Yardstick:

Also, when running Hazelcast benchmarks, the following exception kept popping up in the logs, which keeps me wondering about the consistency of the data cached in Hazelcast overall:

SEVERE: [172.30.1.95]:57500 [dev] [3.4.2] Lock is not owned by the transaction! Caller: fa705359-7154-4346-a5f2-292e1a2a75a5, Owner: Owner: fa705359-7154-4346-a5f2-292e1a2a75a5, thread-id: 105
com.hazelcast.transaction.TransactionException: Lock is not owned by the transaction! Caller: fa705359-7154-4346-a5f2-292e1a2a75a5, Owner: Owner: fa705359-7154-4346-a5f2-292e1a2a75a5, thread-id: 105
        at com.hazelcast.map.impl.tx.TxnPrepareBackupOperation.run(TxnPrepareBackupOperation.java:48)
        at com.hazelcast.spi.impl.Backup.run(Backup.java:92)
        at com.hazelcast.spi.impl.BasicOperationService$OperationHandler.handle(BasicOperationService.java:749)
        at com.hazelcast.spi.impl.BasicOperationService$OperationHandler.access$500(BasicOperationService.java:725)
        at com.hazelcast.spi.impl.BasicOperationService$OperationPacketHandler.handle(BasicOperationService.java:699)
        at com.hazelcast.spi.impl.BasicOperationService$OperationPacketHandler.handle(BasicOperationService.java:643)
        at com.hazelcast.spi.impl.BasicOperationService$OperationPacketHandler.access$1500(BasicOperationService.java:630)
        at com.hazelcast.spi.impl.BasicOperationService$BasicDispatcherImpl.dispatch(BasicOperationService.java:582)
        at com.hazelcast.spi.impl.BasicOperationScheduler$OperationThread.process(BasicOperationScheduler.java:466)
        at com.hazelcast.spi.impl.BasicOperationScheduler$OperationThread.doRun(BasicOperationScheduler.java:458)
        at com.hazelcast.spi.impl.BasicOperationScheduler$OperationThread.run(BasicOperationScheduler.java:432)

Benchmarking Data Grids: Apache Ignite vs Hazelcast, Part I

2015-04-14T05:09:00.000-07:00

Recently I have been doing many benchmarks comparing the incubating Apache Ignite^tm project to other products. In this blog I will describe my experience in comparing Apache Ignite Data Grid vs Hazelcast Data Grid.

Yardstick Framework

I will be using Yardstick Framework for the benchmarks, specifically Yardstick-Docker extension. Yardstick is an open source framework for performing distributed benchmarks. One of the best things about Yardstick is that it generates graphs at the end, so we can observe how the benchmark behaved throughout the whole execution.

Transparency

One of the most important characteristics of any benchmark is full transparency. The code for both, Apache Ignite and Hazelcast benchmarks is provided in the corresponding GIT repos:

On startup, Yardstick simply accepts the URL of a GIT repo as a parameter and executes all the benchmarks provided in that repository. This approach makes it really easy to change existing benchmarks or add new ones.

In the interest of full disclosure, I should also mention that I am one of the committers for Apache Ignite project. However, to the best of my ability, I try to stay away from any opinions and simply state the discovered facts here.

Hardware

Both benchmarks were executed on 4 AWS c4.2xlarge instances used as servers and 1 AWS c4.2xlarge instance used as the client and the driver for the benchmark.

Benchmarks

Yardstick S3 functionality automatically adds benchmark results to the specified S3 bucket on Amazon S3 store. Moreover, if you run multiple sets of benchmarks, e.g. Apache Ignite and Hazelcast benchmarks, then Yardstick will automatically generate comparison graphs and store them in S3 bucket as well.

In this benchmark we attempt to compare Data Grid basic cache operations and transactions only. Both, Ignite and Hazelcast have many other features that you can find out on their respective websites.

After some tweaking and tuning, here is what I found about Ignite and Hazelcast:

Both, Apache Ignite and Hazelcast, support distributed data grids (i.e. distributed partitioned caches). In short, they can be viewed as distributed partitioned key-value in-memory stores.
Both, Apache Ignite and Hazelcast, implement JCache (JSR 107) specification
Both are fairly easy to configure and introduce minimal dependencies into the project.
Both have redundancy and failover. In the benchmarks, we configure both products with 1 primary and 1 backup copies for each key stored in cache.
Apache Ignite and Hazelcast have different configuration properties, but it is possible to configure them in the same way for the benchmark.
Both have support for ACID transactions. Ignite allows to set OPTIMISTIC or PESSIMISTIC mode for transactions. Hazelcast also can be coded to work in OPTIMISTIC and PESSIMISTIC modes, even though they don't call it that way explicitly.
The querying capabilities of both products are very different. I will be benchmarking them in the nearest future and will describe them in my next blog.

Basic Atomic Operations

We compared basic puts and puts-and-gets into the cache.

The code used for the benchmark execution can be found on GitHub:

Apache Ignite: IgnitePutBenchmark and IgnitePutGetBenchmark.
Hazelcast: HazelcastPutBenchmark and HazelcastPutGetBenchmark.

Result:
We found that both Ignite and Hazelcast exhibit about the same performance with Ignite being about 4% to 7% faster on most of the runs.

Here are the graphs produced by Yardstick:

Basic Transaction Operations

We compared basic transactional puts and puts-and-gets into the cache in OPTIMISTIC mode.

The code used for the benchmark execution can be found on GitHub:

Apache Ignite: IgnitePutTxBenchmark and IgnitePutGetTxBenchmark.
Hazelcast: HazelcastPutTxBenchmark and HazelcastPutGetTxBenchmark.

Result:
The performance difference for OPTIMISTIC transactions was much bigger, with Ignite transactions outperforming Hazelcast transactions by about 35% to 45%.

Here are the graphs produced by Yardstick:

In my following blogs I will compare the query performance of both products as well and will post my findings.

Streaming and Transforming Data with Apache Ignite

2015-04-08T08:53:00.001-07:00

In its 1.0 release Apache Ignite^tm added much better streaming support with ability to perform various data transformations, as well as query the streamed data using standard SQL queries. Streaming in Ignite is generally used to ingest continuous large volumes of data into Ignite distributed caches (possibly configured with sliding windows). Streamers can also be used to simply preload large amounts of data into caches on startup.

Here is an example of processing a stream of random numbers.

The stream gets partitioned to multiple cluster nodes in such a way that same numbers will always be processed on the same node.
Upon receiving a number, our StreamTransformer will get the current count for that number and increment it by 1.

try (IgniteDataStreamer<Integer, Long> stmr = ignite.dataStreamer("numbers")) {
    // Allow data updates.
    stmr.allowOverwrite(true);

    // Configure data transformation to count random numbers 
    // added to the stream.
    stmr.receiver(StreamTransformer.from((e, arg) -> {
        // Get current count.
        Long val = e.getValue();

        // Increment count by 1.
        e.setValue(val == null ? 1L : val + 1);

        return null;
    }));

    // Stream 10 million of random numbers in the range of 0 to 1000.
    for (int i = 1; i <= 10_000_000; i++) {
        stmr.addData(RAND.nextInt(1000), 1L);

        if (i % 500_000 == 0)
            System.out.println("Number of tuples streamed into Ignite: " + i);
    }
}

As we are streaming the data into the system, we can also query it using standard SQL. In this case, the data type name (in the example below it is "Long") is treated as a table name.

In the query below, we select 10 most popular numbers out of the stream.

// Query top 10 most popular numbers every.
SqlFieldsQuery top10Qry = new SqlFieldsQuery(
    "select _key, _val from Long order by _val desc limit 10");

// Execute query and get the whole result set.
List<List<?>> top10 = stmCache.query(top10Qry).getAll();

Apache Ignite Word Count Streaming Example

2015-04-05T23:05:00.001-07:00

In this example we will stream text into Apache Ignite and count each individual word. We will also issue periodic SQL queries into the stream to query top 10 most popular words.

The example will work as follows:

We will setup up a cache to hold the words as they come from a stream.
We will setup a 1 second sliding window to keep the words only for the last 1 second.
StreamWords program will stream text data into Ignite.
QueryWords program will query top 10 words out of the stream.

Cache Configuration

We define a CacheConfig class which will provide configuration to be used from both programs, StreamWords and QueryWords. The cache will be a partitioned cache which will store words as values. To guarantee that identical words are cached on the same data node, we use AffinityUuid type for unique cache keys.

Note that in this example we use a sliding window of 1 second for our cache. This means that words will disappear from cache after 1 second since they were first entered into cache.

public class CacheConfig {
  public static CacheConfiguration<String, Long> wordCache() {
    CacheConfiguration<String, Long> cfg = new CacheConfiguration<>("words");

    // Index individual words.
    cfg.setIndexedTypes(AffinityUuid.class, /*word type*/String.class);

    // Sliding window of 1 seconds.
    cfg.setExpiryPolicyFactory(FactoryBuilder.factoryOf(
      new CreatedExpiryPolicy(new Duration(SECONDS, 1))));

    return cfg;
  }
}

Stream Words

We define a StreamWords class which will be responsible to continuously read words form a local text file ("alice-in-wonderland.txt" in our case) and stream them into Ignite "words" cache.

Example

public class StreamWords {
  public static void main(String[] args) throws Exception {
    // Mark this cluster member as client.
    Ignition.setClientMode(true);

    try (Ignite ignite = Ignition.start("examples/config/example-ignite.xml")) {
      // The cache is configured with sliding window holding 1 second of the streaming data.
      IgniteCache<AffinityUuid, String> stmCache = ignite.getOrCreateCache(CacheConfig.wordCache());

      try (IgniteDataStreamer<AffinityUuid, String> stmr = ignite.dataStreamer(stmCache.getName())) {
        // Stream words from "alice-in-wonderland" book.
        while (true) {
          InputStream in = StreamWords.class.getResourceAsStream("alice-in-wonderland.txt");

          try (LineNumberReader rdr = new LineNumberReader(new InputStreamReader(in))) {
            for (String line = rdr.readLine(); line != null; line = rdr.readLine()) {
              for (String word : line.split(" "))
                if (!word.isEmpty())
                  // Stream words into Ignite.
                  // By using AffinityUuid as a key, we ensure that identical
                  // words are processed on the same cluster node.
                  stmr.addData(new AffinityUuid(word), word);
            }
          }
        }
      }
    }
  }
}

Query Words

We define a QueryWords class which will periodically query word counts form the cache.

SQL Query

We use standard SQL to query popular words.
Ignite SQL treats Java classes as SQL tables. Since our words are stored as simple String type, the SQL query below queries String table.
Ignite always stores cache keys and values as "_key" and "_val" fields. In our case, "_val" is the word, so we use this syntax in our SQL query.

Example

public class QueryWords {
  public static void main(String[] args) throws Exception {
    // Mark this cluster member as client.
    Ignition.setClientMode(true);

    try (Ignite ignite = Ignition.start()) {
      IgniteCache<String, Long> stmCache = ignite.getOrCreateCache(CacheConfig.wordCache());

      // Select top 10 words.
      SqlFieldsQuery top10Qry = new SqlFieldsQuery(
          "select _val, count(_val) as cnt from String " + 
            "group by _val " + 
            "order by cnt desc " + 
            "limit 10",
          true /*collocated*/
      );

      // Query top 10 popular numbers every 5 seconds.
      while (true) {
        // Execute queries.
        List<List<?>> top10 = stmCache.query(top10Qry).getAll();

        // Print top 10 words.
        ExamplesUtils.printQueryResults(top10);

        Thread.sleep(5000);
      }
    }
  }
}

Starting Server Nodes

In order to run the example, you need to start data nodes. In Ignite, data nodes are called server nodes. You can start as many server nodes as you like, but you should have at least 1 in order to run the example.

Server nodes can be started from command line as follows:

bin/ignite.sh

You can also start server nodes programmatically, like so:

public class ExampleNodeStartup {
    public static void main(String[] args) throws IgniteException {
        Ignition.start();
    }
}

Here is how the output of the QueryWords program looks like on my MacBook Pro laptop (I have started 2 server nodes and one StreamWords program as well)

...
Query results:
(the,2890)
(and,1355)
(to,1298)
(a,1139)
(of,1029)
(said,1002)
(in,912)
(she,820)
(was,766)
(you,711)
Query results:
(the,1679)
(to,830)
(and,810)
(a,680)
(of,629)
(she,491)
(it,357)
(in,330)
(said,315)
(was,274)
...

Asynchronous APIs Done Right

2015-03-04T01:14:00.002-08:00

Ever seen a product which has duplicated mirrored APIs for synchronous and asynchronous processing? I never liked such APIs as they introduce extra noise to what otherwise could be considered a clean design. There is really no point to have myMethod() and myMethodAsync() methods while all you are trying to do is to change the mode of method execution from synchronous to asynchronous.

So how do we approach this problem? How do we make the same API operate synchronously or asynchronously by simply flipping a switch? I believe that Apache Ignite project came up with a very neat and elegant solution to this problem.

Enhance the Java Futures

First of all, we need to enhance standard Java futures. Standard java.util.concurrent.Future allows you to cancel a task and wait for its completion synchronously, but it misses the whole point of asynchronous execution, which is to be notified about the completion of the operation asynchronously. Java 8 actually addresses this problem with CompletableFuture abstraction, however, it does a lot more than that and may be an overkill for most of the programming tasks.

Apache Ignite has an API called IgniteFuture which extends standard java.util.concurrent.Future and adds ability to register asynchronous callbacks and chain callbacks one after another:

public interface IgniteFuture<V> extends Future<V> {
    ...

    /**
     * Registers listener closure to be asynchronously notified whenever future completes.
     *
     * @param lsnr Listener closure to register. If not provided - this method is no-op.
     */
    public void listen(IgniteInClosure<? super IgniteFuture<V>> lsnr);

    /**
     * Make a chained future to convert result of this future (when complete) into a new format.
     * It is guaranteed that done callback will be called only ONCE.
     *
     * @param doneCb Done callback that is applied to this future when it finishes to produce chained future result.
     * @return Chained future that finishes after this future completes and done callback is called.
     */
    public <T> IgniteFuture<T> chain(IgniteClosure<? super IgniteFuture<V>, T> doneCb);
    
    ...
}

Now that we have the truly asynchronous futures, let's see how we can avoid duplicity of synchronous and asynchronous APIs.

IgniteAsyncSupport

By default, all API invocations in Apache Ignite are synchronous. From usability standpoint this makes sense, as most of the time we all utilize synchronous APIs and resort to asynchronous ones only when we really have to.

For whenever we need asynchronous behavior, Apache Ignite has IgniteAsyncSupport interface which is a parent to all the APIs that require both, synchronous and asynchronous mode of operation. In Ignite, such APIs usually have to do with distributed operations and may take longer to comlete, like storing data in distributed caches, or executing a distributed computation.

The main method here is IgniteAsyncSupport.withAsync() which switches any API into asynchronous mode of operation. Whenever asynchronous mode is enabled, the APIs will always store a future for every previous call on per-thread basis. This way, after having invoked an API in asynchronous mode, you can always get the IgniteFuture for that call and listen for the result asynchronously.

Here is an example of synchronous and asynchronous computations on the cluster.

Synchronous Compute:

// IgniteCompute has synchronous and asynchronous modes.
IgniteCompute compute = ignite.compute();

// Execute a job synchronously and wait for the result.
String res = compute.call(() -> "Hello world");

Asynchronous Compute:

// Enable asynchronous mode (note that the same IgniteCompute API is used).
IgniteCompute asyncCompute = ignite.compute().withAsync();

// Asynchronously execute a job.
asyncCompute.call(() -> "Hello world");

// Get the future for the above invocation.
IgniteFuture<String> fut = asyncCompute.future();

// Asynchronously listen for completion and print out the result.
fut.listen(f -> System.out.println("Job result: " + f.get()));

Here is an example of a how asynchronous mode is enabled for distributed caches.

Synchronous Cache:

IgniteCache<String, Integer> cache = ignite.jcache("mycache");

// Synchronously store value in cache and get previous value.
Integer val = cache.getAndPut("1", 1);

Asynchronous Cache:

// Enable asynchronous mode (note that the same IgniteCache API is used).
IgniteCache<String, Integer> asyncCache = ignite.jcache("mycache").withAsync();

// Asynchronously store value in cache.
asyncCache.getAndPut("1", 1);

// Get future for the above invocation.
IgniteFuture<Integer> fut = asyncCache.future();

// Asynchronously listen for the operation to complete.
fut.listenAsync(f -> System.out.println("Previous cache value: " + f.get()));

See Ignite documentation for more information about Ignite Asynchronous Mode.

Apache Ignite is a distributed In-Memory Data Fabric which allows to distribute and cache data in memory, perform distributed computations, streaming, etc... Since most of the supported functionality in Ignite is distributed, having properly implemented asynchronous mode of operation becomes very critical.

Virtual Cluster Groups with Apache Ignite

2015-03-01T12:55:00.000-08:00

An easy-to-manage network cluster is a cluster in which all nodes are equal and can be brought up with identical configuration. However, even though all nodes are equal, it often still makes sense to assign application-specific roles to them, like "workers', "clients", or "data-nodes". In Apache Ignite, this concept is called cluster groups.

Apache Ignite is an In-Memory Data Fabric composed of multiple distributed components with Clustering APIs serving as the main backbone for the rest of the components, including Data Grid, Compute Grid, and Service Grid. I am one of the committers to this project and generally blog quite a bit about it.

You can create virtual cluster groups in Ignite based on any application-specific custom filter. However, to make things easier, Ignite comes with some predefined filters.

Select Remote Nodes

Here is how you can execute a simple closure on all remote nodes. Remote nodes include all cluster members, except for the member who is starting the execution.

final Ignite ignite = Ignition.ignite();

IgniteCluster cluster = ignite.cluster();

// Get compute instance which will only execute
// over remote nodes, i.e. not this node.
IgniteCompute compute = ignite.compute(cluster.forRemotes());

// Broadcast to all remote nodes and print the ID of the node 
// on which this closure is executing.
compute.broadcast(() -> System.out.println("Hello Node: " + cluster.localNode().id());

Select Worker Nodes

You can assign application specific roles to cluster members, like "masters" and "workers", for example. This can be done via user attributes specified on node startup. For example, here is how you can bring up a cluster node with "ROLE" attribute set to "worker":

IgniteConfiguration cfg = new IgniteConfiguration();

Map<String,String> attrs = Collections.singletonMap("ROLE", "worker");

cfg.setUserAttributes(attrs);

// Start Ignite node.
Ignite ignite = Ignition.start(cfg);

Then here is how you would execute a closure only over nodes with role "worker":

IgniteCluster cluster = ignite.cluster();

// Get compute instance which will only execute over "worker" nodes.
IgniteCompute compute = ignite.compute(cluster.forAttribute("ROLE", "worker"));

// Broadcast to all "worker" nodes and print the ID of the node 
// on which this closure is executing.
compute.broadcast(() -> System.out.println("Hello Node: " + cluster.localNode().id());

Custom Cluster Groups

And finally, you can create custom cluster groups based on any user-defined predicates. Such cluster groups will always only include the nodes that pass the predicate. For example, here is how we would create a group of cluster nodes that have CPU utilization less than 50%:

// Nodes with less than 50% CPU load.
ClusterGroup readyNodes = cluster.forPredicate((node) -> node.metrics().getCurrentCpuLoad() < 0.5);

// Broadcast to all nodes with CPU load less than 50% and
// print the ID of the node on which this closure is executing.
compute.broadcast(() -> System.out.println("Hello Node: " + cluster.localNode().id());

For more on cluster groups, visit Ignite Cluster Groups documentation.

Emergence of the In-Memory Data Fabric

2015-02-26T00:00:00.000-08:00

Some of us may have already heard the terms Data Grid and Data Fabric, however, neither of these terms has been well defined in the industry. In this blog, I will try to add some clarity to both terms by outlining some main features for data grids and data fabrics.

What is a Data Grid

Often when doing meetup presentations about Apache Ignite, I ask the crowd if anyone has ever heard of what a Data Grid is. I usually get only a few hands. However, when I flip the question and ask what Distributed Caching is, everyone in the room immediately raises their hands and nods in understanding. The reality is that a Data Grid can be viewed as a Distributed Cache with extra features, so if you do know what a Distributed Cache is, you probably already know a lot about Data Grids as well.

Generally, the term distributed cache means ability to replicate data in memory, so it is accessible from anywhere in the cluster. Data Grids usually accomplish this by partitioning data in memory, where each cluster member is responsible only for its own subset of the data. You can also think of it as a distributed Hash Table. This way, the more servers are available in your cluster, the more data you can cache.

Data grids are generally known for having a fairly rich feature set on top of in-memory caches. The 3 main features that are absolutely mandatory for any data grid solution are:

distributed transactions
distributed queries
collocation of compute and data

Without the above 3 features, you cannot really call a product a data grid. Many vendors also differentiate between each other by adding other popular features, including:

SQL support
Off-Heap Memory (to avoid lengthy GC pauses)
WebSession Caching
Hibernate Integration
Database Integration

Some of the popular Data Grid providers include Apache Ignite (incubating), Hazelcast and Infinispan in the open source space, and Oracle Coherence and GridGain commercial offerings. GridGain is a commercial offering of the Apache Ignite.

What is an In-Memory Data Fabric

In Memory Data Fabrics represent the natural evolution of in-memory computing. Data Fabrics generally take a broader approach to in memory computing, grouping the whole set of in memory computing use cases into a collection of well-defined independent components. Usually a Data Grid is just one of the components provided by a Data Fabric. Additionally to the data grid functionality, an In-Memory Data Fabric typically also includes a Compute Grid, CEP Streaming, an In-Memory File System, and more.

The main advantage of an In-Memory Data Fabric is that all of the provided in-memory computing components can be used independently, while being well integrated with each other. For example, in Apache Ignite a Compute Grid knows how to load-balance and schedule computations within a cluster, but when used together with a Data Grid, the Compute Grid will also route all the computations that process data to the cluster members responsible for caching that data. The same goes for Streaming and CEP - when working with streamed data, all the processing happens on the cluster members responsible for caching that data as well.

Commonly seen features of In-Memory Data Fabrics include:

Data Grid (must have for any Data Fabric)
Compute Grid
Service Grid
Streaming & CEP
Distributed File System
In-Memory Database

Apache Ignite, an Apache Incubator project, is the only In-Memory Data Fabric available in the Open Source space. GridGain provides a commercial, enterprise edition of Apache Ignite that is targeted toward production, business critical use cases.

Apache Ignite In-Memory Data Fabric v1.0 is Born

2015-02-17T15:49:00.002-08:00

Today, as part of the community of Apache Ignite (incubating), I am proud to announce that we have made the first code drop of the Apache Ignite In-Memory Data Fabric – Apache Ignite v1.0 Release Candidate - available.

The Apache Ignite project started in September 2014, when the Open Source edition of the GridGain In-Memory Data Fabric was donated to the Apache Software Foundation and branded as Apache Ignite. Now, after 5 months of many dev-list discussions and late nights, we finally have released a stable, well-tested release candidate of Apache Ignite 1.0.

Going forward, the role of GridGain engineers will be to continue to actively contribute to the Ignite code base, but also to provide a hardened, enterprise-grade feature set on top of Apache Ignite. GridGain support will be available for both Apache Ignite and the GridGain In-Memory Data Fabric Enterprise Edition.

What is an In-Memory Data Fabric

So why should you care about Apache Ignite? First and foremost, Ignite is lightning fast and has virtually unlimited scale. Ignite is based on the former GridGain In-Memory Data Fabric Open Source edition, the leading open source in-memory data fabric, which has several known 1000+ node deployments.

Apache Ignite has a very rich feature set. From the get-go, our main goal was to make Apache Ignite an all-in-one stop for everything you need for in-memory computing.

Some of the main features of the project include:

Advanced Clustering
Distributed Caching (JCache)
Data Grid
Compute Grid
Service Grid
Streaming & CEP
Distributed File System - IgniteFS
Hadoop Accelerator
Distributed Data Structures
Distributed Messaging
Distributed Events

But one of the coolest, new features in Ignite is its ability to automatically integrate with different RDBMS systems, such as Oracle, MySql, Postgres, DB2, Microsoft SQL, etc. This feature automatically generates the application domain model based on the schema definition of the underlying database, and then loads the data in memory.

While you may be able to get some subset of the above functionality from other individual point solutions, the main benefit you get from the Apache Ignite In-Memory Data Fabric is the integration of all these components. For example, Ignite will automatically route your computations that need to process data to the cluster nodes responsible for caching this data. The same goes for the processing of streaming data as well. This approach is called "collocation of compute and data" and when applied, significantly reduces network traffic and increases scalability and performance.

Here is an example of how you would broadcast a computation in Ignite:

compute.broadcast(() -> System.out.println("Hello World");

Here is cache example of how to perform an in-memory distributed transaction in Ignite:

try (Transaction tx = ignite.transactions().txStart()) {
    Integer hello = cache.get("Hello");
  
    if (hello == 1)
        cache.put("Hello", 11);
  
    cache.put("World", 22);
  
    tx.commit();
}

Ease Of Use

Despite the breadth of its feature set, Apache Ignite is easy to deploy and use.

Installation

The product does not have any custom installers. It comes as one ZIP file, which is ready to go once you unzip it. To startup multiple cluster nodes, simply execute "bin/ignite.sh" script multiple times.

Dependencies

The project has 1 main dependency - ignite-core.jar. All other dependencies, like integration with Spring for configuration, or with H2 database for SQL, etc. can be added to the process a la carte by dragging corresponding folders form "libs/optional" folder into "libs/" folder.

Maven

The project is fully mavenized, and is composed of over a dozen of maven artifacts that can be imported and used in any combination.

Standard APIs

Ignite is based on standard Java APIs. For distributed caches and data grids, Ignite in its final 1.0 release will implement the JCache (JSR107) standard. For distributed computations, you can utilize standard ExecutorService API. There are also distributed Queues and Sets. Ignite also has implemented most of the data structures from the java.util.concurrent package by distributing them in memory. IgniteFS – the distributed file system provided by Ignite implements the standard Hadoop FileSystem API and can be automatically plugged into any Hadoop installation.

Code Quality

Apache Ignite is very stable and well tested. The development process is structured in such a way that before any merge to the main branch happens, over 10,000 tests are executed on top of JetBrains Team City, and all need to pass.

Moreover, Ignite has its own QA team. All the main functionality undergoes scrupulous testing for every release. Also, every release is benchmarked against a previous release to ensure that it is at least as fast (or faster) and as stable as the previous release.

Also, the project inherited several years of thorough testing and stability-tuning from the GridGain In-Memory Data Fabric Open Source edition, which boasts over a thousand production installations.

Community

Even though the project has been in Apache for less than 4 months, it already has a vibrant and growing community. The project currently has 11 committers and about as many contributors, all of whom are very active. Some have joined the project just recently, but have already been actively contributing.

We always welcome community contributions. If you would like to contribute, send an email to the Ignite dev list and we will get you started. And even if you are not ready to contribute immediately, I would like to invite everyone to join our dev list. Most of the discussions happen there, and you can find out a lot about where the project is going and also provide your own ideas.

Another way, of course, to familiarize yourself with Apache Ignite, is to take a look at the code and see what it can do for your project.

You can download the Ignite bits on the Apache Ignite homepage.

Apache®, Apache Ignite, Ignite®, and the Apache Ignite logo are either registered trademarks or trademarks of the Apache Software Foundation in the United States and/or other countries.

Chrontrack Technology, Powered by GridGain, Tracks NYC Marathon Runners

2014-10-31T12:39:00.000-07:00

ChronoTrack is an industry trusted provider of race solutions for race organizers and timing partners. Its hardware and software solutions paired with certified network of partners provide the most comprehensive set of timing, race management and live race services available.

This weekend Chronotrack technology will be used to track NYC marathon runners. Essentially, every runner gets a chip embedded into a tag, and when they pass through the censors installed along the marathon track, their timings and other data are immediately sent back to the servers in real time.

So why is this interesting? It turns out that when it comes to real-time race analytics to track hundreds of thousands of runners across the globe, standard disk-based processing simply does not scale. In order to achieve real-time performance, Chronotrack had to move to in-memory processing, and now uses GridGain In-Memory Data Fabric to provide marathon data in real time, being able to serve all sorts of race analytics in a matter of a few milliseconds.

Bloomberg TV

Watch GridGain CEO, Abe Kleinfeld, talk about Chronotrack technology, powered by GridGain, on Bloomberg TV:

http://www.bloomberg.com/video/nyc-marathon-and-the-tech-used-to-track-the-runners-lpdAk0h4S2WjM94onxk~uw.html

One-Phase-Commit - Fast Transactions For In-Memory Caches

2014-09-16T16:47:00.001-07:00

In my previous blogs I have talked at length about 2-Phase-Commit transaction protocol for in memory caches, and how in-memory caches can handle failures a lot more efficiently than disk-based databases. In this blog I want to cover yet another very important optimization that can be utilized for in-memory caches, specifically for cases where data is partitioned across the network.

In-memory caches, and specifically in-memory data grids, such as GridGain or Oracle Coherence, often employ a technique called data partitioning, where every key in the key set is assigned to a partition, and every partition is assigned to a specific cluster member. Assigning keys to a partition is usually easy and is done similarly to how hash maps work:

key.hashCode() % N, where N is a total number of partitions.

Assigning a partition to a cluster node is a little trickier, as in case of failures or cluster topology changes, the amount of repartitioning has to be minimal. There are various algorithms that can be employed here, such as Rendezvous Hashing, or Consistent Hashing, which we will not be discussing here. Let's assume that after applying some of the partition strategy algorithms, your in-memory cache evenly distributed data among cluster nodes:

Figure 1: Cache Partition Distribution

Usually to achieve best performance we need to minimize the number of cluster nodes participating in a transaction. This can be done by ensuring that all the entries in the transaction belong to the same partition, which consecutively ensures that they all belong to the same node. It will also ensure that the backup copies for these entries will be grouped together on some other node as well (and secondary backups will be grouped together as well, and so on). Such custom key mapping is called custom affinity mapping.

For example, if we have Employee objects and Company objects, then we can ensure that all employees working for the same company will be mapped to the same partition by providing a custom affinity-key for Employees, which in this case will be the "companyId".

Custom affinity mapping helps us ensure that all objects within a single transaction are mapped to the same cluster node, thus collocating the whole transaction logic on that node and minimizing the number of nodes involved in a transaction.

The 1-Phase-Commit optimization is possible only when we have 1 primary and 1 backup copy. Such deployments are most common for distributed caches. If we add 2 backup copies, then we have to resort to 2-Phase-Commit, however, adding a 2nd backup copy is generally considered wasteful from memory standpoint and is rarely done. Diagram below illustrates the 1-Phase optimization:

Figure 2: 1-Phase-Commit for Collocated Partition Transaction

The first deviation from standard transactions is that now the client node sends the whole transaction logic to the primary node. This is possible because we ensure in advance that all the keys we are transacting on are mapped to the same partition on that node. Once the primary node receives the transaction logic, it will acquire all the locks locally, and will send only one commit message (without the prepare message) to the backup node.

Now let's analyze failure scenarios. In this case, failures of client nodes or backup nodes are not very interesting, as they do not affect the primary copies. Failures of primary nodes are a bit trickier, however they are still safe. If the primary node crashes before it sends the commit message, then the backup transaction never starts, so there are no side effects. If the primary node fails after or during the commit acknowledgement is received from the backup node, then the backup transaction is committed, and data consistency is again not violated.

The hardest part is that even though the data remains consistent in case of any cluster failures, how does the client node know whether the transaction was committed or not if it failed to get the final acknowledgement, i.e. if the primary node failed before it was able to send the acknowledgement to the client? In this case the "recovery protocol" is initiated, which was described in detail in my previous blog. Essentially, a message is sent to the backup node asking whether the transaction was committed or not. Since the backup node keeps a backlog of completed transaction IDs in memory, it can always check the backlog. If the backlog does not have the given transaction ID, the backup node will add it in the "rolled back" state and will reply to the client with rollback acknowledgement. If afterwards the backup node actually does receive the commit request for the same transaction ID from the now failed primary node, it can verify in the backlog that it was rolled back and safely ignore it.

Conclusion

By ensuring that all objects participating in a transaction are mapped to the same logical partition, we can remove the whole "prepare" phase from the distributed commit protocol, thus converting the standard 2-Phase-Commit into very light weight 1-Phase-Commit transactions.

Two-Phase-Commit for In-Memory Caches - Part II

2014-09-09T00:51:00.002-07:00

Generally, persistent disk-oriented systems will require the additional 3rd phase in commit protocol in order to ensure data consistency in case of failures. In my previous blog I covered why the 2-Phase-Commit protocol (without 3rd phase) is sufficient to handle failures for distributed in-memory caches. The explanation was based on the open source GridGain architecture, however it can be applied to any in-memory distributed system.

In this blog we will cover a case when an in-memory cache serves as a layer on top of a persistent database. In this case the database serves as a primary system of records, and distributed in-memory cache is added for performance and scalability reasons to accelerate reads and (sometimes) writes to the data. Cache must be kept consistent with database which means that a cache transaction must merge with the database transaction.

When we add a persistent store to an in-memory cache, our primary goal is to make sure that the cache will remain consistent with on-disk database at all times.

In order to keep the data consistent between memory and database, data is automatically loaded on demand whenever a read happens and the data cannot be found in cache. This behavior is called read-through. Alternatively, whenever a write operation happens, data is stored in cache and is automatically persisted to the database. This behavior is called write-through. Additionally, there is also a mode called write-behind which batches up the writes in memory and flushes them to the database in one bulk operation (we will not be covering this mode here).

Figure 1: Read-through operation for K2

When we add a persistent store to the 2-Phase-Commit protocol, in order to merge cache and database transactions into one, the coordinator will have to write the transactional changes to the database before it sends the Commit message to the other participants. This way, if database transaction fails, the coordinator can still send the Rollback message to everyone involved, so that the cached data will remain consistent with database. Figure below illustrates this behavior.

Figure 2: Two-Phase-Commit with In-Memory-Cache and Database

Handling failures is actually more straight forward whenever a database is present rather than when it is not. We always assume that the database must have the utmost up-to-date copy, and it is acceptable to reload data from the database into cache whenever in doubt (see Figure 1). Just like in my previous blog, the most challenging scenario here is when the coordinator node crashes (potentially together with other nodes), because in this case we cannot tell whether it crashed before it was able to commit to the database or not. Other failure scenarios are handled the same way with database present as without.

Whenever we cannot tell whether the database commit had happened or not, we can simply reload the relevant data from database into cache upon committing the transaction. This effectively ensures that database and cache always remain in consistent state.

Conclusion

When working with in-memory caches, we can always manage to keep the data within transactions consistent by slightly enhancing the standard 2-Phase-Commit protocol. The main advantage of in-memory vs. disk is that failure handling does not introduce any additional overhead, and we do not need to add an expensive 3rd phase to the 2-Phase-Commit protocol in order to keep caches consistent with databases in case of failures.

Two-Phase-Commit for Distributed In-Memory Caches

2014-09-02T16:07:00.001-07:00

2-Phase-Commit is probably one of the oldest consensus protocols and is known for its deficiencies when it comes to handling failures, as it may indefinitely block the servers waiting in prepare state. To mitigate this, a 3-Phase-Commit protocol was introduced which adds better fault tolerance at the expense of extra network round-trip message and higher latencies.

I would like to extend these traditional quorum concepts into distributed in-memory caching, as it is particularly relevant to what we do at GridGain. GridGain too has 2-phase-commit transactions, but unlike disk-based persistent systems, GridGain does not need to add a 3rd phase to the commit protocol in order to preserve data consistency during failures.

We want to avoid the 3-Phase-Commit protocol because it adds an additional network round-trip and has a negative impact on latencies and performance.

In GridGain, the data is partitioned in memory, which in the purest form means that every key-value pair is assigned to a specific node within the cluster. If for example, we have 100 keys and 4 nodes, then every node will cache 100 / 4 = 25 keys. This way the more nodes we have, the more data we can cache. Of course, in real life scenarios, we also have to worry about failures and redundancy, so for every key-value pair we will have 1 primary copy, and 1 or more backup copies.

Let us now look at the 2-Phase-Commit protocol in more detail to see why we can avoid the 3rd commit phase in GridGain.

Two-Phase-Commit Protocol

Generally the 2-Phase-Commit protocol is initiated whenever an application has already made a decision to commit the transaction. The Coordinator node sends a Prepare message to all the participating nodes holding primary copies (primary nodes), and the primary nodes, after acquiring the necessary locks, synchronously send the "Prepare" message to the nodes holding backup copies (backup nodes). Once every node votes "Yes", then the "Commit" message is sent and the transaction gets committed.

Figure 1: 2-Phase-Commit Protocol For Distributed In-Memory Cache

Advantages of In-Memory

So, why does in-memory-only processing allow us to avoid the 3rd phase as opposed to persistent disk-based transactions? The main reason is that the in-memory cache is purely volatile and does not need to worry about leaving any state behind in case of a crash. Contrast that to the persistent architectures, where we need to worry whether the state on disk was committed or not after a failure had occurred.

Another advantage of in-memory-only distributed cache is that the only valid vote for the "Prepare" message is "Yes". There is really no valid reason for any cluster member to vote "No". This essentially means that the only reason a rollback should happen is if the Coordinator node crashed before it was able to send the "Prepare" message to all the participating nodes.

Now, that we have the ground rules settled, let's analyze what happens in case of failures:

Backup Node Failures

If a backup node fails during either "Prepare" phase or "Commit" phase, then no special handling is needed. The data will still be committed on the nodes that are alive. GridGain will then, in the background, designate a new backup node and the data will be copied there outside of the transaction scope.

Primary Node Failures

If a primary node fails before or during the "Prepare" phase, then the coordinator will designate one of the backup nodes to become primary and retry the "Prepare" phase. If the failure happens before or during the "Commit" phase, then the backup nodes will detect the crash and send a message to the Coordinator node to find out whether to commit or rollback. The transaction still completes and the data within distributed cache remains consistent.

Coordinator Node Failures

Failure of the Coordinator node becomes a little tricky. Without the Coordinator node, neither primary nodes, nor backup nodes know whether to commit the transaction or to rollback. In this case the "Recovery Protocol" initiates, and all the nodes participating in the transaction send a message to every other participating node asking whether the "Prepare" message was received. If at least one of the nodes replied "No", then the transaction will be rolled back, otherwise the transaction will be committed.

Note that after all the nodes received the "Prepare" message, we can safely commit, since voting "No" is impossible during the "Prepare" phase as stated above.

It is possible that some nodes will have already committed the transaction by the time they have received the Recovery Protocol message from other nodes. For such cases, every node keeps a backlog of completed transaction IDs for a certain period of time. If there is no ongoing transaction with given ID found, then the backlog is checked. If the transaction was not found in the backlog, then it was never started, which means that the failure had occurred before the Prepare phase was completed and it is safe to rollback.

Since the data is volatile, and does not leave any state after the crash, it does not really matter if any of the primary or backup nodes also crashed together with the coordinator node. The Recovery Protocol still works the same.

Figure 2: Recovery Protocol

Conclusion

Note that we are able to fully recover the transaction without introducing the 3rd commit phase mainly because the data in distributed in-memory caches is volatile and node crashes do not leave any state behind.

The important advantage of the 2-Phase-Commit Recovery Protocol in GridGain over the 3-Phase-Commit is that, in the absence of failures, the recovery protocol does not introduce any additional overhead, while the 3-phase-commit adds an additional synchronous network roundtrip to the transaction and, therefore, has negative impact on performance and latencies.

GridGain also supports modes where it works with various persistent stores including transactional databases. In my next blog, I will cover the scenario where a cache sits on top of a transactional persistent store, and how the 2-Phase-Commit protocol works in that case.

GridGain 6.2.0 Released with Portable Objects and Distributed Services

2014-08-27T15:54:00.001-07:00

I am pleased to announce the release of GridGain Open Source In-Memory Computing Platform 6.2.0. The main components of the platform are: compute grid, data grid (or in-memory distributed cache), and CEP streaming. This release revolves primarily around Portable Object functionality as well as Distributed (or Guaranteed) Services.

Portable Objects

One of the main benefits of Portable Objects, which was not available in our prior release, is the ability to access any data or field without having to deserialize the entire object. This brings significant performance improvements to the server-side processing of in-memory cached data, as filtering remote caches and executing SQL or predicate-based queries becomes very inexpensive and fast. As a matter of fact, you don’t even need to have class definitions on the server side, which essentially allows for dynamic structural changes of cached data without the need to restart the cluster.

Here is an example of how a single field can be accessed from a portable object without having to deserialize it into a concrete class representation:

GridCacheProjection<Integer, GridPortableObject> prj = cache.keepPortable();

int myKey = 123;

GridPortableObject val = prj.get(myKey);

String field = val.field("myFieldName");

In addition, portable objects also include the ability to perform the following:

Make any object portable with zero code change to your existing code.
Nest portable objects within each other.
Automatically handle circular or null references.
Optionally avoid deserialization of objects on the server side (objects are stored in GridPortableObject format).
Avoid the need to have concrete class definitions on the server side.
Dynamically change the structure of the classes without having to restart the cluster.
Index into portable objects for querying purposes.

Distributed Services

Distributed Services allow for controlled deployment of any service on the grid. You can think of a distributed service as a fault-tolerant thread pool, for which you can configure how many threads are allowed to be deployed on each grid node.

For example, what if you need to deploy a web server within your cluster, but you are only allowed to have one instance of it deployed? If the node on which your web server is deployed fails, then you want it to automatically start up on another server. Distributed Services allow you to have such functionality, in this case by deploying a cluster-wide singleton service which will start your web server.

With distributed services you can do the following:

Automatically deploy any number of service instances in the grid.
Automatically deploy singletons, including cluster-singleton, node-singleton, or key-affinity-singleton.
Automatically deploy services on node start-up by specifying them in grid configuration.
Undeploy any of the deployed services.
Get information about service deployment topology within the grid.

Here is an example of how a simple cluster singleton service can be deployed:

// Simple service implementation.
public class MyGridService implements GridService {
    @Override public void cancel(GridServiceContext ctx) {
        // No-op.
    }

    @Override public void execute(GridServiceContext ctx) {
        // Loop until service is cancelled.
        while (!ctx.isCancelled()) {
            // Do something.
            ...
        }
    }
}
...
GridServices svcs = grid.services();

GridFuture<?> fut = svcs.deployClusterSingleton("mySingleton", new MyGridService());

// Wait for deployment to complete.
fut.get();

Click here to download GridGain 6.2.0.

GridGain 6.1.9 - One Click Installation - One Jar Dependency

2014-07-03T19:24:00.000-07:00

For those who are not familiar with GridGain, it is an open source distributed data grid product mainly focusing on distributed in-memory caching, distributed computations, and streaming. It generally addresses all sorts of performance and scalability challenges for business applications by bringing and caching data in memory and allowing to compute on it.

With release 6.1.9, GridGain significantly simplified its installation and deployment. GridGain now allows for:

One Click Installation:
The product simply has to be downloaded and unzipped. After that it is ready to be used.
One Jar Dependency:
GridGain now has only one mandatory dependency - gridgain-6.1.9.jar. All other jars are optional.

Essentially, you still have support for the same set of features, like Spring configuration, or Hibernate L2 cache, but you don't have to have dependency on Spring or Hibernate, or anything else, unless you start using it. To add a dependency to GridGain, simply drag and drop a corresponding folder from "libs/optional" to "libs/" folder. For example, to add log4j dependency, drag "libs/optional/gridgain-log4j" folder into "libs" folder.

Here is how GridGain 6.1.9 structure looks with only one mandatory gridgain-6.1.9.jar file. Note that in the example below, gridgain-indexing and gridgain-spring modules are located in the "libs/" folder, and therefore are enabled.

You can download GridGain 6.1.9 Open Source edition here: www.gridgain.org/download/.

G1 Garbage Collector Thrashes On Larger Heaps

2014-06-16T18:12:00.000-07:00

Since the G1 (garbage-first) garbage collector has been released, there were expectations that it would finally perform better for larger heap sizes (>16GB). Unfortunately those expectations were not met. While G1 garbage collector is meant to remove larger GC pauses, the sporadic and unpredictable behavior of G1 collector on larger heaps renders it generally unusable for any system sensitive to performance SLAs.

At GridGain, having worked on a distributed caching (data grid) product for many years, we constantly benchmark with various Garbage Collectors to find the optimal configuration for larger heap sizes. From conducting numerous tests, we have concluded that unless you are utilizing some off-heap technology (e.g. GridGain OffHeap), no Garbage Collector provided with JDK will render any kind of stable GC performance with heap sizes larger that 16GB. For example, on 50GB heaps we can often encounter up to 5 minute GC pauses, with average pauses of 2 to 4 seconds.

Here is a good comparison of a simple test for a cache 'put' benchmark on 50GB of memory. The blue graph was run with G1 garbage collector set to 200ms average GC pauses, and the green one was run with GridGain OffHeap Memory technology. In each test we are storing over 120,000,000 objects in cache from multiple threads under load on a cluster of 2 servers:

As you see, using GridGain OffHeap memory renders a fairly smooth and predictable green graph.

Here are some of the GC printouts we received from G1 garbage collector:

474.404: [GC pause (young) 13278M->9596M(20480M), 0.7638660 secs]
481.850: [GC pause (young) (initial-mark) 13356M->9674M(20480M), 0.7320680 secs]
482.583: [GC concurrent-root-region-scan-start]
482.784: [GC concurrent-root-region-scan-end, 0.2017740]
482.784: [GC concurrent-mark-start]
489.055: [GC pause (young) 13442M->9752M(20480M), 0.7715580 secs]
495.648: [GC concurrent-mark-end, 12.8631950 sec]
495.675: [GC remark, 0.0303560 secs]
495.715: [GC cleanup 13232M->13232M(20480M), 0.2076440 secs]
496.305: [GC pause (young) 13520M->9830M(20480M), 0.7453250 secs]
505.025: [GC pause (mixed) 13598M->9363M(20480M), 2.7538670 secs]
516.698: [GC pause (mixed) 13131M->9241M(26624M), 2.9020230 secs]
...
1160.396: [GC pause (young) 24923M->15508M(51200M), 3.2503740 secs]
1177.879: [GC pause (young) 25052M->15637M(51200M), 3.4448020 secs]
1202.492: [GC pause (young) 25189M->15765M(51200M), 3.1212860 secs]
1220.057: [GC pause (young) 25325M->15891M(51200M), 4.0836240 secs]
1245.212: [GC pause (young) 25459M->16016M(51200M), 3.6344050 secs]
1262.500: [GC pause (young) 25584M->16141M(51200M), 4.0109490 secs]

The JVM Parameters we ran G1 garbage collector with were:

-Xms50g -Xmx50g -XX:+UseG1GC -XX:MaxGCPauseMillis=200

G1 garbage collector had average GC pauses between 2 and 4 seconds, with major mixed or concurrent sweeps happening for 6 to 12 seconds in the middle. From the graphs you can easily see that latencies and throughput (operations / second) for G1 garbage collector are all over the place, and often drop to the bottom, rendering the system completely unresponsive multiple times during the 20 minute test run.

The benchmark was run on 2 Dell R610 blades, with 8 cores, and 96GB of RAM each. The benchmarking framework used, which helped with generation of pretty graphs, was the open source Yardstick framework, available on GitHub.

Distributed Caching In 5 Minutes

2014-06-08T13:23:00.001-07:00

If you prefer a video demo with coding examples, skip to the screencast at the bottom of this blog.

Distributed In-Memory Caching generally allows you to replicate or partition your data in memory across your cluster. Memory provides a much faster access to the data, and by utilizing multiple cluster nodes the performance and scalability of the application increases significantly.

Majority of the products that do distributed caching call themselves In-Memory Data Grids. On top of simply providing hash-table-like access to your data, a data grid product should provide some combination of the following features:

Clustering
Distributed Messaging
Distributed Event Notifications
Distributed ACID Transactions
Distributed Locks
Distributed Data Queries, possibly using SQL
Distributed Data Structures, like Maps, Queues, Sets, etc.
Clustered Web Sessions
OR-Mapping Integration, including Hibernate
Persistent Database Support, like Oracle, MySQL, etc.

Of course the devil is in the details. For example, given the distributed nature of the cluster anything can fail at any point. So a good question to ask is how the failures are handled, especially what if the failures happen during commit. If during commit a cluster can be left in semi-committed state due to failures, it is definitely a problem.

Another example would be queries. Are the predicate queries being supported? Can you do SQL queries, particularly can the SQL Joins be handled? How are the aggregate functions handled, etc.

Simplicity of APIs is very important as well. ConcurrentMap API has become a de facto standard of accessing data stored in distributed caches, but not all the products support it. Also, a good thing to check would be whether other standard data structures are supported. For example, GridGain supports Map, Set, BlockingQueue, AtomicLong, AtomicSequence, CountDownLatch, all in distributed fashion.

And the last, but not least, always check for performance. Load up the cluster and see what the throughput and latencies are, what is the network load on each server, etc. A good benchmarking tool for testing distributed systems is open source Yardstick Framework, available on GitHub.

Coding Example

Here is a GridGain Data Grid coding example of some basic operations on distributed caches:

private static void atomicMapOperations() throws GridException {
    GridCache<Integer, String> cache = GridGain.grid().cache(CACHE_NAME);

    // Put and return previous value.
    String v = cache.put(1, "1");
    assert v == null;

    // Put and do not return previous value 
    // (all methods ending with 'x' return boolean).
    // Performs better when previous value is not needed.
    cache.putx(2, "2");

    // Put asynchronously (every cache operation has async counterpart).
    GridFuture<String> fut = cache.putAsync(3, "3");

    // Put-if-absent.
    boolean b1 = cache.putxIfAbsent(4, "4");
    boolean b2 = cache.putxIfAbsent(4, "44");


    // Put-with-predicate, will succeed if predicate evaluates to true.
    cache.putx(5, "5");
    cache.putx(5, "55", new GridPredicate<GridCacheEntry<Integer, String>>() {
        @Override public boolean apply(GridCacheEntry<Integer, String> e) {
            return "5".equals(e.peek()); // Update only if previous value is "5".
        }
    });

    // Transform - assign new value based on previous value.
    cache.putx(6, "6");
    cache.transform(6, new GridClosure<String, String>() {
        @Override public String apply(String v) {
            return v + "6"; // Set new value based on previous value.
        }
    });

    // Replace.
    cache.putx(7, "7");
    b1 = cache.replace(7, "7", "77");
    b2 = cache.replace(7, "7", "777");
}

Screencast

Here is a brief screencast showing how to get started with basic operations on your cache in under 5 minutes:

Getting Started With Clustering

2014-05-14T12:08:00.000-07:00

If you don't like to read and prefer video demos, you can skip directly to the Screencast at the bottom of this post.

What do Clustering frameworks really do? More often than not clustering frameworks will provide capability to auto-discover servers on the network, share resources, and schedule tasks. Some will also add distributed messaging and distributed event notification capabilities.

While there are some well known clustering frameworks, like Zookeeper or Mesos, they usually provide very rudimentary clustering capabilities. However, often on top of basic clustering, you also need to perform MapReduce computations, distribute closures, or distribute data. For cases like these, Compute Grids (a.k.a. High Performance Computing Grids) or Data Grids become very useful.

For those not familiar with term "Data Grid", it is simply a Distributed Cache with more advanced features, like distributed data querying, transactions, etc...

Compute Grids or Data Grids often provide very advanced clustering APIs which are very simple to use. Here I will show some basic examples on top of GridGain In-Memory Data Grid, which is Open Source and licensed under Apache license.

GridGain clustering supports auto-node discovery, but at the same time adds capabilities to create any virtual sub-groups of grid nodes within cluster and exchange messages between them or get remote event notifications. While I have blogged about it in more detail before, here is a pretty simple example which demonstrates auto-discovery and distributed computations on the cluster:

try (Grid grid = GridGain.start()) {
   // Create sample runnable.
   Runnable r = new GridRunnable() {
      @Override public void run() {
         System.out.println("Hello World");
      }
   }
 
   // Broadcast to all grid nodes.
   grid.compute().broadcast(r).get();
 
   // Broadcast to remote nodes only.
   grid.forRemotes().compute().broadcast(r).get();
 
   // Unicast to some remote node picked by load balancer.
   grid.forRemotes().compute().run(r).get();
 
   // Unicast to some node with CPU load less than 50%.
   grid.forPredicate(new GridPredicate<GridNode>() {
      @Override public boolean apply(GridNode node) {
         return node.metrics().getCurrentCpuLoad() < 0.5;
      }
   }).compute().run(r).get();
}

Screencast

Here is a brief screencast showing how to get started with running computations on your cluster in under 5 minutes:

GridGain 6.1.0 Is Released with Support for JDK8 and Geospatial Indexes

2014-05-01T12:48:00.003-07:00

I am pleased to announce that GridGain 6.1.0 has been released today. This is the first main upgrade since GridGain 6.0.0 was released in February and contains some cool new functionality and performance improvements:

Support for JDK8

With GridGain 6.1.0 you can execute JDK8 closures and functions in distributed fashion on the grid:

try (Grid grid = GridGain.start()) {
  grid.compute().broadcast((GridRunnable)() -> 
      System.out.println("Hello World")).get();
}

Geospatial Indexes

GridGain allows to easily query in-memory data in SQL using in-memory indexes. Now you can extend SQL to geospatial queries. For example, query below will find all points on the map within a certain square region:

Polygon square = factory.createPolygon(new Coordinate[] {
   new Coordinate(0, 0),
   new Coordinate(0, 100),
   new Coordinate(100, 100),
   new Coordinate(100, 0),
   new Coordinate(0, 0)
});

cache.queries().
    createSqlQuery(MapPoint.class, "select * from MapPoint where location && ?").
         queryArguments(square).
             execute().get();

Near Cache in Atomic Mode

Prior to 6.1.0 GridGain supported near cache only in transactional mode. Starting with 6.1.0 near cache support was added to atomic mode as well.

Near cache allows for client-side caching (vs traditional server side caching) and renders significant performance improvements in some cases.

Fair Affinity Functions

Many know that Consistent Hashing provides a consistent distribution of data within a cluster that is resilient to server failures, but not many know that consistent hashing is not very fair. The discrepancies in distribution can be up to 20% which means that some servers will end up with 20% more data than others. This may create uneven load distribution when running cluster-enabled computations or queries.

GridGain 6.1 added two more affinity functions in addition to consistent hashing: Rendezvous and Fair.

Rendezvous affinity function works faster than consistent hashing and for smaller topologies (under 10 servers) provides a pretty fair distribution. One of the nice features here is that cache key affinity survives full cluster restarts. This means that you can back up data to disk and then reload it on restart knowing that all keys are still mapped to the same node.

Fair affinity function provides absolutely fair cache key distribution with all grid nodes holding absolutely equal amount of keys at all times. However, fair affinity function may change key-to-node assignment upon full cluster restarts.

Other Enhancements

Other fixes and enhancements involve improvements to multicast protocol for discovery and significant performance improvements for distributed cache queues.

You can download GridGain 6.1 here.

Five Easy Tips for Benchmarking In-Memory Data Grids

2014-04-16T20:07:00.001-07:00

Having spoken with many customers evaluating our product I am noticing that a majority of folks evaluating in-memory computing, whether it be data grid, map reduce, or streaming, do not know how to appropriately perform benchmarking. The right approach to distributed in-memory benchmarking is very different than benchmarking disk-based products, like databases, and generally requires experience and understanding of the delicate details of how network and garbage collections behave under load. With that in mind, GridGain will soon be releasing a benchmarking framework to help easily overcome all these challenges, but until then here is a list of things to watch out for.

1. Did you allocate enough memory?

If you have not allocated enough memory, the performance of your application may significantly degrade. From a user's stand point it may not really be noticeable until your application gradually becomes slower and slower, which is the first sign that you may be running out of memory.

Also keep in mind that if you allocate more than 10GB of memory per JVM, your application may start suffering from prolonged garbage collections. To mitigate that, check if the product you are evaluating has support for off-heap memory which cheats Java garbage collection by utilizing off-heap memory space.

When running benchmarks, you should constantly monitor memory consumption. You can do it with any light weight profiler, like VisualVm which is shipped by default with JDK.

2. Are your tests multi-threaded?

Benchmarking should not be performed from a single thread or a small number of threads, as generally such tests will not load the system. This is especially true for synchronous API calls, as the next API call has to wait for the previous one to complete. It is generally a good practice to run benchmarks from about 60 to 100 threads (on boxes with more than 16 cores, the number of threads can be higher).

When running benchmarks, you should monitor the CPU load on your system and add more threads if system is below 90% load.

3. Do you perform initial warm-up?

JVM HotSpot compiler will usually inline and precompile portions of the code that get executed the most. However, usually in load tests, this warm-up process takes about 20 or 30 seconds. It is generally a good practice to allow benchmarks to warm up for about 30 seconds before you start measuring performance.

When running benchmarks, you should do periodic print outs of your throughput (number of operations per second). The warm-up is usually finished when the throughput numbers stabilize.

4. Did you tune your garbage collection?

If you are seeing spikes in your throughput, it may be due to JVM Garbage Collection (GC). In this case it is best to use concurrent sweep GC which has proven to provide fairly smooth throughput without any large spikes or long pauses.

Here is a good link, which describes several GC tuning parameters: Tune Garbage Collection.

5. Do you use bulk operations?

Often many developers will sequentially execute multiple single cache 'put(key, value)' operations and then notice that performance is not ideal. The main reason is that whenever network I/O is involved, you should always strive to send as few messages as possible. To achieve this, a majority of data grid or caching products provide support for bulk operations, such as 'putAll(...)'. Bulk updates can often improve performance by 100x magnitude.

When using bulk updates, make sure to experiment with batch sizes - making them too big or too small can hurt performance. Also see if the product you are evaluating has already automated bulk-updates for you.

Five Easy Clustering Tips With GridGain

2014-03-25T12:47:00.001-07:00

Before diving deeper into what it means to easily cluster an application, let's start from defining what a cluster really is. Wikipedia has a pretty good explanation of clustering here, which is a high level definition that covers fault tolerance, load balancing, scheduling, etc. However, the real magic behind clustering is in making these complex distributed operations seem easy.

From development standpoint ability to cluster an application in most cases can be reduced to being able to easily perform the following functions:

Get list of all currently alive cluster nodes
Ability to create sub-groups of nodes within cluster at will
Exchange messages between any nodes within cluster
Listen to events from any node within a specified group
Easily compute and share data on any of the cluster nodes

Here are the coding examples on how to achieve the above in GridGain. I hope the code is simple enough to understand, but would be interesting to get some feedback on it. Feel free to comment.

Let's start from getting list of all cluster nodes:

Collection<GridNode> nodes = GridGain.grid().nodes();

Now, let's create different sub-groups of nodes within cluster:

// Remote nodes (all nodes, excluding local)
GridProjection rmtNodes = grid.forRemotes();

// Random remote node.
GridProjection rmtRandomNode = rmtNodes.forRandom();

// Current CPU load of remote random node.
double cpu = rmtRandomNode.node().metrics().getCurrentCpuLoad();

// All nodes on the same physical host as remote random node.
GridProjection hostNodes = grid.forHost(rmtRandomNode.node());

// All nodes marked by user as worker nodes.
GridProjeciton workers = grid.forAttribute("worker", "true");

Here is an example of message exchange between cluster nodes in GridGain cluster:

// User-defined message topics.
private enum TOPIC { MYTOPIC }
 
// Get message instance to provide messaging functionality 
// over a projection of remote nodes.
GridMessaging msg = grid.forRemotes().message();
 
// Register message listeners on all remote grid nodes.
msg.remoteListen(MYTOPIC, new GridBiPredicate<UUID, String>() {
    @Override public boolean apply(UUID sndrNodeId, String msg) {
        System.out.println("Received message: " + msg");
         
        return true; // Return true to continue listening.
    }
}).get();
 
msg.send(MYTOPIC, "Hello World");

This example shows how to subscribe an event listener on all grid nodes:

// This optional local callback is called for each event notification
// that passed remote predicate listener.
GridBiPredicate<UUID, GridCacheEvent> locLsnr = new GridBiPredicate<UUID, GridCacheEvent>() {
    @Override public boolean apply(UUID uuid, GridCacheEvent evt) {
        System.out.println("Received event: " + evt.name());

        return true; // Continue listening.
    }
};

// Remote listener which only accepts events for keys that are
// greater or equal than 10 and if event node is primary caching node for this key.
GridPredicate<GridCacheEvent> rmtLsnr = new GridPredicate<GridCacheEvent>() {
    @Override public boolean apply(GridCacheEvent evt) {
        System.out.println("Cache event: " + evt.name());

        int key = evt.key(); // Cache key.

        return key >= 10 && cache.affinity().isPrimary(grid.localNode(), key);
    }
};

// Subscribe to specified cache events on all nodes that have "myCache" running.
grid.forCache("myCache").events().remoteListen(locLsnr, rmtLsnr, EVT_CACHE_OBJECT_PUT).get();

And finally, an example that distributes computations to the nodes where the data is cached:

for (int i = 0; i < KEY_CNT; i++) {
    final int key = i;

    // This runnable will execute on the remote node where
    // the data with the given key is cached. 
    GridRunnable run = new GridRunnable() {
        @Override public void run() {
            System.out.println("Computing [key= " + key + ", value=" + cache.peek(key) + ']');
        }
    };

    grid.compute().affinityRun("myCache", key, run).get();
}

Grid-Enable Your Local Operations Across Cluster

2014-03-11T12:19:00.003-07:00

As you may already know, GridGain went open source last week. Going open source was a lot more involved than simply opening up our code. We put significant amount of thought into simplifying our APIs and making our development process as community friendly as possible.

As an example, take a look at how in GridGain you can take any local operation and distribute it across the cluster. Let's take GridCache interface. In addition to distributed methods, like get(...) or put(...), many APIs on this interface are local. For example, method size() will return number of entries locally cached, or method containsValue(...) will check if value is cached on local node. These APIs are made local on purpose - we anticipated that for certain methods providing local information would be safer and more useful. However, what if you need to see if value is contained across the whole cache, not just local node cache.

In GridGain, to make any operation distributed you need to execute it across multiple nodes using GridCompute functionality. Here is how the global contains method would look like:

public boolean contains(final V val) {
    // Not all nodes in the grid may participate in caching data.
    // We want to make sure that we send our computation only to caching nodes.
    GridProjection cacheNodes = GridGain.grid().forCache("myCache");

    Collection<Boolean> results = cacheNodes.compute().broadcast(
        new GridCallable<Boolean>() {
            @Override public Boolean call() {
                return GridGain.grid().cache("myCache").containsValue(val);
            }
        }
    ).get();
    
    return results.contains(true); 
}

You can do the same for any other operation as well. For example, if you need to find out total cache size across all cache nodes, you would simply broadcast a computation that returned cache.size() from all nodes, and them add them up to get a total value.

GridGain Goes Open Source Under Apache v2.0

2014-03-04T09:19:00.000-08:00

Yesterday GridGain released it's 6.0 version under the Apache 2.0 open source license. Our CTO, Nikita Ivanov, wrote about the new GridGain features and licensing in his blog here, so I will not repeat them. Instead, I will briefly describe our vision behind In-Memory Computing and why we made the move to open source.

Why is In-Memory Computing important? The simple answer is that there is no other way to process today’s enormous data volumes. In order to get answers from 100’s of terabytes of data in milliseconds you absolutely must have an In-Memory solution in your architecture. This is being validated by not just GridGain. Large vendors, such as Oracle (in-memory database and in-memory Exadata), IBM (BLU analytics), SAP (Hana), are also moving in the same direction.

So, with all those solutions out there, what makes GridGain different? In a nutshell, we provide a unified In-Memory Computing Platform aimed to solve a wide range of use cases. Our platform is composed of multiple natively integrated products, including High Performance Computing (HPC), the industry’s fastest In-Memory Data Grid (IMDG), CEP-based Streaming, and a plug-and-play Apache Hadoop Accelerator. With our new open source strategy, all of these products are now freely available for download, either a la carte or together as part of a larger platform edition.

With GridGain In-Memory Computing Platform you can process in parallel 100s of thousands of computational jobs per second, store terabytes of data in memory for fast transactional access and SQL querying, index into never-ending streams of incoming data, or give your Hadoop installations up to 100x boost.

We've been around the block as well. The product has been vetted by many customers, including large production deployments exceeding thousands of nodes. Open sourcing our platform just seemed like a natural way to share our technology with community and continue growing as a part of a larger in-memory eco-system. Unlike other commercial open source offerings, we went with a very liberal Apache license and with a feature set more than adequate to give GridGain open source users the ability to deploy in production. The product even includes Management and Monitoring, which most vendors rarely offer free of change.

In the upcoming days, I will be giving coding examples, demonstrating the ease of use of our APIs, and sharing various use cases. In the mean time, please feel free to download GridGain and give it a try. You can start by taking a look at our Getting Started guide and trying a few examples.