Billions of Entries and Terabytes of Data - BigMemory

Combine BigMemory and Terracotta Server Array for Performance and Scalability

The age of Big Data is upon us. With ever expanding data sets, and the requirement to minimize latency as much as possible, you need a solution that offers the best in reliability and performance. Terracotta’s BigMemory caters to the world of big data, giving your application access to literally terabytes of data, in-memory, with the highest performance and controlled latency.

At Terracotta, while working with customers and testing our releases, we continuously experiment with huge amounts of data. This blog illustrates how we were blown away by the results of a test using BigMemory and the Terracotta Server Array (TSA) to cluster and distribute data over multiple computers.

Test Configuration

Our goal was to take four large servers, each with 1TB of RAM, and push them to the limit with BigMemory in terms of data set size, as well as performance while reading and writing to this data set. As shown in Figure 1, all four servers were configured using the Terracotta Server Array to act as one distributed but uniform data cache.

Figure 1 - Terracotta Server Array with a ~4TB BigMemory Cache

We then configured 16 instances of BigMemory across the four servers in the Terracotta Server Array, where each server had 952GB of BigMemory in-memory cache allocated. This left enough free RAM available to the OS on each server. With Terracotta Server Array, you can configure large data caches with high-availability, with your choice of data striping and/or mirroring across the servers in the array (for more information on this, read here http://terracotta.org/documentation/terracotta-server-array/configuration-guide or watch this video http://blip.tv/terracotta/terracotta-server-array-striping-2865283. The end result was 3.8 terabytes of BigMemory available to our sample application for its in-memory data needs.

Next, we ran 16 instances of our test application, each on its own server, to load data and then perform read and write operations. Additionally, the Terracotta Developer Console (see Figure 2) makes it quick and simple to view and test your in-memory data performance while your application is running. Note that we could have configured BigMemory on these servers as well, thereby forming a Level 1 (L1) cache layer for even lower latency access to hot-set data. However, in this blog we decided to focus on the near-linear scalability of BigMemory as configured on the four stand-alone servers. We’ll cover L1 and L2 cache hierarchies and hot-set data in a future blog.

Figure 2 - The developer console helps configure your application's in-memory data topology.

Data Set Size (2 billion entries; 4TB total)

Now that we had our 3.8 terabyte BigMemory up and running, we loaded close to 4 terabytes of data into the Terracotta Server Array at a rate of about 450 gigabytes per hour. The sample objects loaded were data arrays, each 1700 bytes in size. To be precise, we loaded 2 billion of these entries with 200GB left over for key space.

The chart below outlines the data set configuration, as well as the test used:

Test Configuration
Test:	Offheap-test
svn url:	https://svn.terracotta.org/repo/forge/offHeap-test
Element #	2 Billion
Value Size	1700 bytes (simple byte arrays)
Terracotta Server #	16
Stripes #	16 (1 Terracotta Server per Mirror Group)
Application Node #	16
Terracotta Server Heap	3 GB
Application Node Heap	2 GB
Cache Warmup Threads	100
Test Threads	100
Read Write %age	10
Terracotta Server BigMemory	238 GB/Stripe. Total: 3.8 TB

The Test Results

As mentioned above, we ran 16 instances of our test application, each loading data at a rate of 4,000 transactions per second (tps), per server, reaching a total of 64,000 tps. At this rate, we were limited mostly by our network, as 64K tps of our data sample size translates to around 110 MB per second, which is almost 1 Gigabit per second (our network’s theoretical maximum). Figure 2 graphs the average latency measured while loading the data.

Figure 3 - Average latency, in millisecond, during the load phase.

The test phase consisted of operations distributed as 10% writes and 90% reads on randomly accessed keys over our data set of 2 billion entries. The chart below summarizes the incredible performance, measured in tps, of our sample application running with BigMemory.

Test Results
Warmup Avg TPS	4k / Application Node = 64 k total
Warmup Avg Latency	24 ms
Test Avg TPS	4122 / Application Node  = 66 k total
Test Avg Latency	23 ms

We reached this level of performance with only four physical servers, each with 1 terabyte of RAM. With more servers and more RAM, we can easily scale up to 100 terabytes of data, almost linearly. This performance and scalability simply wouldn’t be possible without BigMemory and the Terracotta Server Array. 

Check out the following links for more information:

·         Terracotta Server Array: http://terracotta.org/documentation/terracotta-server-array/introduction

·         Terracotta BigMemory: http://terracotta.org/documentation/bigmemory/overview

Performance Benefits of Pinned Cache

Terracotta is a Tiered Data Management platform that enables applications to keep as much of it's important data as close to where it needs to be. It does so automatically and can help keep data retrieval times at the micro-second level for up to a Terrabyte of data.

Terracotta Server Array tiered architecture

With Automatic Resource Control, we try to utilize the local cache in an efficient manner keeping the most accessed data nearest to the application. More the hits, more its locally cached. An application which accesses all the data (readonly & readwrite) equally, then ARC will try to distribute the local cache space equally to all the caches. Read more in my last blog.

In some applications, we are concerned about throughput (or minimum latencies) accessing one of the data set, which can be certainly achieved if we have it in local cache. Sometimes the developer/admin knows some specific things about an application that allow them to make specific performance decisions about subsets of data. The admin may know that by keeping the "US States Cache" in local heap one can keep a system running at maximum speed.

With Cache Pinning, we can pin the small caches locally(which can fit in local heap) while other caches are controlled by ARC. The latencies for the pinned cache would be minimum at micro-second level.

Here is a test which tries to simulate this kind of scenario and brings out the performance benefits of the Pinned Caches.

The test has 4 caches out of which one cache is a readonly cache. Each cache loads around 250MB of data, total of 1GB data. With 512MB of local cache, ARC tries to equally distribute the resources. Since the local heap is not enough to hold all the data, we store the data at lower layer.

Period Throughput Chart - All caches are Unpinned

The above chart shows that all the caches are getting accessed equally achieving almost same throughput. ARC is working properly but readonly cache is mostly reading from the lower layers as the some of the data is being moved to lower layer i.e. Terracotta, by other caches.

After pinning the readonly cache locally, the applicaion can access the data in readonly cache super-fast way. Readwrite caches will anyway be storing the updated values in Terracotta Server Array.

Period Throughput Chart - Readonly Pinned Cache/ReadWrite Unpinned Caches

The above chart shows that with cache pinning, readonly cache gets a huge boost from somewhere around 1000 tps is now touching 500k tps. As we keep the data closer to the application, it performs better. Now ReadOnly cache is holding up 250 MB (total data) out of total 512 MB of local heap.

Application Throughput

The total throughput of the app also gets a huge boost with cache pinning as ReadOnly cache is performing faster with low latencies and high throughput.

To enable cache pinning add following to cache config. Read more at ehcache.org.

<pinning store="localmemory|localcache|incache">

 Example:

<ehcache                           
    maxBytesLocalHeap="300m">      
    <defaultCache/>                
    <cache                         
        name="readonly-cache"      
        eternal="true">            
        <pinning                   
            store="localmemory"/>  
    </cache>                       
    <cache                         
        name="readwrite-cache-2"   
        eternal="true">            
    </cache>                       
    <cache                         
        name="readwrite-cache-3"   
        eternal="true">            
    </cache>                       
    <cache                         
        name="readwrite-cache-1"   
        eternal="true">            
    </cache>                       
    <terracottaConfig              
        url="localhost:9510"/>     
</ehcache>                         

To download the test click here.

From "Terrabytes :O" to "Just Terrabytes!!"

My first computer had 64MB of RAM, since then technology has improved and got a lot cheaper. We can get Terrabytes of RAM easily.

But when i talk about storing TB of data on a single java application/process,
I get reaction like are you insane or what!! TB of data on Java application, it wont even start and if it gets into GC (Garbage Collection), you can go and have a coffee at Starbucks even then it wont finish.

Then I say, BigMemory is the saviour you don't have to worry about GCs any more.
But still can BigMemory store TBs of Data without any performance degradation?

Here is my experiment, i tried loading 1 TB of data on a single JVM with BigMemory.
Tried loading 1 Trillion (yes, you read it correctly its thousand times a billion, which we call as trillion) elements of around 850 bytes of payload each. Total data is ~900G, hit the hardware limitaions, but we can sure make it more than TB if hardware is available.

Came across a huge box with 1TB of RAM, which made this happen. To reduce any GC issues, reduce the JVM Heap to 2G.

The test create an Ehcache and loads the data onto it. ehcache configuration used for the test.

<ehcache    
    name="cacheManagerName_0"     
    maxBytesLocalOffHeap="990g">    
    <cache         
        name="mgr_0_cache_0"         
        maxEntriesLocalHeap="3000"     
        overflowToOffHeap="true"/>
</ehcache>

Here is the graph of period (=4 secs) warmup thoughput over time. Secondary Axis of the chart show the total data stored.

There are few slight dips in the chart these are when BigMemory is expanding to store more data. The Throughput is above 200,000 all the time with an average of 350,000 txns/sec.

The latencies are also a big concern for the applications.

Its below a 1 ms and average being 300 µs.

Okay, we have loaded TB of data, now what. Does it even work?

Yes, it does. The test phase does read write operation over the data set. Randomly selects an elements updates it and put it back to the cache.

I will say throughput and latencies are not that bad :)
The spikes are due to JVM GC, even with 2GB heap we will have few GCs, but the pause time is not even 2 secs. So we get the max latency for the test to be around 2 secs but the 99 percentile is around 500 µs

So if you application is slowed down by database or you are spending thousands of dollars on maintaining a database.
Get BigMemory and offload your database!

There would be concerns about searching this huge data, we have ehcache search that makes it happen.

New efficient Ehcache Automatic Resource Control (ARC)

To speed up an application, the most common technique used is caching and Ehcache is most commonly used in Java world. BigMemory Ehcache with Terracotta can cache terrabytes of data without any GC issues, bringing data closer to the clustered application in efficient manner. Terracotta Server stores whole data in BigMemory and provides the required data to the clustered application. With new feature, BigMemory at application level (Terracotta client or Layer1/L1), we are bringing cached data more closer to the application.

In an application, we might need to cache different types data in different caches. At some point of time, we would be using one of the caches heavily with few hits to other caches. Before Terracotta 3.6.0 release, we can specify the number of elements to be cached per cache, but with limited heap/BigMemory it gets tough to allocate space for each cache in efficient manner. The memory allocated to each cache is fixed so even if its not used much, the data will reside with application.

With the new feature, Automatic Resource Control (ARC), Ehcache will manage the heap/BigMemory allocated depending on the cache usage. With increased usage of cache_0, Ehcache will try to allocate more memory to cache_0 and as usage of cache_1 increase, it tries to manage the space between both the caches.

Test Case

Here is a small test which creates two caches, loads up the data to both the caches. During the test phase, threads access both the caches, say cache_0 & cache_1, but for one of the caches, cache_0, introduced a small delay after each transaction reducing the hits to the cache.

Both cache can store 1.5GB of data, total of 3GB of data, while BigMemory allocated at the application level is 2GB only. Its enough to store all cached data for one of the cache but not both.

Access Pattern

1. 2 mins, both the caches are being used
2. Next 10 mins cache_0 is being used more often
3. Again for 2 mins both the caches are being used
4. Next 10 mins cache_1 is used heavily
5. Repeat

First, will like to discuss the case without ARC.




The tps remains almost constant even if other cache is not being used. We can see if both the caches are being used then also application throughput is same.



The L1 BigMemory usage remains constant throughout the test.

Now we should praise the benefits of ARC




Now with ARC, we can see if full cached data is at application BigMemory, the throughput gets a boost and touches 140k txn/sec. With both caches being used the tps is almost same as with and without ARC.



The throughput variation can be understood by the graph of L1 BigMemory usage. We can see that the L1 BigMemory usage for the cache_0 increasing, as it is heavily used. Overtime, it uses most of the memory for cache_0. As cache_1 usage increases, the memory usage for it also increases giving boost to the throughput.

To enable ARC, we just need to provide maxBytesLocalOffheap at Cache Manager level.

here is a sample ehcache.xml

 <ehcache           
   name="cacheManagerName_0"            
   maxBytesLocalHeap="512m"             
   maxBytesLocalOffHeap="2g">             
   <defaultCache/>                 
   <cache                      
     name="cache_1"             
     overflowToOffHeap="true">          
     <terracotta/>                 
   </cache>                     
   <cache                      
     name="cache_0"             
     overflowToOffHeap="true">          
     <terracotta/>                 
   </cache>                     
   <terracottaConfig                
     url="localhost:9510"/>             
 </ehcache>                      

This is 1 client attached to Terracotta Server to keep the testcase simpler. With multiple nodes, it will bring out the benefits more. :)

The above testcase is with only two caches, just picture if we have 10s of caches and tuning each cache would be a problem. With Ehcache ARC, its Ehcache responsibility to manage the data efficiently giving maximum throughput out of the application.

"Terracotta Fairy" brings BigMemory for Java users

I used to discuss about the Java goodness with my friends working on native platforms. But they always used to crib about Java., its slower, max latency is high, Garbage Collection(GC) ruins the user experience and tuning GC is a NIGHTMARE! Didn't had anything to defend on these points :(, as its the fact.

Garbage Collection kills the Java Application.

Being in Java world, loading 100G of data on java heap sounds crazy. I did some experiments to load 100G of data on single JVM. Even for Read-only case (no writes/updates to reduce the GC problems), it wont fit in 150G of heap.

On tuning GC, reducing Young Gen Space, reducing Survivor Ratio, etc, didnt helped much. The test just gets into back-to-back full GCs, killing the application throughput and latency. To get it working had to give 200 GB of heap and it performed so badly.

Then I wished, wont it be really nice to fit whole data, without any GC problems. "Terracotta Fairy" listened to us and here we have BigMemory Ehcache. BigMemory is a GC murder weapon from Terracotta, like an AA12 ShotGun.

Now, we can store 350GB of data with no GC .
Can you believe this !! Literally NO GC!!

Wanna see it with your eyes, here are the charts from the battle of Troy - On-Heap vs BigMemory.

The following charts shows the Ehcache use case which I thought would be fair enough. Ehcache being most widely used Java cache already outperforms other available cache. Didn't wanted to choose best use case for BigMemory i.e. 100% writes NOR the best for On-Heap i.e. Read-only. The read/write ratio is 50% reads and 50% writes. The hot set is that 90% of the time cache.get() will access 10% of the key set. This is representative of the the familiar Pareto distribution that is very commonly observed. The Test loads up full data in the Ehcache and then starts doing r/w operations on it.

These test machine was a Cisco UCS box running with Intel(R) Xeon(R) Processors. It had 6 quad-core 2.93Ghz Xeon(R) cpus for a total of 24 cores, with 378GB (396191024 kB) RAM, running RHEL5.1 with Sun JDK 1.6.0_21 in 64 bit mode.

For BigMemory testcase, had just 2G of java heap even when loading 350 GB of data while for OnHeap testcases, had java heap of twice the data size.





This chart compares the Largest Full GC duration occurred during the full run of the test. The numbers are taken from verbose GC logs.
If you take a microscope, then you can see there is a small green bar beside huge Al-burj tower types red bars. Those are GC duration for BigMemory :). Merely going above 1.2 sec, BigMemory surely kills Garbage Collection and removes the stigma on the Java.



This charts compares the Max Latency during the test run. As expected, this should be equal to the Max Full GC duration, since GC just blocks the application. BigMemory fairly defeats OnHeap here. Anyway who will like to have 4-5 mins of pause in his application, not me atleast !



Lets see how BigMemory throughput behaves with increase in the data size. The chart above shows that after certain point the throughput remains unaffected by the data size. Also, did a run with 350G of data and the tps/latency was constant. (Did we ever thought of caching 350G of data in an application :O). The drop in tps from 512M - 4G of data is because for smaller data size, ehcache stores the entries on heap (remember test is 10% hotset, so till hotset is on Heap & SMALL enough to fit, its faster). And we don't have much GC occurring for smaller data size.



Latency, the most worried about factor for user experience. We don't want our users to wait for 5 secs, because first impression is the last impression. The charts show the mean latency for the tests. Note that all those numbers are in micro-seconds. So they are anyway less than 0.5 secs, meeting your deadliest SLAs. BigMemory wins undoubtedly.



Here comes biggest test for BigMemory. Why will someone use BigMemory if it doesn't performs good enough as On Heap. We can't just ignore throughput for latency. We can see the BigMemory throughput numbers outperforms On Heap numbers here also. On-Heap throughput just goes on decreasing as full GCs would be killing the test. Imagine pausing for 4 mins during the test, will surely reduce the average throughput significantly.
Note: The test which I ran is 50% writes, so we might be overshadowing the onHeap throughput but 10% writes throughput were also comparable.



Mean latency graphs also says the same story which I have been saying all over my blog. BigMemory outperforms OnHeap :)


Do-it-yourself: Here is the svn link to the test, a Maven-based performance comparisons between different store configurations.
Note: You will need to get a demo license key and install it as discussed above to run the test.

So bottom line:
If you are fed-up of GCs, check BigMemory.
If you want to cache 350GB of data, check BigMemory.
If you want to use most AWESOME java cache ever made, check BigMemory.


If you like the post, vote it up on dzone. :)

Fire up Firefox

Make you firefox rock and change the way you browse. Optimize your bandwidth. Get what you want, exclude unwanted. Add-ons that change firefox

1. Adblock Plus - This one rocks than any other damn plugin. Blocks all the ads and you can put your own regex to block
2. All-in-One Gestures - Make gestures on Firefox to do what you want. Make a gesture zoom-in a image, open tab, much more.
3. Tweak Network - Have bandwidth still firefox downloading pages slowly. Install this and change to power mode.
4. Tiny Menu - Want more screen space for webpage than menus, etc.
5. Flashgot - Strictly for heavy downloaders. Configure to use Leechget and fire up your downloading.
6. Speed Dial - Speed dial your fav sites, instead of typing and wasting time.
7. Session Manager - Did you ever wanted to recover your last session, get your searches and pages back. Try this.
8. RAM Back - Firefox eating up your RAM. Get your RAM back.
9. Web Developer - Using GPRS/slow connection, dont want to waste your bandwidth on images etc, block them using this. Clear cookies, and more for a web developer.
10. IE Tab - Few of damned sites work on IE only, use this to get rid of them. Configure to use IE rendering engine everytime you visit those sites.
11. Firebug - Like developing/inspecting web applications, wanted to know whats slowing your page. Check this out!!!
12. Fox Clocks - Boss in different continent. Get this to get time there.
13. Cooliris - iPhone interface to browse images
14. Better Gmail2 - Gmail users try this out. Nice one!!!
15. ForecastFox - Get forecast on your firefox.
    

Glassfish with Standalone Terracotta Ehcache

While running standalone ehcache sample app (https://svn.terracotta.org/repo/forge/projects/tim-ehcache/branches/tc-3.1/tim-ehcache-1.7-sample) with glassfish server, glassfish client was able to make connection to tc server but fails with java.nio.channels.CancelledKeyException.


terracotta-server.logs

2009-09-25 12:34:56,037 [L2_L1:TCComm Main Selector Thread (listen 0:0:0:0:0:0:0:0:9510)] INFO com.tc.net.protocol.transport.ServerMessageTransport - ConnectionID(14.e02630feb1904743b5782b6324fa26a2): Attaching new connection: com.tc.net.core.TCConnectionJDK14@24008846: connected: true, closed: false local=10.0.4.158:9510 remote=10.0.4.158:53243 connect=[Fri Sep 25 12:34:56 PDT 2009] idle=1ms [28 read, 0 write]
2009-09-25 12:34:56,038 [L2_L1:TCComm Main Selector Thread (listen 0:0:0:0:0:0:0:0:9510)] WARN com.tc.net.core.CoreNIOServices - Skipping modifyInterest - WRITE on com.tc.net.core.TCConnectionJDK14@24008846: connected: true, closed: false local=10.0.4.158:9510 remote=10.0.4.158:53243 connect=[Fri Sep 25 12:34:56 PDT 2009] idle=0ms [28 read, 0 write]
2009-09-25 12:34:56,038 [L2_L1:TCComm Main Selector Thread (listen 0:0:0:0:0:0:0:0:9510)] INFO com.tc.net.core.TCWorkerCommManager - Selecting [L2_L1:TCWorkerComm # 0, wt:0] from [[L2_L1:TCWorkerComm # 0, wt:0], [L2_L1:TCWorkerComm # 1, wt:0]]
2009-09-25 12:35:06,041 [L2_L1:TCWorkerComm # 0] WARN com.tc.net.core.CoreNIOServices - Exception trying to shutdown socket output: Transport endpoint is not connected
2009-09-25 12:35:06,042 [L2_L1:TCWorkerComm # 0] WARN com.tc.net.protocol.transport.ServerMessageTransport - ConnectionID(14.e02630feb1904743b5782b6324fa26a2): CLOSE EVENT : com.tc.net.core.TCConnectionJDK14@24008846: connected: false, closed: true local=10.0.4.158:9510 remote=10.0.4.158:53243 connect=[Fri Sep 25 12:34:56 PDT 2009] idle=10004ms [81 read, 81 write]. STATUS : START
2009-09-25 12:35:06,042 [L2_L1:TCWorkerComm # 0] WARN com.tc.net.protocol.transport.ServerMessageTransport - ConnectionID(14.e02630feb1904743b5782b6324fa26a2): closing down connection - com.tc.net.core.TCConnectionJDK14@24008846: connected: false, closed: true local=10.0.4.158:9510 remote=10.0.4.158:53243 connect=[Fri Sep 25 12:34:56 PDT 2009] idle=10004ms [81 read, 81 write]

terracotta-client.logs

2009-09-25 12:34:56,035 [httpSSLWorkerThread-8081-0] INFO com.tc.net.protocol.transport.ClientMessageTransport - ConnectionID(-1.ffffffffffffffffffffffffffffffff): Attaching new connection: com.tc.net.core.TCConnectionJDK14@13859800: connected: true, closed: false local=10.0.4.158:53243 remote=10.0.4.158:9510 connect=[Fri Sep 25 12:34:56 PDT 2009] idle=1ms [0 read, 0 write]
2009-09-25 12:35:06,044 [L1_L2:TCComm Main Selector Thread (listen 0:0:0:0:0:0:0:0:50565)] WARN com.tc.net.protocol.transport.ClientMessageTransport - ConnectionID(-1.ffffffffffffffffffffffffffffffff): CLOSE EVENT : com.tc.net.core.TCConnectionJDK14@13859800: connected: false, closed: true local=10.0.4.158:53243 remote=10.0.4.158:9510 connect=[Fri Sep 25 12:34:56 PDT 2009] idle=10009ms [0 read, 81 write]. STATUS : SYN_SENT
2009-09-25 12:35:06,044 [L1_L2:TCComm Main Selector Thread (listen 0:0:0:0:0:0:0:0:50565)] WARN com.tc.net.protocol.transport.ClientMessageTransport - ConnectionID(-1.ffffffffffffffffffffffffffffffff): closing down connection - com.tc.net.core.TCConnectionJDK14@13859800: connected: false, closed: true local=10.0.4.158:53243 remote=10.0.4.158:9510 connect=[Fri Sep 25 12:34:56 PDT 2009] idle=10009ms [0 read, 81 write]
2009-09-25 12:35:06,044 [httpSSLWorkerThread-8081-0] WARN com.terracottatech.console - Timeout connecting to server: Timeout of 10000 milliseconds occured
2009-09-25 12:35:06,044 [L1_L2:TCComm Main Selector Thread (listen 0:0:0:0:0:0:0:0:50565)] WARN com.tc.net.core.CoreNIOServices - java.nio.channels.CancelledKeyException occured

glassfish logs

[#|2009-09-25T12:37:21.157-0700|WARNING|sun-appserver9.1|javax.enterprise.system.stream.err|_ThreadID=15;_ThreadName=httpSSLWorkerThread-8081-0;_RequestID=f34b40b7-5245-48d5-9df8-03fb63e09a7c;|2009-09-25 12:37:21,157 WARN - Timeout connecting to server: Timeout of 10000 milliseconds occured
[#|2009-09-25T12:37:36.169-0700|WARNING|sun-appserver9.1|javax.enterprise.system.stream.err|_ThreadID=15;_ThreadName=httpSSLWorkerThread-8081-0;_RequestID=f34b40b7-5245-48d5-9df8-03fb63e09a7c;|2009-09-25 12:37:36,169 WARN - Timeout connecting to server: Timeout of 10000 milliseconds occured



This could be solved by adding the following jvm option to glassfish domain.xml.

<jvm-options>-Dcom.sun.enterprise.server.ss.ASQuickStartup=false</jvm-options>