Grid Designer's Blog

Java tricks, reducing memory consumption

2010-01-06T23:01:00.000-08:00

In this blog post, I want to discuss optimization of java memory usage. The Sun JDK has two simple-but-powerful tools for memory profiling -- jmap and jhat.

jmap has two important capabilities for memory profiling. It can:
• create a heap dump file for any live java process
• show a heap distribution histogram

Neither of these capabilities requires any special parameters for the Java virtual machine (JVM). Below is a heap distribution histogram produced by jmap.

>jmap -histo:live 16608

num #instances #bytes class name

----------------------------------------------

1: 464773 45411544 [C

2: 148131 22404200
3: 21461 13197200 [S
4: 148131 11862064
5: 231304 11562840
6: 448533 10764792 java.lang.String
7: 14801 9520800
8: 14801 6706536
9: 20178 6060872 [B
10: 250951 6022824 java.util.HashMap$Entry
11: 12530 5532096
12: 26538 4437384 [Ljava.util.HashMap$Entry;
13: 54179 4381328 [I
14: 73294 3939656 [Ljava.lang.Object;
15: 34872 3905664 org.eclipse.swt.graphics.TextLayout$StyleItem
16: 27360 1551928 [Ljava.lang.String;
17: 16056 1541376 java.lang.Class
18: 20222 1294208 org.eclipse.core.internal.resources.ResourceInfo
19: 47006 1128144 java.util.ArrayList
20: 25235 1009400 java.util.HashMap
21: 22911 983784 [[I
22: 13398 857472 org.eclipse.swt.custom.StyleRange
23: 14362 831832 [[C
24: 16264 780672 org.eclipse.ui.internal.handlers.HandlerActivation
25: 36992 591872 org.eclipse.ui.internal.handlers.LegacyHandlerListenerWrapper
26: 423 591304 [Lorg.eclipse.swt.custom.StyleRange;
27: 13427 537080 org.eclipse.jface.text.TreeLineTracker$Node
28: 20254 486096 org.eclipse.core.internal.dtree.DataTreeNode
29: 19700 472800 org.eclipse.swt.internal.win32.SCRIPT_ANALYSIS
30: 19700 472800 org.eclipse.swt.internal.win32.SCRIPT_STATE
31: 17108 410592 org.eclipse.jface.text.Line
32: 1255 401600
33: 16650 399600 java.util.Hashtable$Entry
34: 10952 350464 org.eclipse.ui.internal.services.EvaluationReference
35: 8665 346600 org.eclipse.ui.commands.HandlerSubmission
36: 10624 339968 java.util.TreeMap$Entry
37: 6598 307376 [Lorg.eclipse.swt.graphics.TextLayout$StyleItem;
38: 18520 296320 java.lang.Integer
39: 8567 274144 org.eclipse.ui.commands.ActionHandler
40: 11268 270432 org.eclipse.ui.internal.help.WorkbenchHelpSystem$6
...
Total 2894736 198472504

For each class we can see the class name, the number of instances of the class in the heap, and the number of bytes used in the heap by all instances of the class. The table is sorted by consumed heap space. Although it is very simple, it is extremely useful. This information is enough to diagnose 60% of heap capacity problems.

If a heap distribution histogram is too high-level, you can try jhat. jhat can read and explore a heap dump. It has a web interface and you can click though your dumped object graph. Of course, clicking through a few million objects is not fun. Fortunately, jhat supports query language. Let's give it a try.

>jmap -dump:file=eclipse.heap 16608
Dumping heap to C:\Program Files (x86)\Java\jdk1.6.0_11\bin\eclipse.heap ...
Heap dump file created

>jhat -J-Xmx1G eclipse.heap
Reading from eclipse.heap...
Dump file created Wed Oct 14 08:41:07 MSD 2009
Snapshot read, resolving...
Resolving 2300585 objects...
Chasing references, expect 460 dots.............................................
................................................................................
................................................................................
.......................................
Eliminating duplicate references................................................
................................................................................
................................................................................
....................................
Snapshot resolved.
Started HTTP server on port 7000
Server is ready.

Now open http://localhost:7000 and you can see a summary of all classes. You can use standard queries via links or go straight to the “Execute Object Query Language (OQL) query” link at the bottom and type your own query. Query language is quite awkward (it is based on the java script engine) and may not work well for large numbers of objects, but it is a very powerful tool.

Enterprise applications are 80% strings and maps

Let’s look at the jmap histogram again. In the top row, we can see "[C" class consuming most of the heap space. Actually, these char arrays are part of String objects, and we can see that String instances are also consuming considerable space. From my experience, 60-80% of heaps in an enterprise application are consumed by strings and hash maps.
Strings
Let's look how the JVM is storing strings. String objects are semantically immutable. Each instance has four fields (all except hash are marked final):
• Reference to char array
• Integer offset
• Integer count of character
• Integer string hash (lazily evaluated, and once evaluated never changes)

How much memory does one String instance consume?

Here and below are size calculations for the Sun JVM (32bit). This should be similar for other vendors.
Object header (8 bytes) + 3 refs (12 bytes) + int (4 bytes) = 24 bytes. But the String instance is only a header. Actual text is stored in char array (2 bytes each character + 12 bytes array header).

String instances can share char arrays with each other. When you call substring(…) on a String instance, no new char array allocation happens. Instead, a new String referencing subrange of existing char arrays is created. Sometimes this can become a problem. Imagine you are loading a text file (e.g., CSV). First you are loading the entre line as string, then you seek the position of the field and call substring(…). Now your small field value string object has a reference to an entry line of text. Sometime later, a string header for the text line object is collected, but characters are still in memory because they are referenced via other string object.

In the illustration above, useful data is marked by yellow. We can see that some characters cannot be accessed any more, but still occupy space in the heap. How can you avoid such problems?
If you are creating a String object with a constructor, a new char array is always allocated. To copy content of a substring you can use the following construct (it looks a bit strange, but it works):
new String(a.substring(…))

String.intern() – think twice
String class has a method intern() that can guarantee the following:
a.equals(b) => a.intern() == b.intern()

There is a table inside of the JVM used to store normal forms of strings. If some text is found in a table then value from a table will be returned from intern(), else the string will be added to table. So a reference returned by intern() is always an object from JVM intern table.

String intern table keep weak references to its objects, so unused strings can be collected as garbage when no other references exist except from intern table itself. It looks like a great idea to use intern(), and eliminate all duplicated strings in an application. Many have tried … and many have regretted such decision. I cannot say this is true for every JVM vendor, but if you are using Sun’s JVM, you should never do this. Why?

JVM string intern tables are stored in PermGen -- a separate region of the heap (not included in -Xmx size) used for the JVM’s internal needs. Garbage collection in this area is expensive and size is limited (though configurable).
You would have to insert a new string into the table which has O(n) complexity, where n is table size.

String intern tables in JVMs work perfectly for the JVM’s needs (new entries are added only while loading new classes so insertion time is not a big issue) and it is very compact. But it is completely unsuitable for storing millions of application strings. It just was not designed for such a use case.
Removing of duplicates
The JVM’s string intern table is not an option but the idea of eliminating string duplicates is very attractive. What can we do?

public
class InternTable {

private
   final Map table = new HashMap();

public X intern(X val) {
X in = table.get(val);
if (in == null) {
table.put(val, val);
in = val;
}
return in;
}
}

Above is a simple snippet of a custom intern table. It can be used for strings or any other immutable objects. Looks good, but it has a problem. Such an implementation will prevent objects from being collected by GC. What can we do?

We need to use weak references. There are WeakHashMap classes in the JDK. It is a hash table that uses weak references for keys. Let's try to use it.

public
class InterTable {

private
final Map table = new WeakHashMap();

public X intern(X val) {
WeakReference ref = table.get(val);
X in = ref == null ? null : ref.get();
if (in == null) {
    table.put(val, new WeakReference(val));
    in = val;
}
return in;
}
}

Did you notice we also need a weak reference for the wrap value? This will work, but what is the cost?
• Reference from hash table – 1 ref * ratio of unused slots (~6 bytes)
• WeakHashMap$Entry object – 40 bytes
• value WeakReference – 24 bytes

The cost per entry in the table is about 70 bytes. That’s expensive. Can we reduce it?

Right now we have to have two weak references per entry. If we rewrite WeakHashMap to return an entry from the get(...) method (instead of the value), we can drop the second weak reference and save 24 bytes. But the cost of such an intern table is still high. You should analyze/experiment with your data to see if such a trick will bring greater benefits in your case.

UTF8 strings
Java strings are UTF and encoded using UTF16 in memory. If they are to be converted to UTF8, the actual text is likely to consume half the memory. But using UTF8 will break compatibility with the String class. You have to create your own class and convert it to standard String for a majority of operations. This will increase CPU usage, but in some edge cases this approach can be useful for overcoming heap size limitations (e.g., storing large text indexes in memory).

Maps and sets
Good old java.util.HashMap is used everywhere. Standard implementation in JDKs is to use the open hash table data structure.

References to key and value are stored in the Entry object (which also keeps the hash for the key for faster access). If several keys are mapped to same hash slot, they are stored as a list of interlinked entries. The size of each entry structure: object header + 3 references + int = 24 bytes. java.util.HashSet is using java.util.HashMap under the hood so your overhead will be the same. Can we store map/set in more compact form? Sure.

Sorted array map
If the keys are comparable, we can store the map or set as sorted arrays of interleaved key/value pairs (array of keys in the case of a set). Binary search can be used for fast lookups in an array, but insertion and deletion of entries will have to shift elements. Due to the high cost of updates in such a structure, it will be effective only for smaller collection sizes, and for operation patterns that are mostly read. Fortunately, this is usually what we have -- map/set of a few dozen objects that are read more often than modified.

Closed hash table
If we have a collection of a larger size, we should use a hashtable. Can we make the hashtable more compact? Again, the answer is yes. There are data structures called closed hashtables that do not require entry objects.

In closed hashtables, references from the hashtable point directly to an object (key). What if we want to put in a reference to a key but the hash slot is occupied already? In such a case, we should find another slot (e.g., next one). How do you lookup a key? Search through all adjacent slots until key or null reference is found. As you can see from algorithm, it is very important to have enough empty slots in your table. Density of closed hash tables should be kept below 0.5 to avoid performance degradation.

Maps structures summary

Open hash map/set
Cost per entry:

1 ref * 1.33 (for 0.75 density)
Entry object
Total: ~30 bytes

+ Better handling hash collisions.
+ Fast access to hash code.

Closed hash map/set
Cost per entry (map):

2 ref * 2 (for 0.5 density)
Total: 16 bytes

Cost per entry (set):

1 ref * 2 (for 0.5 density)
Total: 8 bytes

- Generally slower than open version.

Sorted array map/ser
Cost per entry (map):

2 ref
Total: 8 bytes

Cost per entry (set):

1 ref
Total: 4 bytes

- For small collection only.
- Expensive update/delete.

The JDK is using an open hashtable structure because in general case it is a better structure. But when you are desperate to save memory, the other two options are worth considering.

Conclusion
Optimization is an art. There are no magical data structures capable of solving every problem. As you can see, you have to fight for every byte. Memory optimization is a complex process. Remember that you should design your data so each object can be referenced from different collections (instead of having to copy data). It is usually better to use semantically immutable objects because you can easily share them instead of copying them. And from my experience, in a well-designed application, optimization and tuning can reduce memory usage by 30-50%. If you have a very large amount of data, you have to be ready to handle it. At Grid Dynamics, that’s our day-to-day job! So, if you are building system capable of handling enormous amounts of data, don’t hesitate to ask us for assistance :)

Oracle Coherence memory usage, indexes

2009-10-26T05:26:00.000-07:00

In previous posts we discussed a memory consumption in Oracle Coherence. We loaded 1M of DomainObj into cached and looked into memory dumps aquired with jmap utility. Now, we will talk about memory overheads caused by another powerful feature of Coherence - indexes.

Coherence support indexes. Indexes are another major memory consumer, and if you going to use indexes you should plan memory usage for them. We will use a distributed scheme with a local scheme on the back end in this case. An index is created for each of the DomainObjAttrib fields, and the number of unique values of this field should be about 1M/16 (each value of indexed field should match 16 objects).

<distributed-scheme>
<scheme-name>simple-distributed-scheme</scheme-name>
<backing-map-scheme>
<local-scheme/>
</backing-map-scheme>
<backup-count>0</backup-count>
</distributed-scheme>

First let's look at Coherence 3.5 :

row 6 (com.tangosol.util.SegmentedHashMap$Entry) - 1M of instances related to main storage and 1062.5k to the index;
row 7 ([Lcom.tangosol.util.SegmentedHashMap$Entry;) – 19Mb is related to main storage, the rest to the index;
row 12 – 7.2 Mb is related to main storage, the rest to the index.

Internally the index is constructed from 2 maps:

Forward map: maps indexed attribute value to a set of cache entries (I suspect that reference to the key is actually stored, but I am not sure);
Reverse map: maps cache entry (or its key) to the value of the indexed attribute for this entry.

In our case we have 62,500 unique values of indexed fields.

Now let's compare the memory picture with 3.4:

Rows related to the index are highlighted by yellow, also:

row 6 ([Lcom.tangosol.util.SafeHashMap$Entry;) – 7.2Mb related to main storage and the rest to the index.

An interesting thing to note: in 3.4 there are 1M DomainObjAttrib objects on the heap, but in 3.5 only 62.5k (and we know there are exactly 62.5k unique objects of this kind), so 3.4 is storing duplicates of objects, which 3.5 avoids. You can find more detailed information about this on the Coherence forum.

Summary for Coherence 3.4:

Domain objects: 227.3 Mb
Overhead: 116.4 Mb
Index: 88.5 Mb
Other: 6.5 Mb

Summary for Coherence 3.5:

Domain objects: 227.3 Mb
Overhead: 167 Mb
Index: 65 Mb
Other: 8.4 Mb

Again, we can see that Coherence 3.5 uses additional data structures to make partition operations more efficient, but, in the case of index, removing duplicated attribute values from the heap makes 3.5 more memory efficient.

Let's find a formula for index size (N number of entries, M number of unique indexed field values)

Coherence 3.4:

Forward map

4 * M – reference in hash table
24 * M – SafeHashMap$Entry
16 * M – SafeHashSet
56 * M – SafeHashMap (backend for SafeHashSet)
4 * N – reference in hash table of hash set
24 * N – SafeHashMap$Entry

Reverse map

4 * N – references in hash table
24 * N – SafeHashMap$Entry
<Size of attribute value> * N – stored attribute value

Total index size - 100 * M + 56 * N + N * <Size of attribute value>

In our case the estimated index size for 3.4 should be 100 * 6250 + 56 * 1M + 24000000 = 81.4Mb. The formula assumes a 100% fill ratio of hash tables; it produces the lower bound for index size. The actual size will always be little bigger (unless the hash table fill ration is greater than 100%, but it should never happen).

Coherence 3.5:

Forward map – 100 * M (bytes) + size of value objects.

4 * M – reference in hash table
24 * M – SegmentedHashMap$Entry
16 * M – SafeHashSet
56 * M – SafeHashMap (backend for SafeHashSet)
4 * N – reference in hash table of hash set
24 * N –SafeHashMap$Entry

Reverse map – 100 * M (bytes) + size of value objects.

4 * N – references in hash table
24 * N – SegmentedHashMap$Entry
<Size of attribute value> * M – stored attribute value

Total index size - 100 * M + 56 * N + M * <Size of attribute value>

In our case the estimated index size for 3.5 should be 100 * 6250 + 56 * 1M + 1500000 = 60Mb. Again we have the lower bound for index size.

Summary

Let's put the results of our experiments in single table. N is a number of entries in the cache, M is a number of distinct values of indexed field.

Coherence 3.4: >100 * M + 56 * N + N * A
Coherence 3.5: >100 * M + 56 * N + M * A

N is a number of entrees, M - is a number of distinct entries, A is a size of indexed attribute value.

Because the fill ratio of the hash table will always be below 100%, in practice, the cache will always consume slightly more memory than the value calculated by the formulas above.

Conclusion
Working with Coherence caches, it is always important to remember that both main data and indexes are stored in main memory. Here I tried to give a simple tool to get a good estimation of memory consumption.
Hope you will find it useful.

Oracle Coherence, memory structure of cache

2009-10-23T07:17:00.000-07:00

Imagine that you are working on project involving an in-memory data grid. You have analyzed your requirements and you can see that you need to store 10 million objects in your grid. The next question is how much physical memory do you need to provide such capacity? I recently have been facing the same question, and want to share some findings about how Oracle Coherence (popular in-memory data grid middleware) uses memory.

My approach is very simple. I am creating a cache, then puttung 1 million objects in it, and then analyze heap memory usage.

Local scheme

Lest start with local scheme. While it may not be so often used by itself, a local scheme may serve as a backing map for other schemes.

<local-scheme>
<scheme-name>local-schemescheme-name>
<local-scheme>

The memory picture for one million objects is:

(Sun JDK contains a 'lmap' tool that can display memory usage by objects for a live Java process. It is an extremely handy tool for memory profiling.)

On the diagram we can see our domain objects (DomainObjAttrib, DomainObject, DomainObjKey). Also we can see 1M of com.tangosol.net.cache.LocalCache$Entry objects. They are part of hash table implemented in Coherence. If you put the same objects in java.util.HashMap, the picture will be almost the same, but you will see 1M of java.util.HashMap$Entry instead.

Lets summarize memory consumption:

Domain object: 140.6Mb
Hash table: 77.3Mb
Other: 4.8Mb

In short, we have overhead of about 77 bytes per cache entry. How this space is used by Coherence?

72 bytes – size of LocalCache$Entry (BTW, the size of HashMap$Entry is just 24 bytes because it does not need to store additional data needed to support eviction and statistics collection);
4 bytes – reference from hash table;
Because the hash table fill ration is always less than 100%, some references are unused. This produces an additional 1.3 bytes per entry in our case.

Distributed scheme

Now let's switch to a distributed scheme (distributed hash table implementation of Coherence). As in the first case, we will use a simple configuration with a local scheme on the back end.

<distributed-scheme>
<scheme-name>simple-distributed-schemescheme-name>
<backing-map-scheme><local-scheme/>backing-map-scheme>
<backup-count>0</backup-count>
<distributed-scheme>

The memory picture for one million objects is:
Now things look much more complicated. First, there are no domain objects on the heap – that's because in distributed scheme objects are stored in serialized form. Coherence is using com.tangosol.util.Binary
objects to wrap the actual byte array (this way they can be store in the backing map), so we have 2M of wrapper objects (for each key, and each value). The next strange thing is SegmentedHashMap, which is not a part of local scheme, so it should be related to the distributed scheme. In Coherence prior to version 3.5 the data for all partitions was stored in a single backing map instance and to relocate a partition data from backing map, a full scan was required. Version 3.5 introduced a new mechanism - an additional index that remembers a key set for each partition. It improves partition relocation performance drastically but increases memory usage. Finally there is a pack of int[] objects, but this is still a mystery for me.

I think it is worthwhile to compare the 3.5 results with the 3.4 results at this point. With exactly the same configuration, the memory picture for Coherence 3.4 is:
Two things to note:

Size of LocalCache$Entry in 3.4 is 64 bytes instead of 72 in 3.5
There is no SegmentedHashMap related staff in 3.4

Let's summarize memory usage.

Coherence 3.4:

Domain objects: 227.3Mb
Overhead: 116.4Mb
Other 11Mb

Coherence 3.5

Domain objects: 227.3Mb
Overhead: 167Mb
Other 8Mb

Conclusion:

In the distributed scheme we have 2 additional Binary objects per entry (+48 bytes)
Also 3.5 adds additional data structures, which consume about 50 bytes per object

Data structures added in 3.5 are a trade-off to provide better performance in partition-related operations. They have their merits for sure, but is still would be preferable to have an option for turning them off.

External scheme

A local scheme is not the only option to use as a backing map. Coherence also has a so-called external scheme, which can store data off the heap using the BinaryStore plugin (plugins for NIO memory storage and several disk backend storage models are supported out of the box). Let's analyze the distributed scheme with an external scheme (nio memory) as the back end.

<distributed-scheme>
<scheme-name>external-distributed-schemescheme-name>
<backing-map-scheme>
<external-scheme>
<nio-memory-manager/>
external-scheme>
backing-map-scheme>
<backup-count>0backup-count>
distributed-scheme>

The memory picture for one million objects is:

Please keep in mind that we are analyzing only the Java heap and in this case some amount of data is stored out of the heap using direct memory buffers. You may expect that all your business data will be stored out of the heap, but still we can see plenty of binary objects in memory. These binaries are keys. Yes, while values are stored entirely in external storage, keys are stored in the heap (actually they are stored in both the heap and external storage). Coherence 3.4 shows a similar picture, so I will just omit it.

In this case the diagram does not show the full memory picture (only the heap) so you should not compare it directly to the previous cases.

Coherence 3.5:

Key duplicates: 32Mb
Overhead: 66Mb
Other: 18Mb

Conclusion:

While the external scheme is using non-heap memory for data storage, it istill consumes enough of the heap for keys and other structures.

Replicated cache

Cache configuration:

<replicated-scheme>
<scheme-name>simple-replicated-schemescheme-name>
<backing-map-scheme>
<local-scheme/>
<backing-map-scheme>
<replicated-scheme/>

The memory picture for one million objects is:
I rarely use a replicated scheme, so the number of additional data structures is a little surprising. As you can see, the replicated scheme is storing plain Java objects (unlike the distributed scheme, which is always operating with serialized blobs).

Memory summary is (for Coherence 3.5)

Domain objects: 140.6Mb
Overhead: 263.3Mb
Other: 7.6Mb

Conclusion

Lets put all results in a simple table with capacity formulae:

Here N is a number of entries, K is a size of keyset.

Stay tuned
Hope, this will help you to more presizely estimate memory consumptions of your Coherence cluster. In next blog posting I will describe the structure and overheads of secondary indexes in Oracle Coherence. Stay tuned!

Oracle Coherence using POF, without a single line of code

2009-09-18T03:36:00.001-07:00

People developing distributed Java applications know the importance of wire formats for objects. Native Java serialization has only one advantage—it is built in. It is relatively slow, not very compact, and has other quirks. Starting with version 3.2, Oracle Coherence is offering its own proprietary binary wire format for objects—POF serialization. POF is not only cross platform, but also much more compact and faster compared to built-in serialization. Both compactness and speed are extremely important for data grid application. The only disadvantage of POF is that you should write custom serialization/deserialization code for each of your mobile objects. Not only domain objects stored in cache should have serializers, but also entry processors, aggregators, etc. The amount of code you have to write may look daunting and force you to stick with built-in Java serialization.

But there is a simple way to get best of both worlds. In a recent project I have implemented a generic POF serializer. It uses reflection and doesn't require any changes in code, although you still need to register classes in coherence-pof-config.xml. Still it offers the advantages of the POF format – compact object size and performance. While performance is a bit degraded from using reflection (but still much faster than Java serialization), the sizes of serialized objects are similar to a handmade POF serializer.

Some people believe that reflection is slow. I have performed simple speed tests between Java reflection, a handmade POF serializer, and a reflection-based POF serializer—a very simple single threaded test doing serialization and deserialization of an object in a loop.

Java serialization – 8K ops/sec
Handmade POF – 46.4K ops/sec
Reflection based POF – 35K ops/sec

Sweet! I got 4 times speed up, 8 times reduced size just by writing 5 lines in coherence-pof-config.xml. Numbers may be different for different application but the trend is obvious.

And one more good thing—you can combine a reflection-based POF serializer with handmade ones. You can start with using generic implementation for all objects, and later write a few handmade serializers to reclaim ~24% lost on serialization for hot objects.

If you are interested you can look at the implementation here, available under the Apache 2.0 license.

GoGrid is out of beta

2009-08-31T05:06:00.000-07:00

GoGrid is officially out of beta, so now it's time to compare GoGrid with the public cloud market leader, Amazon EC2. Both vendors have relatively stable performance metrics and they both implement the basic functionality of a cloud provider:

Programmatic control via REST API
Scalable data storage solution: Elastic Block Storage in Amazon AWS and CloudStorage in GoGrid
Ability to assign external IP addresses
Ability to grant root/administrator credentials
Many options to configure the cloud cluster
Windows and Linux support
Web console for monitoring and management

So far, as cloud providers GoGrid and Amazon EC2 look very similar, but practical experience of using them in real-life projects helps us identify some unique features that can be important for specific applications.

For Amazon EC2 we can highlight the features listed below:

We can choose instances with different performance characteristics: Amazon EC2 has a product line that can be scaled by performance
Support for multiple OSes, some of them powered by marker leaders such as Oracle, Sun, and IBM. Also, the community has made a big commitment to Amazon EC2 resources.
User-friendly web console and web help
Service popularity: most questions about EC2 can be answered on the web; it's very possible that someone else has already solved the same problem
Start-up time for instances based on a popular AMI. It takes from 2 to 4 minutes to invoke an AMI and start the OS.
We always have access to instances. Even if we can't have SSH access into an instance we can look onto system logs or reboot and instance using the web console.

Although there are many positive aspects for Amazon EC2, some negative points can be identified as well:

There is no SLA for CPU, very low performance for m1.small and performance level of m1.small is unstable
Pricing for high performance instances m1.large and m1.xlarge is not attractive
No logging for API access
There is only one way to build hybrid clouds: use a VPN connection to boxes outside the EC2 cluster.
No control over instances deployment. It may happen that instances will be deployed onto one physical box, but for some applications it's critical to have instances deployed on different boxes. We can specify different regions and geozones for this application, but further control would be desirable.
Independent namespaces for different regions: each region has its own namespace for keys, AMI, instances, etc.
Security management isn’t always logical. For example, we can create instance without specifying a key-pair that will be inaccessible via ssh (in those rare cases when we really need it) or we can accidentally have access (and even terminate) instances that belong to the same account, but which were created with other security keys; for example, another development group in the same company.
Startup time of custom AMI : an instance's pending time can be about 10-15 minutes.

Compared to Amazon EC2, so far GoGrid does not have a big list of different features, but despite their recent entrants to the market, they already have some unique features:

True hybrids: we can build hybrid clouds along with colo servers, which will be located in the same network switch with virtual cloud instances.
Logging API usage by a job's history. Actually, we can monitor all requests to GoGrid’s API.
Non-transient instances: We can reboot instances without data-loss.

And, as with Amazon EC2, GoGrid has some challenges as well:

The web interface needs to be more user-friendly.
Using and mounting Cloud Storage is more complex than it is expected and it performs as a local disk drive only in case of low network load.
Choice of OS is limited only to Windows, RedHat and CentOS.
For cloud instances we can choose only RAM size, not different CPU levels. This point can be partially reconciled by using bare metal as part of the cluster infrastructure.
Practical experience with GoGrid's instances shows that maintenance windows happen more often than anticipated.

Obviously, choosing a cloud infrastructure provider for a particular project should be based on determining which application's requirements will be best fitted for this cloud provider and which requirement will not be satisfied. Providing hybrid clouds with bare metal boxes can be a unique differentiator, but not all applications really need it. The main promise of cloud computing is 'resources by demand' and this can be delivered by Amazon EC2 more than GoGrid—they provide a wide variety of different infrastructure parameters (such as CPU, RAM, and OS) for instances, they have more users, and they're currently more experienced as an infrastructure cloud provider.

Provisioning in Microsoft HPC

2009-08-25T03:00:00.000-07:00

Suppose you have five machines under your desk and need to establish a small HPC cluster for development from this estate, but you are too lazy to do it manually machine-by-machine. Or even better--you are in charge of a large HPC cluster that has thousands of nodes spanning multiple racks installed in an area over 2000 square meters. Obviously, manual installation is not an option here.

What to do? The answer is simple: use automatic provisioning. In this post, I'll try to share my recent experience provisioning with Microsoft High Performance Cluster 2008.

Good news: provisioning is an intrinsic feature of HPC 2008, which uses Microsoft Windows Deployment Server technology under the hood. WDS is a tool to exploit PXE (Preboot execution environment), which is a "must have" feature of all modern network cards.

This solution completely shields me from the complexity of WDS. Like many MS products, HPC Cluster manager provides a wizard for creation of a template. A template is a central notion of the whole system. It's merely an installation image, bootable over the network, accompanied by a script that does additional steps needed to get a node ready for joining the cluster.

However, just having an image is rarely sufficient. It allows you to install just the basic OS without anything on top of it, which is usually not what we want. That is where the script comes into play with its ability to execute (almost) arbitrary OS commands, even including executables from the network shares. If one tries to code those commands by hand, it would be somewhere between boring and very boring. Fortunately, HPC cluster manager's designers applied a lot of effort to make things simple. Just follow the wizard and you'll get a fully operational template in almost no time.

Hey, but what if you wanna add some specifics, something not provided by default? Well, nothing is lost, you can create a default template with the wizard and run the edit tool from the context menu. In the opened editor you can add new commands or delete the existing ones. When the template is ready, you invoke another wizard, which controls the process of installation. The only thing to do is to choose a template and turn on all the machines you want to have installed with this template. Simple.

Finally, if you don't have an installation image, you may create it from the distribution media by means of another wizard, embedded into HPC Cluster Manager. I'm not sure why this feature's needed, because for years the Microsoft installation media has included the installation images, but it might be valuable for image developers.

So far so good and it sounds like a magic, but there are some pitfalls I encountered when playing with MS HPC auto-provisioning.

First of all never ever try to run this kind of installation on a network that is not under your control, and first of all, DHCP. Obviously, HPC will need to add a record to DHCP for a new node.

Always use multicast mode when you are dealing with more then one node. It significantly reduces the time to provision, although even with multicast it may take a long time in some cases. In one of my experiments, I tried to provision 3 virtual nodes on the same VMware server with 5 additional VMs running. It took about 150 min on my not-so-old server (Supermicro SuperServer 6015B-UR 2 x XeonE5430 @2.66GHz (Quad-core) 16GB RAM).

And last but not at all least. Do not use the clusrun utility in your template scripts. Obviously, commands from a template will be executed on all nodes for which the provisioning is requested. If you try to run clusrun (whose task is to run its argument command on every nodes in cluster) on each node , you may get an unpredictable result.

All in all, Microsoft HPC implements a PXE style deployment in a comfortable way, allowing you to do this task without serious troubleshooting.

Enjoy!

GoGrid Management Tools and Library

2009-08-19T03:02:00.000-07:00

When we started working with GoGrid, we began using the web interface at my.gogrid.com. It's done pretty well -- powerful and user-friendly. However, after some time we decided to implement our own set of tools -- fortunately, GoGrid provides a great designed API -- for the following reasons:

While the UI is good, it has no facilities for automating things. For instance, if you need to create several servers of the same configuration using the web interface you have to create all the servers manually one-by-one, while using gg-tools, you can do something like this:

for i in `seq 5`; do gg-server-add -i 10 -r 2GB -n server$i; done

To create 5 servers with names like server1, server2, ..., server5 of the same configuration.

gg tools allow you to use all the power of the Unix command line. For example, if you need a password for server with ip, say 192.168.0.1, you can do:

gg-password | grep 192.168.0.1

and you will see the password for the required server right on your screen instead of searching in a long list of passwords and servers in the UI.

With the GoGrid UI, you have to remember your login and password. If you're using gg tools, you can configure it once and then do not have to type logins and passwords all the time. If you have it installed on a *nix box with ssh access, you can use it from every place where you have ssh client configured.

Initially, gg tools started as a few Python scripts to start, bootstrap and shut down servers. After working with these scripts they have grown pretty quickly and we decided it would make sense to split the bootstrap functionality and provide general tools for GoGrid management. Later on it turned out it would make sense to re-organize the code one more time and split out a general library that could be used not only by gg tools, but any project that needs to control GoGrid.

As a result, now we have a GoGridManager -- a Python module that supports almost all operations provided by the GoGrid API and gg tools -- a set of CLI tools that use GoGridManager.

GoGridManager is also quite easy to use. Here's a basic example:


(11:53) novel@hybrid:~/zf0 %> python
Python 2.5.2 (r252:60911, Oct  5 2008, 19:24:49)
[GCC 4.3.2] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> from GoGridManager import GoGridManager
>>> manager = GoGridManager()
>>> server = manager.get_servers()[0]
>>> print server
server my_go_grid (id = 28940, descr = , state = Started, ip = 173.1.54.221)
>>>

We have been working for almost half a year now with GoGrid and for many months gg tools has become an everyday tool for us. Recently it was rewritten to use XML responses
from the API and MyGSI support has been added. However, there is a lot of work left to do. For example, Load Balancer API calls are not implemented yet, as we haven't use it so far and
for example, Job and Billing support could be better.

Check GG Tools out on http://github.com/novel/gg/tree/master
Enjoy!

Further reading:

http://wiki.gogrid.com/wiki/index.php/API
http://wiki.gogrid.com/wiki/index.php/MyGSI

OpenLDAP perfomance testing

2009-07-13T03:17:00.001-07:00

Introduction

Our team is working on the metadata repository for a large-scale grid file storage project. This repository should make grid storage searchable and browseable. Since grid storage is presumed to store huge amounts of data, metadata repository should be able to store large amounts of metadata as well.

In our solution the metadata is typed, heterogeneous and has a hierarchical structure; metadata storage should be searchable, replicated and scalable. Also, we have strict requirements for new entry types to be definable at runtime. Taking these considerations into account, we chose to push for a LDAP-based solution. Since we had a requirement not to be tied to any particular vendor, OpenLDAP was a natural choice.

We've made a number of OpenLDAP performance tests to examine how OpenLDAP worked with large numbers of entries. Below, we will share some of the results we've achieved. We do not dive deeply into details of the testing machine configuration, since absolute timing values do not give many data points when you are trying to evaluate scalability of the solution. Trends, in this particular case with regards to the time consumed, are much more important.

The benchmarking stand can be described as a single-core Linux box using OpenLDAP version 2.4.16 (the latest at the moment of testing). OpenLDAP was built from the sources, as is recommended by the OpenLDAP documentation. OpenLDAP native HDB (based on Berkeley DB database with hierarchical database layout that supports subtree renames) was used as the database backend. All tests were performed by connecting to the OpenLDAP server through the loopback interface (127.0.0.1) to eliminate the impact of network bandwidth saturation.

For read performance measurements we used three LDAP entry attributes:

dn (distinguished name) - distinguished name of the LDAP entry, identifying an entry and its location.
cn (common name) - general string attribute of the LDAP entry.
entryUUID - builtin, server-maintained unique entry identifier. It remains unchanged when entry is being moved from one subtree into another on the contrary to dn, which is being changed.

Directory reading

Generally speaking, OpenLDAP finds an entry using an index much faster than delivering an entry client through the network. A more interesting case is how OpenLDAP handles concurrent reads and writes with regards to the number of clients.

Test I. Concurrent reading

The case: the OpenLDAP directory consists of a flat list of about 100K entries indexed by cn. Each client sequentially queries 1000 random existing entries by dn, cn and entryUUID. A number of clients (1 - 20) are executed simultaneously. Three curves on the graph show total time with regards to the number of concurrent clients in three different scenarios:

search by dn;
searching by cn;
search by entryUUID.

Testing results are shown in two charts, containing absolute timing values and normalized by dividing timing value by the number of concurrent clients.

Searching without indices was not tested because it is too slow on big data sets. As we can see, searching by distinguished name takes much less time than searching among children of the parent by indexed attribute. Another conclusion is that it is better to avoid searching by entryUUID despite the convenience of this attribute.

Directory reading combined with writing

Test II. Concurrent reading & writing

The case: the OpenLDAP directory consists of a flat list of about 100K entries. Each client sequentially performs 1000 requests. Each request is a read of a random existing entry by dn or by creating a new one. Three cases were tested: 99.9%, 99.0% and 90.0% of read operations against the corresponding number of write operations. A number of clients (1 - 20) were executed simultaneously.

So, we have three curves showing total time with regards to the number of concurrent clients in three different scenarios:

99.9% reads; 0.01% writes;
99.0% reads; 1% writes;
90.0% reads; 10% writes.

Testing results are shown in two charts, containing absolute timing values and normalized by dividing timing value by the number of concurrent clients.

These data contradict the OpenLDAP 2.1.21 measurements of Andreas Krennmair and Rainer Lischka (http://www.lizenzfrei.at/downloads/ldap-paper.pdf), where time consumption grows linearly depending on the number of concurrent clients. Moreover, there is no big difference between 90%-read and 99.9%-read curves. So, we can say that OpenLDAP performance has been significantly improved since version 2.1.21.

Directory writing

Test III. Directory writing with indices

The case: the OpenLDAP directory is empty in the beginning of the test. It is indexed by cn and objectClass. A single client uploads bunches of 100 entries at a time under one particular parent and measures the time of each bunch upload. After 50 000 entries were uploaded, the client uploads 50 000 entries under another parent in the same manner. Then the directory contents are erased and 50 000 entries are uploaded in bunches of 100 entries at a time. Each bunch is stored under a different randomly named parent. So, we have three curves:

Upload under first parent depending on the number of entries already existing under this parent.
Upload under second parent depending on the number of entries already present under this parent (not regarding 50K entries that exist under first parent already).
Upload under different parents depending on the total number of entries in directory tree.

Obviously, storing data as a tree is preferable to storing data as a flat list, because it is 2-3 times faster.

Test IV. Directory writing without indices

The case: the OpenLDAP directory is empty in the beginning of the test. It is not indexed at all. A single client uploads bunches of 100 entries at a time under one particular parent and measures the time of each bunch upload. After 50 000 entries were uploaded, the client uploads 50 000 entries under another parent in the same manner. So, we have two curves:

Upload under first parent.
Upload under second parent.

As we can see, the children count of one parent merely affects the creation time of objects under another parent. But while creating objects under one particular parent, object creation time grows depending on the number of children of the parent. This proves the conclusion, made in Test III, that storing data as a tree is preferable to storing data as a flat list.

Test V. Lots of objects created with and without indices

The case: the OpenLDAP directory is empty in the beginning of the test. It is not indexed at all. A single client uploads bunches of 1000 entries at a time under different parents and measures the time of each bunch upload. After 500 000 entries were uploaded, directory content is erased, indices on cn and objectClass are added and 500 000 entries are uploaded in the same manner. So, we have two curves:

Upload without indices.
Upload with indices.

As we can see, indexed and non-indexed directory write performances become similar on big directory sizes.

Dynamic schema extending

Test VI. LDAP schema writing

The case: the OpenLDAP directory is empty. The OpenLDAP schema is the default core schema. A single client creates object classes using a ldapmodify operation. Each operation adds 100 object classes. Each operation execution time is measured.

As we can see, the time to extend the schema grows linearly with regards to the total number of objectClass entries. This may be a bottleneck in LDAP system scalability, because the directory tree may be replicated, clustered some way, but it is impossible to split the LDAP schema into parts. But there is some good news: objectClass creation does not depends on directory data size. (This test gives exactly the same results on an empty directory and on a directory filled with 100 000 objects.)

Conclusion

In our performance tests OpenLDAP has shown pretty good results in most test cases. We were glad to see dramatic concurrent writing performance improvement since version 2.1. But there are still a couple of issues with the current version.

First, the OpenLDAP client developer always faces a dilemma: either to use DN for identifying an entry and face the risk of losing renamed or moved entries, or to use entryUUID for identifying an entry and get much lower performance. It looks like it is very hard to solve this problem due to the distributed directory tree structure.

The second and the most important issue is poor performance of dynamic LDAP schema extension. Also, the slapd daemon starts very slowly when the directory schema contains 10K objectClasses, even with slaptest disabled. (On our test machine OpenLDAP startup took several minutes with slaptest disabled and 10K objectClasses in the schema.)

We are eagerly awaiting schema management performance improvements in the future releases.

Data-Aware Routing on a Cloud, featuring Sun Grid Engine, GemFire and EC2

2009-05-18T04:19:00.000-07:00

We are excited to announce that we have taken our Convergence project to the next step in the last few weeks. Last time we demonstrated how data aware routing can speed up the combination of compute grids and data grids. Since then, we have developed new grid adapters for our Convergence project and moved to the cloud.

New adapters: Sun Grid Engine, GemFire

Sun Grid Engine (SGE) is an open source batch-queuing system, developed and supported by Sun Microsystems. SGE is typically used in a server farm or high-performance computing (HPC) cluster and is responsible for accepting, scheduling, dispatching, and managing the remote and distributed execution of large numbers of standalone, parallel or interactive user jobs. It also manages and schedules the allocation of distributed resources such as processors, memory, disk space, and software licenses. We have developed an adapter that wraps SGE's Java API and enables data aware routing of tasks.

GemFire is an Enterprise Data Fabric (EDF) solution from GemStone Systems, Inc. It is a high performance, distributed in-memory-data-grid (IMDG) that offers very low latency, high resiliency, scalability and high throughput data sharing and event distribution features for high performance computing applications that need access to real-time data. We have developed a monitor component, which can query location of data from GemFire regions with partitioned schema.

Running on the cloud

Along with integrating new grids, we changed our demo application to use a cloud infrastructure. It is deployed now on Amazon EC2. Our demo now allocates servers, completes setup of cluster software (SGE and GemFire in this case) and starts the demo application server on the fly. Just a single click and few minutes later you will have a new cluster up and running.

The deployment process is straightforward and requires just a couple of scripts. First, we allocate one EC2 server and start the master node. When the master is started, it starts allocating worker nodes and sets up the grid software. When the cluster setup is complete, we start the demo control UI on the master node, and from that moment our interactive data aware routing demo is available. Each cluster has a time to live, and when its lifespan expires, all servers are returned to EC2. We are using OpenSolaris AMI, provided by Sun Microsystems, for all our servers. Deployment of GemFire is trivial --we just need to copy few jars. Installing SGE has its quirks, but after we have figured out how to do it correctly, it just works.

From a usability point of view, cloud hosting has huge benefits. Each developer can work with his own cluster, or even several clusters (e.g., comparing the effects of data aware routing between clusters of different size).

Data-aware routing demo

We have adapted our existing demo application to work with new grids and be scalable on a cloud. In our demo application we are simulating a financial application. We have a set of trade objects, which are loaded into a partitioned GemFire region. 50k objects are placed on each server (size of each object is about 2Kb). Those trades belong to different equal-sized books (5K trades in each book). The job here is to evaluate all books using Sun Grid Engine to distribute work across cluster nodes. This means that a job consists of 10*(number of servers) tasks and each task should fetch 5K trades (about 10MB of data) from the data grid and perform some calculation over them. For simplicity, we just sum the IDs of the trades and return the result.

Unlike DataSynapse GridServer, SGE is a process-oriented computational grid. Each task in SGE is an operating system process, and execution time includes all JVM startup and class loading overheads. Usually tasks are long enough, and process start up time does not make much difference, but in the case of an interactive demo we should find a balance. We cannot make the user wait half an hour to see the first result, but if we make tasks too short, the effect of data aware routing will not be visible due to JVM startup overheads.

The demo runs a constant flow of jobs and constantly measures job completion latency and task completion latency. To illustrate data aware routing advantages we introduced 3 different scheduling modes. “Data aware” mode is where we perform data aware routing, ensuring local space access for the engine. “Neutral” mode is where we do unguided DS scheduling as SGE sees fit. “Anti data aware” mode is where we deliberately violate data awareness and ensure network space access. The user can change the task scheduling mode on the fly and see the impact of the scheduling mode on performance.

The diagram on the left shows the completion time of the job and small gray bars show cumulative task execution time on each server in cluster. Gray bars do not include JVM start up time, so cumulative execution time on each server is considerably smaller then job execution time.

The diagram on the right shows average task completion latency (without JVM startup time), so you can see effect of data aware routing on task and job level.

Conclusion

We have demonstrated how a generic data-aware routing approach implemented in the Convergence project can be used with different grid products. We had a very good experience migrating our research/demo platform to the cloud. Using a cloud provided us flexibility and comfort of development, which are hard to achieve in traditional resource constraint environment.

Stay tuned for more advances in the Convergence project!

Waters Power 2009 Conference

2009-05-05T09:50:00.000-07:00

Grid Dynamics was one of the sponsors for Incisive Media's Waters Power 2009 event that was held in New York City. Main agenda of the conference was to showcase the most up-to date developments in HPC with in-depth analysis of cloud computing, virtualization and SOA solutions, bringing the latest strategies, techniques and technologies that give optimum performance and maximum efficiency for any data center. Major HPC vendors and many wall street firms were represented at the conference

Key note speech was presented by Jeffrey Birnbaum from Merrill Lynch. It was a well presented speech that highlighted the opportunities, challenges and approaches of cloud computing for HPC. Highlights from his speech are

Main attraction of cloud computing is to drive the cost of computing down. Google and Amazon had done a good job getting the cost pretty low. How can the enterprises do the same, or come close?
Most cloud providers like Amazon, Google etc. are GbE Based. Everyone is moving towards 10GbE and this serves as foundation for viable clouds
Enterprise cloud infrastructure needs a global file system. All software is installed on that file system. This makes it simple for the end users, in order to run any compute environment - just mount a file system
In order to get global scalability - use multiple copies of the files
Replicate the files in real-time on any update. It might be better to wait on update than deal with eventual consistency
For better performance cache the files in regional locations
Do not provide node-level redundancy and increase the cost of hardware, buy commodity hardware and design for failure. Route the workload from a node that failed to some available node (like what Google and Amazon do)
Design Data Centers around PODs connected by layer-2, not layer-3.

Technologies that will change the world,

Multi-core/multi-socket commodity compute nodes (run more VMs, writing more parallel code)
10GbE with iWARP or RDMA (lower latency to storage)
Flash-based storage at 200K IOPS (totally changes how you think about the problems)

Implications of the above changes are dramatic and will affect the way we design the Data Centers and Applications

There was a Panel Discussion about "A silver lining for cloud computing". Panel included Victoria Livschitz, Founder and CEO of Grid Dynamics. Questions asked by the moderator triggered insightful discussions and sometimes contradicting answers,

Define cloud computing? This is probably most asked question in cloud computing forums and glad to hear we are closing in on a definition. This brought up another question about what stage of technology maturity is cloud computing in. Victoria's response to this was that it is in really early stages and it will take about 5 yrs to be mature enough for enterprise adoption
What applications are not suitable for cloud computing? It was acknowledged by few panel members that low latency applications may be bad candidates before cloud infrastructure matures but one speaker thought that this is up to the design of the cloud and it is possible to host low latency applications if architected well
An interesting conversation that came up is when cloud will get more adoption. This is same question that keeps coming up when any new technology comes into light. Old companies with huge investments in the existing infrastructure and approaches will need to spend more time adopting while newer players will benefit from them much sooner
Victoria made an insightful argument about the way applications were designed and implemented, this was about the data and its ownership. Most of the applications now are designed with data ownership as the premise but the promise of consuming and providing data as a service has many opportunities. For e.g., cloud computing is making infrastructure, software and computation as a service and having data also as a consumable entity gives the applications unbound possibilities

The thoughts about the data ownership and promise of being able to consume this as a service were echoed by few other speakers in later discussions

Ken Michellini's (from CitiHub) speech about "Keeping your feet on the ground: Unveiling the truth about cloud computing" was educational. It talked about the ROI calculations in a public cloud, 3rd party hosted and internal cloud scenarios and gave some ideas on how you can make these decisions when building your next big application. He also talked about characteristics of a good cloud application and typical challenges in building cloud applications.

There were few more Panel discussions that we attended including "Building the 21st century data center", "Preparing for Next phase in grid computing: What it takes to build the perfect data grid" and "Virtual reality: Optimizing storage, application and network virtualization" and another presentation about real life lessons learned while administering grids "Nuts and Bolts: Practical issues in grid administration"

I would like to extend congratulations to Incisive Media for a well conducted conference that let many financial industry experts brain storm and discuss the opportunities of Cloud. Great job guys!

Full-text & faceted search over In-Memory Data Grids

2009-02-09T22:39:00.000-08:00

Modern in-memory data grid (IMDG) solutions provide different facilities for execution of queries over whole stored data sets with different levels of sophistication. Oracle Coherence provides Query facilities (one time full scan and continuous querying with Cost-Based-Optimized). GigaSpaces has a JDBC Query interface with the ability to use hash and B-Tree indexes. Those solutions work quite well for many problem areas. However, for heavy loads and complex multi-criteria queries those facilities can quickly become a bottleneck.

There is a class of workloads that produce high loads with complex queries on IMDGs. Retail companies that use IMDGs for their item catalogs are a good example. Those catalogs are hit by diverse stream of multi-criteria queries. A typical query you may see there looks like:

give me cell phones with MP3 support, Java and in red color.

Fortunately, the Compass Framework allows you to process such queries effectively. You can build inverse indexes with Apache Lucene and store them on a grid. This capability is based on the very modular design of the Lucene framework. All index I/O operations are well-hidden by the abstraction of FileDirectory.

For now Compass provides implementations for Coherence, GigaSpaces and Terracotta, introducing an unprecedented ability to build a vertical search solution on top of In-Memory Data Grids.

In addition, Compass has a sophisticated object-to-document mapping system that allows you to make stored objects searchable just by adding Java annotations or XML mapping files. Mappings can also be built in runtime.

However, despite its great codebase, Compass documentation is pretty sparse. It may take significant time to dive into the code and docs to get what you want. But the results will overcome all your expectations. Search engine performance on top of data grids easily overcomes any old-generation search technology.

Enjoy!

Speeding up data-intensive HPC applications with Velocity

2008-12-22T00:57:00.000-08:00

Massively parallel, data intensive applications that run on computer grids require timely access to their data. When the application data is distributed among the grid nodes, data access is not a problem because it scales along with the number of computational task. However, when the application data is stored in a centralized database, access to data can quickly become a serious bottleneck.

We encountered such a problem in one of our recent projects, where a single database was used as a centralized application data repository for the entire compute grid. In this case, access to application data became bogged down enough to cause an overall degradation of the system performance, despite the fact that the database was being hosted on sufficiently powerful server.

In this article we will describe different ways of reducing database load and how they affect the overall system performance. We will also explore the advantages and disadvantages of Velocity as one of the ways to reduce data access latency. More specifically, we will show that:

Database load can be reduced by introducing Velocity distributed cache, or by simplifying SQL queries with moving all calculations from SQL to application logic.
Velocity CTP1 distributed cache improves overall system performance dramatically. In our case, querying data with active Velocity cache was up to 31 times faster than without it!
In CTP1, we experienced some scalability issues in particular configurations, but we are confident these issues will be resolved in future releases.

Compute Environment Overview
HPC++ CompFin Labs is a compute cloud that was created to give university students the ability to run massive analytical financial computations.

When a computation needs to be performed, one has to create a custom computational model that defines how the calculation will be performed and how the resulting data will be handled. The computational model operates in accordance with the MapReduce paradigm and includes an Excel-based UI with a list of input parameters, and the .NET assembly with the logic for splitting the computation into tasks, running a single task (map task) and combining task results into computation results (reduce task).

For example, one may write a model that performs some statistical calculation for a given stock symbol over some period of time. Later, someone else can open the Excel UI, change the stock symbol and the time period, and re-run the calculation. The following diagram illustrates how the example model would work in a CompFin cloud:

Picture 1. Original CompFin

The system consists from the following components:

A Microsoft HPC Server 2008 cluster.
A Sharepoint Server, where the CompFin website and the computational model files are hosted.
A SQL Server, where computation results are stored.
A SQL Server Intermediate Storage Provider (ISP), where task results are stored for subsequent consumption in the “reduce” phase.
A centralized market data database, where all financial data is located.

The computation proceeds as follows:

The user navigates to the CompFin website. On this website, the user chooses the model to run and opens the appropriate Excel file. Next, the user fills in the required computation parameters and, with CompFin Excel plug-in, submits the job.
The job start request is handled by the CompFin web service, which typically runs on the Sharepoint server as well. This web service calls the model logic to split the computation into tasks.
All computational tasks are submitted to the Microsoft HPC Cluster.
The map tasks are started first. Each task retrieves appropriate data from central market database, performs calculations on this data, and submits intermediate results in the SQL Server ISP.
When all map tasks are finished, the reduce task is launched. It retrieves intermediate data, combines it into final results data, and stores final results into the result storage.
After the computation is finished, the user may request job results via the CompFin website.
The computation results are retrieved from the result storage and returned to the user.

The above compute grid uses one SQL Server database for storage of stock prices, and one SQL Server database for storage of intermediate and final results. The combined compute capacity of the HPC compute cluster is 400GB of RAM and 200 (2.00GHz) Xeon cores, in particular, 50 machines each with 4 (2.00GHz) Xeon cores and 8GB of RAM. The capacity of each SQL Server is 32GB of RAM and 4 (2.33GHz) Xeon cores. So, the compute cluster had 12.5 times more RAM and 50 times more CPU. Based on these numbers, we anticipated two potential places of bottlenecks – the central market database, and the ISP.

Finding the bottleneck
The computational model, chosen for our experiments, computed correlation between stock prices over some period of time and contained very data-intensive computations with complex queries to the SQL Server that required significant grouping and sorting. The model used significant amount of data, most of which could be cached. The anticipated maximum cache hit ratio for the model was greater than 90%.

First, we needed to conduct a test of the model. We decided to test all three ways of scalability measurement (one data logical unit is roughly equal to 32 millions of records in database or 512MB of tick data returned from queries):

Leave amount of data to process, increase processing power:

Increase amount of data, leave processing power:

Increase amount of data and processing power synchronously:

We tested the model using this test plan and measured the entire time of map task, time spent in retrieving financial data from central market database (GetTrades method), and time spent in SQL Server ISP. There is also the graphic, which shows how the task time will behave if the system is completely linear:

Picture 2. Map task time distribution of original model in CompFin

You may notice that the bottleneck was in the central market database: time, spent in retrieving financial data is almost equal to the entire task time. You also may notice that currently the system is much worse than linear. So, it was a good chance for Velocity to help to improve the performance and scalability.

Getting rid of the bottleneck – Velocity cache for tick data
At first, we created a Velocity distributed cache for financial data. This was just an intra-job cache for reducing the negative impact of data-reuse inside the computation job. Complete replacement of central market database by the Velocity cache was not considered at this stage. This approach is shown as on the picture 3 (blue cloud):

Picture 3. CompFin + Velocity improvements (distributed and local caches)

Velocity cache for tick data ran on HPC cluster nodes. More precisely, the same machines, where the computation tasks ran, were also used as Velocity hosts.

We compared this approach with the Original CompFin:

Picture 4. Comparison of map task time between original CompFin and CompFin with Velocity distributed cache

In the beginning, everything was working well. Velocity distributed cache reduced the number of requests to the central market database and performance increased dramatically.

As the number of processors used for the computation increased to 128, the performance started to decrease. On 32 machines (128 processors), Velocity still dramatically outperformed the original CompFin results, but some performance degradation was experienced relative to the 16 machine (64 processors) scenario. On 50 machines (200 processors), in CTP1, we began to experience some failures and timeouts.

Nevertheless, keeping in mind that Velocity was in first CTP, the results were still quite amazing.

Velocity scalability – local cache & data-aware routing
Fortunately, we had a plan of how to further improve the performance and reduce the load on Velocity distributed cache. We enabled Velocity “local cache” and added data-aware routing in the CompFin (see picture 3). This was possible, because all tasks, which execute on the same HPC node, execute in the same Windows process. The local in-process cache can improve performance if the data-reuse factor for tasks that run in the same process is high. Hence the data-aware routing was required.

Below is the comparison between this approach and the approach with Velocity distributed cache alone:

Picture 5. Comparison of map task time between distributed cache and local cache

The problems with Velocity distributed cache scalability were solved, and we can conclude that Velocity can work well in CTP1 even on large clusters.

This approach performed from 5.6 to 31.9 times better than original CompFin. It also scaled better than any other approach, although not linearly. We believe that this is because the interaction with the SQL Server was still required before data was retrievied from the cache.
However, in the “Finding the bottleneck” chapter we mentioned that the model used complex queries with grouping to SQL Server. So, maybe the performance can be improved by just simplifying these queries and moving all logic from database to compute nodes.

Getting rid of the bottleneck via simplified queries
Following the previous assumption, we tried to reduce the load on the central market database by simplifying the queries used in the model through removal of grouping operations. The results were largely positive:

Picture 6. Comparison of map task time between original CompFin with complex and simple queries

The time, spent on retrieval of tick data, was reduced from 8.5 to 52 times and the task time became almost linear. However, the scalability problem with the SQL Server remained the same, so we tried to further improve the time by using our previous best approach – Velocity distributed cache with enabled local cache and data-aware routing:

Picture 7. Comparison of map task time between original CompFin and CompFin with Velocity local cache. Simplified queries are used in both cases.

Surprisingly, doing so did not improve the task performance significantly. Perhaps, the overhead of putting items in Velocity distributed cache was too big, or the cache services just stole CPU cycles from computations. Nevertheless, we believed that these factors had only a minor impact on performance so we decided to compare only the time spent in GetTrades method, where the financial data was retrieved:

Picture 8. Comparison of time, spent on retrieving tick data, between original CompFin and CompFin with Velocity local cache. Simplified queries are used in both cases.

Our assumptions were right and local cache approach greatly reduced time spent for retrieval of financial data. However, one question remained – why the overall task time was not reduced? To answer this question, we performed a number of tests to investigate the time distribution of task time, when the local cache was used:

Picture 9. Time distribution in CompFin with simplified queries and Velocity local cache.

These tests revealed that the SQL Server ISP, that was typically the least time consuming element of the system, now moved to the foreground. Since retrieval of financial data was now very fast, the frequency of write operations to the SQL Server has increased, and the SQL Server ISP became a bottleneck.

Our conclusion is that Velocity can improve even a very fast model, where only simple queries are used. In this environment, however, the SQL Server ISP remained as a bottleneck. We have yet to experiment using Velocity as the ISP with simple queries.

Conclusion
In this article we have shown that the database is the bottleneck for data-intensive applications, running on compute grids. We tried to reduce database load using two approaches: introducing Velocity distributed cache between the database and compute grid, and simplifying SQL queries to the database by moving all calculations and aggregations to the application logic.

Both these approaches improved the database performance dramatically. In particular, querying data with active Velocity distributed cache was up to 31 times faster than without it. In certain configurations, when experimenting with CTP1, we also observed some Velocity scalability issues; however, for a first CTP, Velocity performed very well. At the moment of publishing, Velocity is already in CTP2 and its development continues to be in very active stages. In the near future we expect the Velocity team to spend a lot of time focused on improving scalability and performance prior to release. Velocity team has big plans and not only will some issues soon be resolved, but lots of new functionality will be added as well.

So, if you have problems with database scalability and you use Microsoft technology stack, you may consider introducing Velocity to your system.

There are still many interesting topics to investigate regarding the Velocity and CompFin. Also, there can be some other improvements can be made in CompFin, so we hope that this project was not the last one, where we were working with Microsoft technologies.

Cloud Performance Reports

2008-09-24T14:45:00.000-07:00

Cloud computing is getting a great deal of attention these days. Unfortunately, there is very little data available about performance, scalability and usability of the cloud deployment platforms.

I was recently invited to speak at Silicon Valley Cloud User Group where I tried to bring a "practitioner's prospective" and present the results of three different recent performance and scalability benchmarks related to the cloud computing. The first benchmark aims to establish the scalability of EC2 on a perfectly parallel mathematical problem, a Monte Carlo simulation, executed by Grid Gain's popular open source map/reduce platform - and to document lessons learned in making the application scale to 512 nodes.

The second benchmark looks at a scalability of a more complex stateful application, typical to Risk Management, that required both in-memory data grid and compute grid. Both grids were running on EC2 and executed by GigaSpaces' data & compute grid platform.

The third benchmark looks at a prototypical data-intensive Portfolio Analysis application used heavily in the financial services industry, and studies the performance impact of data being located close to computing, or "on the cloud" vs. "off the cloud". This work was done in collaboration with Microsoft on their HPC++ CompFin Lab that integrates Microsoft Windows HPC Server, a central market data database and Microsoft productivity products to provide academic community with an online service to publish, execute and manage computational finance models.

You can find the presentation, with summary of results here. Please, note that these results are very fresh and the benchmarks in two cases are still going on. You can find far more details on the first benchmark in our previous blog post. We will be coming with more detailed blog reports for the second and third benchmarks soon.

Scalability Benchmark of Monte Carlo Simulation on Amazon EC2 with GridGain Software

2008-07-31T06:08:00.000-07:00

This blogpost presents the report on recently concluded scalability benchmark of Monte Carlo simulations running on Amazon EC2 using the GridGain framework. It consists of two parts: Part I is a technical report on the benchmark goals, method and results and Part II is an account of the development process and lessons learned.

Part I: Benchmark description & results

The goal of the benchmark was to study the scalability characteristics of massively parallel algorithms executed on Amazon ’s Elastic Computing Cloud (EC2) and managed by GridGain

A Monte Carlo simulation was chosen as an algorithm to represent its widespread use in financial applications. The same algorithm with different parameters was used on a wide range of grid sizes: 2, 4, 8, 16, ..., 256, 512. The parameters guaranteed that the amount of work performed by the whole grid was always linear with respect to the number of nodes. In other words, twice as many nodes always performed twice as much work. Perfect linear scalability would demonstrate the identical *completion time* of Job-1 running on 2 node and Job-2 running on 512 nodes, given that Job-2 had 256-times more work to do.

As Amazon permitted us to use the maximum of 550 nodes in a single run, the upper limit o fthe benchmark was chosen at 512.

The test utilized a full open source software stack, including GridGain, the Linux operating system, Sun Microsystem’s Open MQ JMS messaging and Sun’s Java 5 VM. Amazon’s preinstalled Fedora Core 8 with custom testing framework was used to conduct the benchmark.

The results justified our hopes: we could successfully run up to 512 nodes without significant performance degradation. The results graph is shown on figure 1.

Figure 1. Average task execution times on 2-512 nodes grids.

The performance degradation of 3 seconds (about 20%) should be considered minor given roughly 250-fold increase in scale. The curve rises two times: in the ranges 2-8 and 256-512, while 8-256 remains almost flat.

The range 2-8 growth can be explained by initial rapid growth of grid size, which causes the rise of maximum task execution time among nodes (the overall time is determined by the “weak link”). The range 256-512 growth could be explained by OpenMQ limitations as applied to our use case (the JMS load is quadratic against the grid size). In order to continue scaling the grid beyond 512 nodes, a clusterization of OpenMQ is likely to be required.

We can conclude that the obtained result showed near linear scalability and performance improvements from 2 to 512 nodes in all test runs.

Part II: How did we do it?

To start using EC2, just open Amazon EC2 Getting Started Guide, download EC2 API tools and follow the instructions. You can use one of the predefined public machine images or create a bundle with your own image.

The relatively big problem is the lack of persistence options in EC2. When an instance goes down, all the data is lost. However, there are third party solutions capable of mounting Amazon's S3 storage interface as a Linux filesystem.

Now, we needed to create a simple framework allowing us to manage the grid on Amazon’s EC2. The basic functionality would look like:

start the grid;
add/remove nodes to/from the grid;
display the grid health;
run single calculation task interactively;
run a benchmark (batch task execution on 1, 2, 4, …, 512,... nodes).

The first problem we encountered was that multicast is not supported in the EC2 infrastructure. By default GridGain is configured to make initial discovery by IP multicast. Unfortunately, as far as we know, the EC2 guys don’t plan to create fully functional multicast support anytime soon.

Luckily, GridGain is shipped with a set of various Service Provider Interfaces (SPIs). So, we could choose another DiscoverySPI that does not use IP multicast for discovery purposes, and we chose JMSDiscoverySPI. At first, we picked Apache ActiveMQ as a JMS implementation and soon ran into stability issues. Then we switched to OpenMQ and it proved to be sufficiently robust.

Our second problem turned to be the default maximum number of running instances per user – 20. Since we were going to run grids much larger than 20 nodes, we needed to override the default limit. It took several steps and a few more days to negociate with Amazon EC2, but eventually, we were granted the right to run up to 550 nodes. It seems that the business process of requesting a large amount of nodes is still not very well-defined by Amazon.

The image creation was not a very hard task: we got the public Fedora 8 i386 image, ran it, installed Java Runtime Environment 6, GridGain and ActiveMQ, then bundled this new image.

We still required some extra configuration since the grid node must start on system startup. The hardest thing was to write shell scripts for starting the grid. These scripts had to parse user data sent to the instance (different for the Head Node and Worker Nodes), determine the node type and start the necessary software in each given configuration. Later these scripts were replaced by more convenient Java-based tools and Head Node’s ActiveMQ was replaced by a standalone OpenMQ.

Figure 2. Initial GridGain on Amazon EC2 architecture.

So, how does the grid work? First, we start the Head Node (see figure 2). Head Node automatically starts ActiveMQ (only in early versions), GridGain in master mode (will be described later) and runs the requested number of Worker Nodes. Since Workers should connect to the JMS server, Head Node passes its IP address as user data for Worker Nodes. Worker Nodes just start GridGain.

We realized quickly that we needed some user-friendly interface to manage the grid. So, we rewrote the scripts in Java using the typica library. Thus we integrated Amazon and GridGain management. Now we can automatically start EC2 instances, wait for them to start, get their IP’s, track status, etc. Together with GridGain management it becomes a very powerful tool. Let’s imagine the grid running completely autonomously. The management module can automatically bring up more EC2 instances or shut them down depending on the current grid load.

When we had the simple web UI with capabilities of seeing the table of grid nodes and running our benchmark, we felt ready to run grids larger than 20 nodes. We asked Amazon to increase our running instances limit to 1050 nodes. Amazon agreed to let us run 550 instances. The know-how of our UI was the embedded scripting engine, allowing us to change a benchmarking schema without restarting the grid. To understand how simple it is to run a benchmark consisting of several task runs on different number of nodes, just look at this code:

var itersPerNode = 5000;
var cnode = [1, 2, 4, 8, 16, 32, 64, 128, 256, 512];
for (var i in cnode) {
var n = cnode[i];
grid.growEC2Grid(n, true);
grid.waitForGridInstances(n);
runTask(itersPerNode * n, n, 3);
}

Listing 1. Sample benchmarking script.

The closing remarks:

The framework we built is a good place to start developing a GridGain appliance. But what if we want to create, for instance, an In-Memory Data Grid on this basis? Our next step is making the framework more generic to allow easy integration with virtually any grid framework (either computing or IMDG). There are also other ideas such as creating a generic image and storing specific grid software/configurations in S3, which should ease debugging and small alterations to the appliance.

We are currently evaluating if our framework will be something more than just a GridGain testing framework. We'd love to hear from the community if this line of work is interesting to others beside ourselves. If you'd like to know more about this benchmark, or get access to the full source code, please contact me at mgorbunov@griddynamics.com

Grid technologies in middle-size applications

2008-06-19T02:03:00.000-07:00

Grid technologies were born to solve extreme problems and currently they are used primarily by large-scale applications. However, like computers, which were initially used only for solving complex scientific tasks and later came to almost every house, grid technologies are coming into middle-size enterprise market. In this article I will try to answer a question about how medium application can take advantage of grid technologies.

When writing large application that solve extreme problems, scalability is always an issue and you do not have a choice: you must invest resources in scalability and developers often have enough time and knowledge to solve this problem. When writing small applications, you usually do not care about this kind of problems, because all will work fine on a single server. But when writing middle-size application, you are in trouble, because you are already big enough to start thinking about scalability, but you are not big enough to invest a lot of resources in solving this problem. When middle-size applications grow from little ones the trouble becomes really serious. However, scalability problems in middle-size applications are usually caused by a very limited set of architecture decisions. Knowing these causes and their resolutions will help to build a more scalable application.

Whatever architecture you choose for a middle-size application, it will usually have a web server and a database. In the best case, it will have a single physical machine with both. In the worst case, it will have web servers, application servers and database server on separate machines. In all cases, the request processing chain will include several processes and will take a lot of time. Usually, the database server is also a bottleneck for the entire system, because it doesn’t scale well. While resolving these problems, mankind invented caches. Caches are useful to store some data that is costly to compute or retrieve. They can reduce both database load and request processing time a lot. However, caches can also be scaling killers.

Assume you have an architecture with a dedicated web server that hosts your application process. It is relatively easy to maintain a cache on a single server. But suppose you want to scale and the single server becomes a load balancing cluster, where each machine should maintain its own cache, and this cache should be synchronized with caches on other servers. For example, if some item is removed from the cache in one server, it should be immediately removed in every cache on every other server in the cluster. This is extremely hard to accomplish. But since developers often implement simple local caches by themselves, they try to enhance their caches to support distributed behavior. The problem is that developers often do not have enough knowledge and experience to do this. Fortunately, the problem of distributed caches is well known and the solution already exists.

The solution is to use third-party distributed caches or In Memory Data Grids (IMDG). In Memory Data Grids were created to solve problems of scaling data in extreme applications, where the cluster contains a hundreds of servers. Data stored on these servers can be partitioned between them or replicated. If data is partitioned, each server contains only one chunk of data and each chunk is stored on multiple servers to provide failover. This allows huge amounts of data to be stored in memory. If data is replicated, it is stored in full on each server. Of course, the data cached on each server is synchronized with data on other servers. This allows very fast access to data, because it is always available locally. In Memory Data Grids provide lots of other interesting features, which deserve an entire book to describe. Distributed caches are essentially a simplified form of an In Memory Data Grid. Currently there are many implementations of both IMDGs and distributed caches and if you choose to use these technologies you have a number of options.

The concrete choice will depend on what technology or framework you use in your application:

If you are using .NET, you may use ScaleOut, NCache or Microsoft’s new distributed cache Velocity, which is currently available as a Community Technology Preview. They all provide an ASP.NET session state provider – the easiest way to gain a benefit from grid technologies. With this session state provider you will not need to maintain a special SQL Server for storing ASP.NET session data, because all data will be distributed between web servers in the cluster in a reliable and robust way.
If you are using Java, Oracle Coherence and GigaSpaces are the most famous In Memory Data Grids and, hence, can be used as distributed caches. They both provide a second level cache for Hibernate, so, if you use it, you can scale easily with no additional development efforts.
If you are using C++, PHP, Ruby or Python, you should consider memcached. This is a very famous distributed cache initially developed for LiveJournal, which has already helped to scale many extreme applications, like Wikipedia, YouTube, Facebook and others.

All these implementations will help you to solve distributed cache problems and scale well. In simple cases, to start using them, you will need to replace your old local caches with new distributed ones. If you need to scale a Hibernate second-level cache, or an ASP.NET session state provider, you will not be required to write any code. However, in the case of complex and serious scalability problems you can consult with us at Grid Dynamics any time and we will help you to solve them in a most effective way.

Open Source Grid Expertise and Thought Leadership

2008-05-31T15:22:00.000-07:00

Last week, an exciting event took place here in the Bay Area: The Open Source Grid and Cluster Conference was held in Oakland. Engineers and scientists who care about the complex interactions between open source and grid computing from around the world, discussed their projects and research insights. Grid Dynamics was represented in two ways:
Eugene, our CTO, gave a presentation about Convergence, one of our key project about integrating Data grids with Compute grids. Check out his presentation.

Eugene Steinberg, CTO, Grid Dynamics

Eugene presented the latest version of Convergence which integrates DataSynapse GridServer5.0 and GigaSpaces XAP6.0. Future versions of Convergence will provide adaptors for Sun Grid Engine, Oracle's Coherence and GridGain.

The second event was moderated by Victoria Livschitz, our CEO: A panel about Trends in Open Source Data Grids with experts from the open source grid computing industry: Doug Cutting, creator of Hadoop, Nati Shalom, CTO and co-founder of GigaSpaces, Nikita Ivanov, President and Founder of GridGain and Daniel Templeton, Manager at Sun Grid Engine.

The discussion started with each of the panelist introducing their products, current initiatives and their commitment to the Open Source community. Each participant had an interesting perspective particularly to the question on why Data Grids technology is deeply anchored in the Open Source community, but many Compute Grids are not.

Attendees directly participated by asking the panel specific questions around their technologies and development efforts.

Left to right: Nati Shalom, Nikita Ivanov,
Doug Cutting, Daniel Templeton

The discussion continued informally after the panel with conference attendees and speakers discussing product specific questions and key future technology.

Doug Cutting in discussion with conference attendees

Victoria Livschitz and Nikita Ivanov

Nati Shalom with conference attendee
The conference setting allowed attendees to actively share their experience first hand, ask questions in an informal setting, hear the experts talk about technical details and strategic directions about the industry.

If you want to know more about the Open Source Grid and Cluster Conference, here is link to this year's conference. I also found a few bloggers taping the key note speech by Fritz Ferstl and some pictures of the event. Enjoy!

Match Making in Grid Computing

2008-05-29T04:15:00.000-07:00

Data partitioning is widely used to store and manage huge amounts of data across several servers. Often there is a need to introduce global distributed transactions to keep the data consistent. Those global transactions represent a natural limit to the scalability for the entire system.

Project Convergence was born as an example of managing interaction between in-memory data grid and computational grid to address specifically problem of data consistency in systems with global distributed transactions. Based on the idea of data aware routing, network utilization and overall performance will be increased. In addition, data aware routing can also improve scalability.

How does it work? Let me illustrate the solution by using a simple example: You are the owner of a dating agency. A candidate comes to your agency and looks for a person with specific parameters such as the love for operas, an interest in cooking and a liking for dogs . Let's assume that the set of these parameters is unique for each person and all parameters need to be matching. So, your agency will find him a perfect match or he will be put into your database as a single new customer waiting for a match. Considering your candidate’s specific parameters, you expect that it will take some time to find him an exact match.

Your company has an outstanding reputation and you are flooded with candidates and your company cannot handle all the new candidates. You realize that you need to get a business partner and you partner with another dating agency in town. You also hire as secretary who greets your candidates and sends them randomly to one of the offices.

One day, two candidates with the exact same parameters (a perfect match) enter your office and your secretary sends one of them to your office and the other to your business partner’s office. Both of you start looking independently for an exact match simultaneously, but both of you cannot find one in your databases. As a result, you and your companion mark both candidates as single, waiting for a match while you miss the opportunity to generate a fee from the match. How can you solve this unfortunate incident?

First, you can have your secretary manage the assignment better by having her search your and your business partner's combined database: But while she is using the database, nobody else can access the database until she is done with her search. This may produce a match of the candidates, but create a bottleneck and hence impact the performance of your match making capabilities.

Secondly, you divide all candidates in two groups (for example, those that like dogs and those that do not, assuming there are about the same number of candidates that like or dislike dogs). This way, you and your partner are responsible for an equal amount of candidates. You will instruct your secretary to send incoming candidates to the offices depending on the likes/dislikes of dogs. Because you are processing only half of the candidates with one parameter already matched in your database, you will achieve a match at a faster rate than before.

This is exactly how in-memory data grids and computational grids interact and how data aware job scheduling can address the ‘matching of data grid and compute grid requests.’ As in our example, a database or an in-memory data grid (Oracle Coherence or GigaSpaces XAP), represents the databases with all the parameters of the candidates. You and your business partner’s represent the computational grid (GridGain, DataSynapse GridServer or Sun Grid Engine), and your secretary fulfills the function of a data aware job scheduler.

Choosing graph manipulation library for your Flex app

2008-05-27T02:22:00.000-07:00

The open-source movement plays a pretty big role in everyday life here at Grid Dynamics. We always try to do our best to improve the software we're working with and contribute back into the community.

While developing an RIA project we faced a need to display complex graphs within an Adobe Flex application. It turns out that there are not many solutions to this problem existing at the moment. Actually AFAIK there are only Flare toolkit, yFiles Flex, SpringGraph and Flex Visual Graph Library (further on FVG). Having done the comparison between the libraries we have chosen Flex Visual Graph Library as our current graphing solution.

Good Things and Bad Things

Here is the list of the most distinctive features of the FVG:

It is an open-source project so everyone can use it for free and modify it in the way he likes;
It can render not only trees but also (relatively) complex graphs;
It contains a number of algorithms for performing layout;
Laying out the graph is performed on the client-side and no server resources are involved in the process.

The library has some documentation and samples, although sometimes (and maybe too often) one has to dig into its source code to figure out the way it works.

However, it is a young project so it is expected to be imperfect. Here are the main drawbacks that sometimes annoy me:

A somewhat monolithic library design, god-classes anti-pattern here and there; favor is given to state changes instead of the parameter passing. I would also mention here coding conventions, although it is surely my personal impression;
Lack of documentation on layout algorithms; incomplete design diagrams, tutorials and examples.

When dealing with someone's else products instead of the homegrown ones you will almost always find its feature set is incomplete. This is the case not only for open-source projects but also for closed-source ones (including commercial) and the greatest advantage of open source is that it allows you to implement the desired features and not wait until someone else does it. And surely there were missing features in FVG that our customers and we ourselves wanted to have.

Our humble help

So here is a list of the things we improved:

Fixed and successfully harnessed functionality for adding and removing of nodes. Before the fix it was impossible to do that after the graph was initialized. It seems that the developers of the library just did not need it so they haven't got it into shape.
Fixed support for directed graphs. It was one of the most controversial changes to the library. Before, the layouting algorithms were tested only on nondirected graphs so the spanning tree in the case of a directed graph was computed in a wrong way. This led to the graph being laid out inappropriately. Now the user can choose if he wants to get the new behavior or emulate the old one.
Enhanced hierarchical layouter. It now supports interleaving of nodes. Sibling nodes (the nodes on the same level) are placed now in a checkerboard formation so every even node is put a bit higher than an odd one. This enhances the readability of the graph and the overall appearance if nodes have text labels under them.
Added possibility to sort sibling nodes in a specified order. This again enhances readability. Suppose that you have some text on your nodes. You can sort the siblings in the lexicographical order so you can more easily find the node you want with your eyes. Or you may want to display a certain type of nodes first. Plugging in your own sorting logic is tremendously easy.
Added a very cool feature: graph zooming. There was nothing like that in the library before. One could only try to tune some parameters of layouters in order to achieve zooming effect but the real zooming for every layouter was impossible.

In addition, we've been implementing numerous small fixes and improvements — and we are definitely going to continue this effort in the future.

Conclusion

In general, supporting the open-source library has the following noticeable peculiarity: as with other source management models, it is always needed to keep compatibility with those who already use your library in their applications. But when a need to make significant and incompatible changes to the library appears, one can always commit the changes into a separate branch. It turns out to be a rather common practice in the open-sourced world. Although this is not as convenient as just committing the changes directly to the trunk, the approach has advantages. It allows you to not break the compatibility and let other developers test and review the new functionality. We've been maintaining a separate branch for quite a long time, partly because of tensions with the maintainers, partly because of a lack of time to perform all the necessary merges and testing. However, not long ago we decided that this will likely cause the branches to diverge in some point of time — definitely that will not add value to the library, but rather subtract from it. So we've consolidated our efforts with the maintainer of the library and presented a demo of the new features. The guys were very collaborative and accepted our changes into the trunk with just a couple of minor changes.

As we going to use the library further we will definitely continue to put effort in. So our next step is going be reviewing the FVG's architecture and proposing plans for refactoring the code. On the features side, we're going to have a hierarchical layouter that will be able to support multiple root nodes, a zooming engine that will preserve label and node size, node renderer that supports overlay icons, improved edge-node connections... lots of fun things to come.

Stay tuned!

Partnership with GridGain

2008-05-05T08:52:00.000-07:00

Today we announced a new partnership with GridGain (see the press release here). I am extremely pleased to be launching this alliance and hope it will accelerate the already impressive adoption of GridGain's new and exciting open source Java platform for high-performance grid applications.

At Grid Dynamics we see GridGain as a simple, elegant, extensible, open-source, pure-Java platform that makes traditional high-performance computing and all modern map/reduce applications significantly simpler and cheaper to develop and deploy than it has ever been possible before. The platform's extensibility also allows GridGain to be used for a wider set of applications and enables the integration with other tools, systems and frameworks.

The partnership gives GridGain's customers an integrated Software + Service offering, enhancing GridGain's innovative technology with Grid Dynamics' expertise and track record in designing complex, scale-out applications. Customers are getting best-of-breed solutions from the industry leaders in their respective areas and should expect a seamless experience working across both organizations. Grid Dynamics and GridGain share the same core values, highly regarded by their mutual customers:

Both are fundamentally engineering companies, nearly obsessed with the performance, scalability and quality of applications we write.
Both recognize the importance of the Open Source as a medium for cost-effective and open software products with almost "organic" properties, which thrive and evolve in the marketplace under a watchful eye and the collective efforts of their designers and users.
Both are cost-efficient, global-from-the-cradle organizations with significant R&D operations in Eastern Europe and headquarters in the Silicon Valley.

Under the terms of the relationship, Grid Dynamics is creating a "Center of Excellence" around the GridGain product and extending its design patterns for scalable high-performance applications to include mappings to GridGain technology. Both companies are collaborating on the joint R&D projects, including the planned release of a GridGain plug-in for Grid Dynamics' open source project Convergence, designed to enable interoperability between the leading compute and data grid platforms. Together, Grid Dynamics and GridGain simplify high-performance computing, lower the barrier of adoption of map/reduce paradigm for thousands of customers and ultimately enable a better scalability of a wide range of applications, networks and web sites.

I am personally looking forward to a lasting relationship between our companies that benefits GridGain users and the entire open source high-performance computing community.

Binary Calculator Project - What is the footprint of my GigaSpaces Entries?

2008-04-03T06:57:00.000-07:00

In the initial stages of every Data Grid project it is always essential to get good estimates of memory requirements. How much memory will my domain objects converted to Entries consume and what will be the indexing overhead? The answer to this question defines what JVM heap size to choose and how many of JVMs will be needed to store the intended dataset - or determine how much data can be stored within a given hardware footprint.

For GigaSpaces, it is hard to provide accurate theoretical estimates of your Entry size. Here are the reasons:

Space doesn't store entries as heap objects - they are stored decomposed to fields
String uid is generated and stored along with each entry
Index overhead is dependent on type of index (ordered or unordered ) and on a field dataset cardinality.

For these reasons it is far more practical to take an experimental approach in measuring entry footprint than trying to apply a theoretical formula. The goal of the Binary Calculator project is to build a convenient toolkit for this problem. The main idea is quite simple:

First, collect the basic statistics on the efficiency of storage of the individual entities:

Connect to remote space
Get the batch of tested entries from some entry source
Write a batch to remote space
Perform remote garbage collecting
Measure memory usage
Repeat step 2

After a sufficient number of iterations is done, we will get a number of data points in the format (entriesWritten, memoryUsage). Performing a linear approximation (e.g. min square linear fit) we receive the approximation of single entry footprint (including index overhead).

Implementing this idea, we have built an initial version of the Binary Calculator, which can be used as a toolkit for measuring arbitrary entry footprint. It has a very simple GUI that shows progress of the memory experiment.

We are planning to turn this simple toolkit into a much more powerful tool, which will generate entries on the fly, based on user-supplied meta data. This way, the user can specify an Entry Description as a simple table in a GUI:

Type	Indexed	Number of fields	Avg Length
Long	Yes	1	N/A
String	No	3	1000
String	Yes	2	5000
Integer	No	3	N/A

BinaryCalculator will generate Entries at runtime based on this description, populate it with random data, perform memory experiments and show estimated entry size.

Also, we are planning to build a lightweight plugin system to supply a custom EntrySource,
for example, your own random entry generator or JDBC or Hibernate data source. Consequently, performing full fledged capacity experiments loading real data from the database will be much easier.

We hope that this tool will be quite useful for GigaSpaces implementors in the field.

Introducing the PackRat Project

2008-04-01T07:12:00.000-07:00

Remember the legendary Fallout 2 game? One of its cool features was perks. You could increase the number of action points, strength, critical strike percentage, etc. One of the most useful perks for those who wanted to carry as much as possible was the perk called Pack Rat.

But how does it relate to Data Grids? The answer is very simple: it is possible to be more efficient at storing the data in the data grid by "packing" the objects. That's the goal of the PackRat project, released yesterday on OpenSpaces.org. The initial idea came out of a customer project with TuneWiki, a lyrics distribution site developed with GigaSpaces XAP 6.0 where it was important to pack as many lyrics as possible in a space without noticeable loss of search and retrieval performance.

Let's say we want to develop a hypothetical lyrics distribution service. First, we need a simple mechanism that packs together all entities related to a single lyric:

import org.openspaces.packrat.annotations.*

@CompressedClass(type = "entry")
public class Lyrics {
  @IndexField
  private Integer id;
  @IndexField
  private String title;
  @IndexField
  private String artist;
  @CompressedField
  private Integer version;
  @CompressedField
  private String language;
  @CompressedField
  private String comment;
  @CompressedField
  private String album;
  @CompressedField
  private String year;
  @CompressedField
  private String genre;
  @CompressedField
  private String lyrics;

  // other stuff
}

Let's say a song can be searched by artist and title (kind of a composite key). This means that all other non-searchable fields (like our lyrics field) can be stored in compressed form in a binary field and unpacked on demand. The definition for such entry looks like this:

public class LyricsPacked implement Entry{
  private Integer id;
  private String title;
  private String artist;
  private byte[] binary;

  // other stuff
}

To pack and unpack entries we need to create an instance of Packer. This class implements Entry interface and has a method pack, which packs a given object. This is how you may write a packed entry to a space:

Packer packer = new Packer();
IJSpace space = (IJSpace) SpaceFinder.find("jini://*/*/mySpace");

Lyrics lyric = new Lyrics();
// assign necessary fields

space.write(packer.pack(lyric), null, Lease.FOREVER);

At some point in time, you'll need to search the lyrics that are stored packed in a space. This code snippet shows how you'd do it using
Packer.packForTemplate(Object template), which packs the template before searching for it:

Lyrics templateLyrics = new Lyrics();
Entry[] entries = space.readMultiple(packer.packForTemplate(templateLyrics), null, 1000);

for (Entry entry:entries) {
  Lyrics lyrics = packer.unpack(entry);
  // process unpacked lyrics
}

For more examples, check the project documentation page.

Update: PackRat was submitted to the OpenSpaces Developer Challenge. It has won the Early Bird Draw.

April Fool's Day - Big OpenSpaces.org Release Date

2008-03-31T20:43:00.000-07:00

Late last year, our friends at GigaSpaces launched an open source community OpenSpaces.org with the following mission (quote from the community web site):

"OpenSpaces.org is a website sponsored by GigaSpaces Technologies and dedicated to creating and serving a community that develops and shares open source software for the OpenSpaces development framework."

For a company like ours this was a very good news. As an independent consulting firm with deep expertize in grid technologies and a whole practice dedicated to the popular GigaSpaces XAP technology, we constantly create tools, utilities and frameworks that make GigaSpaces product better and easier to use for our customers. Once the tool is created, we want to give it to the hands of all GigaSpaces users, and ideally let the users help us - and everyone else - to refine and improve the tool. OpenSpaces.org is a perfect venue for such projects.

Let's take an example - project Gigapult. If you ever had to write a bunch of scripts that start up a cluster on your Windows-based development environment, then re-write them to work on a corporate Linux-based testing lab and then do this again for a differently configured production environment, you know how frustrating and counter-productive this is. We had to do something like this last summer for a customer where the production environment wasn't even known to the developers who were creating the deployment scripts. This script-porting business led to hard-to-find bugs and delays in the production roll out.

Kirill Ishanov, the project lead, finally had enough and suggested an idea: let's make Gigaspaces configuration and bootstrapping work cross-platform, and in the process simplify the API and make the configuration files more maintainable and readable. After all, if everything else executes in the JVM, why should deployment logic be any different? Kirill explains his motivation in more details in his two blog postings, Project Gigapult and GigaSpaces Cluster Bootstrapping and Automated Testing, but this is basically how project Gigapult was born.

Today Gigapult is a JRuby-based domain-specific language designed to specify all aspects of grid configurations in a simple and intuitive way. For more information and to download the code, please, visit the project's homepage. We've used Gigapult internally on several projects now, and the technology does make life easier.

Now, along comes OpenSpaces.org and offers Gigapult the vehicle to release the technology into the hands of developers who need it most and can contribute to its enhancements going forward. Since the launch of OpenSpaces.org, Grid Dynamics opened three more active projects:

Convergence: a pluggable architecture for interoperability between computational grids and in-memory data grids capable of data-aware scheduling, with initial bindings for GigaSpaces XAP and Data Synapse GridServer

PackRat: a library that helps increase the space capacity by efficiently packing a non-indexed part of entities in binary format

OpenSpaces Binary Calculator: a tool to accurately estimate memory size required to store objects in cache

At the launch of OpenSpaces.org, GigaSpaces announced the OpenSpaces Developer Challenge. Naturally, our four projects entered the competition. And since the challenge ends today, tonight was the first beta release for all four! Nothing like a little competition to encourage the respect for deadlines... Despite today being April 1st, this is no joking matter. We want to win!

On more serious note, over the next few days I expect project leads to blog in more details about their respective projects. Meanwhile, please visit the projects, download the technology and give us the feedback - the good, the bad and the ugly so that we can make these tools better, for you and everyone else.

And please cheer for us as we await the results of the competition!

Notes from MS Developer Conference for Financial Services

2008-03-16T23:43:00.000-07:00

Last week a few folks from our company attended MS Developer Conference for Financial Services in Manhattan. The event was targeting Wall Street developers that came to learn what Microsoft is doing in the area of HPC. It is exciting to see Microsoft get serious about performance and scale-out computing and have a broad plan to address the needs of this already vast and rapidly expanding space.

I will not get into the details of technologies being showcased in this blog. Rather, I'd like to talk a little bit - selfishly - about what we were doing there, and then share one discovery - a grid company we didn't know about that has an interesting data caching solution for .NET community.

1. Excel-on-the-Grid

We were showing a demo of Excel application acting as a client for a grid-based processing. This is a powerful concept that is gaining wide spread on Wall Street. Excel is traditionally used in financial applications for heavy-duty data aggregation, analysis and visualization, all of which requires local data and processing on the desktop. This approach has serious limitations:

- If you wanted realtime data embedded in Excel app, how'd you deliver that data to the spreadsheet?
- If you wanted really computationally intensive spreadsheets, where'd you take the resources?
- If you wanted to share your application with others, how'd you do that?

Turns out, grid computing offers an elegant answer:

Step 1: Move the data from spreadsheet to Data Grid
Step 2: Move the computations from spreadsheet to Compute Grid
Step 3: Instrument Excel to both push the computations to the Grid and pull the results off the Grid as appropriate.
Step 4: Share the "dumb client" easily, it's now light-weight

Many grid vendors - and their Excel customers - are recognizing the business benefits and building solutions that makes Excel-on-the-Grid work better, ideally someday out-of-the-box. Amongst such vendors are Microsoft, GigaSpaces, Data Synapse, and others. We at Grid Dynamics were asked to develop a demo that showed Microsoft Windows Compute Cluster Server 2008 as a computational engine behind Excel-on-the-Grid, backed by GigaSpaces XAP 6.0 as in-memory data grid.

The demo works as follows:

1. Excel sends the job to MS WCCS
2. The job is directed to & executed on the server that runs the right local data cache by XAP
3. Excel gets a notification when the job is finished and displays the results

The main points of the demo are three-fold:

a. Excel-on-the-Grid is COOL
b. One can mix-and-match various compute and data grid solutions to achieve business objectives
c. Data-aware job scheduling (which is the ability of job scheduler to co-locate job execution local to the cached data) gives 2x - 3x performance boost to data-hungry jobs. What's especially important to point out is that the job itself is a "black box" to the grid - no changes are done to the algorithm itself, it may be considered a legacy code for all intents and purposes.

2. ScaleOut Software

An interesting discovery of the conference for our team had been a vendor, ScaleOut Software, who has a pure .NET implementation of data grid with seemingly good architecture and nice features.

Traditionally, Java community had been way ahead of the game on Data Grids. GigaSpaces, Tangosol and GemStone - the three big mature commercial players in data grid space - are all Java-based technologies. Yet the applications that require data grid access are a good mix between Java and .NET these days (there is always some C/C++ and other varieties, of course). While these Java technologies are supporting .NET clients in various ways, it makes sense for pureplay .NET caching and data grid solutions to be available on the market as well.

We weren't aware of such solutions when we ran into ScaleOut Software last week. They offer replicated data grid with a lot of nice features. They don't have two big ticket items that leading Java caching vendors do: data partitioning and strong out-of-the-box DB persistence. Still, for application session data, and many other stateful application needs in Windows environments, I can see ScaleOut providing a good solution for developers. We haven't really had a chance to do anything with ScaleOut software yet, but I hope to try it out soon.

DataGridFS - let your legacy code store data into IMDG

2008-03-10T04:12:00.000-07:00

The one of the most discussed topics on grid computing here in Grid Dynamics is bringing Computational Grids and In Memory Data Grids together. And DataGridFS concept was born in such discussions.

First of all, let me describe where it applies most efficiently. There are a lot of systems, which are built around Computational Grid where jobs produce files as result of computations. These files stored to NFS, so they are accessible locally after mounting on every node and client. Sun Grid Engine and Platform LSF are examples of such Computational Grids. Imagine that such system needs enhancement and such enhancement is based on bringing IMDG in this system. So, the codebase will consist of legacy code which saves results to files and modern code which uses IMDG API to get initial values and store results too. More over, these two parts should communicate with each other somehow. For example, results calculated by legacy code are initial values for new codebase jobs. It is clear, that we need the way to make new jobs able to process results stored in files on file system. There are several ways to achieve that. One is to build daemon process which scans directory on filesystem with results, parse them and stores data into IMDG using it's provided API. This approach has few cons. At least you should be able that this process is always alive and so on. The second way which is to use DataGridFS.

DataGridFS is filesystem which stores files transparently in IMDG itself. So, this filesystem inherits most features of used IMDG. For example, if IMDG allows partitioning then filesystem automatically becomes distributed and so on. So, when job process writes file to filesystem object (or set of objects) appears in IMDG. This object can be accessed using regular IMDG API or DataGridFS API which just "wrappers" for IMDG API calls. "What is about file content?", you may ask, "Is it still to be parsed in object model?" Yes, but this can be done by workers inside IMDG, which is more convenient and flexible way than separate process.

I've managed to code very simple prototype to illustrate idea. It is just read-only filesystem built using FUSE and FUSE-J. File data stored with GigaSpaces XAP. On the screenshot you can see space content via Space Browser, and the same content via UNIX command line utils.

(Click to enlarge)

GigaSpaces Cluster Bootstrapping and Automated Testing

2008-03-03T01:15:00.000-08:00

In projects which use GigaSpaces XAP it's often needed to have a set of functional tests which verify the interaction of your code and GigaSpaces cluster. For example, if an application makes initial data loading it's good to have a test that verifies that the client's code really found a space and loaded all necessary data.

To implement such tests we need to bootstrap a cluster with an appropriate configuration and to run a set of tests for this cluster. It can be achieved by following the next steps:

Use a testing framework which allows to run some set up code before running group of tests (looks like TestNG is the most appropriate choice for it cause it allows to group tests and has a sexy @BeforeGroups annotation)

Put all the code which starts a cluster to the set up method (kinda of prerequisites for tests)

Implement a functional test which uses the standard assertion mechanism to verify the cluster's state

This approach looks very useful, but the 2nd point is very tricky cause GigaSpaces (5.* and 6.0*) doesn't provide a mechanism, which will allow to start a cluster from code without black magic and dances with tambourine.

Here is a tricky workaround (which we've used on a Convergence project for testing purposes). It assumes the usage of Maven as a build tool (sorry Maven folks, I know that it's project management framework, but this definition sounds too mysterious :) ) and the project is a multi-module.

The idea behind it is very simple: to perform a cluster bootstrapping via Maven plugin before testing phase. I haven't found the appropriate plugin in Maven's repository so I wrote my own using Gigapult (check the usage and installation instructions on OpenSpaces.org).

Such approach allows to write the functional tests using a testing framework of your choice without putting cluster configuration and bootstrapping code into the setup methods. Moreover, it simplifies the creation of different build configurations on Continuous Integration server, cause it turns the configuration to the simple enumeration of Maven goals.

But putting the cluster bootstrapping to the build lifecycle and a functional tests, which can take much more time, then an impatient developer can wait, leads to the skipping of these phases. To avoid it, all time-consuming tests can be put to the separate module and run on demand using Maven's profiles. To create such configuration add a submodule to the Maven project (it can be done using the Maven Archetype Plugin) and adding the following section to the parent pom.xml:

<profiles>
  <profile>
    <id>functional-tests</id>
    <modules>
      <module>functional-tests</module>
    </modules>
  </profile>
</profiles>

After that, Maven will add the sub-module called "functional-tests" only if -Pfunctinal-tests was specified. Assuming that CI server supports Maven now we can specify goals for two build configurations:

clean install - for running all unit test and packaging

gigapult:gsm gigapult:gsc gigapult:pudeploy install -Pfunctional-tests - for running all unit tests, packaging, bootstrapping cluster and running functional tests (gigapult goals depend on cluster configuration)