Grid Designer's Blog

Ignoring test failures at CI

2012-05-23T13:30:00.000-07:00

Problem

Today, I want to talk about one obvious issue which occurs when your code and tests are written by the different people and you have a properly established CI process. In this scenario, tests appear in VCS first, because without a test the code can’t be committed, but this test breaks the build. Obviously, test author skips it or disables it from CI test suite until this functionality is actually implemented. I believe that this is not the best practice, because it leads to a lot of wasted efforts from both dev and QA teams. Let’s look at what’s going on in typical project team:

Commit

Developer checks out the disabled test for his new feature from VCS, runs it, and checks that the code is ready. Now he’s able to commit the code. He does it, but unfortunately makes two usual mistakes: he performs one-line-improvement right before commit without full test run (of course, what can possibly go wrong?); and he forgets to commit one property file. Doesn’t it happen to you? Really?

Test

Then he pings the test’s author, - Hey! I’ve done it. Check your test pls. Tester needs to run a private build to verify that all code changes have been committed, and then enables the test for CI. If tester skips the private build phase, which can include pulling snapshot or build and creating deployment, you’ll have CI failure without genuine code issue.
What it should be
Just free tester from the unnecessary private build and testing cycle. Let CI work for you. Don’t skip test from CI suite, let it run and fail without impacting the build status. Tester sets the expectation for functionality, and let developers reach it, without a necessity to verify an uncommitted code.
Tooling
JUnit has only one workaround - assumeTrue, which allows to mark failed test as skipped. To introduce a fancier way you need to write your Runner.

TeamCity has nice looking “mute” feature. Nice work, JetBrains!

TestNG has something along these lines. It works fine, but surefire doesn’t support it doesn’t support it.

JBehave has ignoreFailureInView property, but it belongs to whole suite not for paticular test.

Credits
I realized that some time ago but decide to post after found the same concern in the mail thread.

Highlights from our OpenStack-specific blog

2011-10-25T10:28:00.000-07:00

A few highlights from our OpenStack blog:

How our RHEL port is built and tested, using Gear + Mock

Setting up VPN for projects with CloudPipe

Using SSH keys with novaclient

Improving novaclient CLI: boot a server with specific SSH key

Also, for those who do not follow our OpenStack blog closely, OpenStack Diablo release is out for both RHEL and CentOS (version 6, as usual). Sources are available as well.

Solr Experience: search parent-child relations. Part III

2011-10-03T01:16:00.001-07:00

In the previous posts (part I and part II) I presented challenges in searching parent-child relations by Solr, a couple of workarounds and handsome final solution - SpanQueries.

In this post I will describe the infrastructure for SpanQueries we’ve built inside Solr . It includes indexing, query parsing (creating SpanQueries), caching filter queries, faceting and sorting.

Indexing

Fist of all I should say that we use SolrJ - embedded Solr server, and this section is about adding documents into Solr by its’ Java API only.

If you put a collection of values into SolrInputDocument.addField(“field”, values) with termPositions=”true” in schema.xml, Solr will interpret them as several different terms for one field and for each value it sets position to 0.

The straightforward approach is adding all values as one text string (e. g. “red blue green”) and using analyzer (StandardAnalyzer works fine), which adds proper term positions into the index. In this case, if we want to index several numbers, we have to add them as one string, e. g. “123.45 42.15 98.27”. It’s not useful, so we implemented field types for String, Integer, Double and Float. They create Field instance with special TokenStream, which uses List as a source and saves positions (indexes in List).

Parsing SpanQueries

By default Solr uses LuceneQParser plugin to parse user queries and create Lucene Queries. We created a wrapper for it - SpanQParserPlugin - which transforms regular BooleanQueries and TermQueries into SpanQueries. Then we registered it in solrconfig.xml:
    <queryParser name="span" class="com.griddynamics.solr.spans.SpanQParserPlugin"/>

This plugin transforms BooleanQuery with AND clause into SpanNearQuery(...) and BooleanQuery with OR clause into SpanOrQuery.
As result the query ”fq={!span tag=spanFQ}+color:Red +size:XL” is parsed into:
SpanNearQuery(
    SpanTermQuery("color", "Red”),
    FieldMaskingSpanQuery(
              SpanTermQuery("size", "XL"),
              “color”
    ), -1, false
)

Here {!span} is defType local param. It says to Solr to use SpanQParserPlugin and spanFQ is a tag we will use for faceting later.
Please be informed that there are a couple of better ways to extend query parsing.

Filter queries caching

Solr caches docSets (materialized FilterQuery results, which are intersected on searching) and docLists (ranked query results that store document lists). But they are not enough for us, because for faceting we need spans, which passed the filter.

We’ve created SpanQueryFilterCache for this - it’s similar to SpanFilterResult but uses primitives and arrays instead wrappers and collections. It stores spans which were found by SpanQuery in the user cache provided by Solr. To get spans we call SpanQueryFilterCache.getSpans(spanQuery, solrIndexSearcher) that initializes the cache lazily and gets spans by spanQuery key.

Faceting

I suppose it’s the most interesting part of this article.
Standard field faceting mechanism calculates counts of documents that contain field value. It doesn’t fit for us, because we need to count facets on the matched spans only (parts of a document).

To count span facets we made a special component SpanFacetComponent, which is activated with the parameter span.facet.field.

The query for searching Red sweaters and count facets for size and color fields looks like this:

q={!span tag=spanFQ}+color:Red&facet=true&span.facet.field={!by=spanFQ aggOp=unique}color&span.facet.field={!by=spanFQ aggOp=unique}size

And we have a Product, which passes the filter:

Only two of its’ Items pass the filter and must be counted for faceting. The tag “spanFQ” (see above) gets SpanQuery instance and uses SpanQueryFilterCache to enumerate matched spans. It works for color and size fields, because span doesn’t have a field and we can count size facet by spans matched by color filter. For counting facets we need field values on matched spans. We get from the use forward view of data - UnInvertedSpans (it’s allusion to UnInvertedField; it stores mapping {(docId, spanNum) -> {termNum}}).

We have two Items, which passed the filter, but we have a constraint that a single document increases facet count for a value only once i.e. document gives a set of values (not a bag) for facet counts. The match above gives {Red,XL,XXL}. To determine which span should be counted we use aggOp (aggregation operation) parameter that can be one of three values:

unique - each unique value increases facet count;
first - only first span value increases facet count;
last - only last span value increases facet count.

The last two operations are used for calculating min/max price (here we use natural ordering enforced during indexing, instead honest query time min/max). It’s worth to implement min/max operations also.

Sorting

Our Product can have several prices. To sort by price we need to use different values depending on sorting direction: for ascending - min price, for descending - max price.
The solution is store prices aligned by positions and SpanQueries to get a price values for sorting. During indexing we put prices in the Product document in ascending order. In query time we use SpanQuery and UnInvertedSpans to get spans with prices and then our MultiValueFieldComparatorSource uses value from first (on asc) or last (on desc) span for sorting. For example, our query with additional price filter and sorting by price field will be:

q={!span tag=spanFQ}+color:Red

&fq={!span tag=priceFQ}+price:[5 TO 15]

&sort.field={!bySpanFQ=priceFQ}price ASC

Conclusion

That’s all. We hope our experience and ideas will help someone to resolve his or her issues about parent-child relations. In future we want to try storing two or more terms for one field in one position - by this way we want to handle the same issues with multi-valued field.
We also work on deeper understanding of the problem and building a powerful model of nested documents on inverted lists. One of the last findings is that ‘false positive matching on multi-value fields’ is well-known Fan Trap in ER-modelling.

There is a considerable alternative approach LUCENE-3171. It’s a little bit raw at the moment, but the idea is promising.

Solr Experience: search parent-child relations. Part II

2011-07-27T23:32:00.001-07:00

In the previous post I presented an e-Commerce site that sells sweaters. I showed how we tried to implement Faceted Navigation with Solr, found the problem with the multi-valued field for searching parent-child relations. I also proposed a workaround that solves the problem but that is far from perfect.

In this post I’ll describe the ultimate solution - SpanQueries.

SpanQuery

SpanQueries are purposed for proximity search, which considers terms’ positions inside text. Lucene can index terms’ positions in addition to terms. For example, “crazy cat and crazy dog” gives us the following inverted index:

Word “and” is excluded as stop word.

SpanQuery considers that terms’ positions during searching. Span - is a term position range or tuple that contains document ID and start and end positions. A term can have several spans per document. There is a basement method SpanQuery.getSpans(...), which returns iterator for matched documents and spans.

SpanQuery - is an abstract class, there are several its descendants. You can find their description in the post in the Lucid Imagination blog and "Lucene In Action" book.

We are interested in three of them:

SpanTermQuery - finds all spans by the term;
SpanNearQuery - combines several SpanQueries considering the distance between matched spans;
FieldMaskingSpanQuery - masks SpanQuery by one field as query by another field.

Now let’s take a look at how do we use this SpanQueries to search parent-child relations.

Parent-child relations with SpanQueries

We index Products in a way when attributes of each Item are located on the same position:

Combining the SpanTermQueries with FiledMaskingSpanQuery and putting them inside SpanNearQuery allows searching the intersections of the terms' positions, and properly find the Product, which contains the specified Item (color:Red and size:XL):
SpanNearQuery(
    SpanTermQuery("color", "Red”),
    FieldMaskingSpanQuery(
               SpanTermQuery("size", "XL"),
               “color”
    ), -1, false
)

This query returns the only documents, which have color:Red and size:XL at the same position i.e. belonging to the same Item:

The third product (Style&co.) is not returned, because its' terms "Red" and "XL" are located in different positions.

We've got what we want out-of-the box. But there is another problem: Solr doesn't have any infrastructure for supporting SpanQueries. In the next post I will tell about how do we bring SpanQueries into the Solr, including interesting issues of caching, faceting and sorting.

P. S.: recently we found the similar approach had been proposed and ongoing work in Lucene core (see LUCENE-2454, LUCENE-3133, LUCENE-3171). Looking forward for incorporating this feature in Solr.

Spring Nested - Part III

2011-07-02T11:26:00.000-07:00

In the previous posts I, II I told you how we came up with our own Service Locator framework based on Spring. Frankly speaking, a root application context creates child application contexts and keeps a services registry, where the child contexts advertise and look up service beans by names. Today I will tell you about two vital features.

Handling Web Requests

We use Spring MVC to expose platform services via Web. To handle Web requests WAR should has handlers or controllers declared in Spring application context. In our platform all business logic resides in component contexts aka child contexts. Therefore, a component developer declares request handlers and controllers in component context, but Spring MVC’s DispatcherServlet looks for request handlers in the root context only. So, we need to pass the request to these controller/handler beans declared in the target child context. We went through three approaches below one by one:

the first was really explicit: child contexts publish their Web request handlers to the service registry. Root context contains a special request handlers, every of which delegates a request to a handler obtained from service registry by the specified name. So, we had to amend the root config xml when a new web handler appears in any component;
then, we introduced the handler mapping registry in the root context. Child contexts have factory beans register the child request handlers in that registry. Thus, the registry can handle a request by looking up a child handler by an URL and delegating request to it. That’s better than the first approach because we don’t need to update the root config xml on introducing a new request handler. But it was too far from non-invasive approach, i.e. we had to consider these declarations when add a new handler;
the final approach is completely implicit: root context has the same child handler mapping registry (url -> handler), but it’s populated automatically during scanning the child contexts for handler/controller beans. Then, this mapping registry handles requests by delegating them to particular child handlers found by an url. As result, component developer doesn’t care about the framework when he delivers a new Web request handler.

In contrast to Spring DM way we have a single WAR with the single DispatcherServlet contains the several Web components.

Component Initialization and Dependencies

It’s the most feature I am proud of. To instantiate all child contexts we need to know a correct initialization sequence - which child context goes after which. Initially we relied on the list of config XMLs explicitly provided in the root context. Then, we found problems with this straightforward approach, e.g.

absence or fail of one export breaks the related lookup;
a looked up bean (which is actually a proxy) can be accessed before its’ context initialization.

We decided to implement Dependency Analyzer that verifies dependencies across child contexts before any child context initialization.

How have we done it? We rejected the idea of parsing Spring context XMLs to analyze dependencies, just because we had two XML formats already. Spring has quite extensible and elegant design: BeanDefinitionReader parses XML and passes BeanDefinitions to BeanDefinitionRegistry. For default flow BeanDefinitionRegistry is an ApplicationContext, which creates singleton bean instances etc. But we can run this routine against a fake one: SimpleBeanDefinitionRegistry that just keeps bean definitions in a map . Then analyzer can iterate through bean definitions, recognize exports and lookups, verify it, raise concerns and/or arrange initialization order. Here is the list of rules which analyzer enforces now:

verifies that an every bean class is exist in classpath;
warns unnecessary exports (without any lookup);
prohibits imports without a matching export;
prohibit dependency cycles between contexts (a but allows to resolve them manually in “god mode”);
orders child context initialization considering exports before to imports;
finds a minimal list of the components which are required for running the given sub-set of services (transitively finds the dependent contexts);
skips the dependent contexts after a some context initialization failure;
outputs the derived component dependencies graph into debug log in Graph Viz format.

Some of these features just spot the trivial errors and reduce the number of log lines to look through. Remaining features provide a flexible and robust container for our platform.

Finally I’d like to spot a significant difference between Spring DM and Spring Nested. Spring DM has an asynchronous ad-hoc initialization which is too complex and inconvenient sometimes. Spring Nested proposes a single thread deterministic initialization. It’s rigid enough - prohibits cycles, but flexible - you code against services, don’t care about a component initialization sequence. And it’s developer friendly: you shouldn't change you Spring habits much; troubleshooting is easy as possible with the reduced logs and the preliminary checks.

If you accept some simplifications, which has been made in Spring Nested design this framework can help you with your large Spring application.

The Real Life Application

Here is the modules dependencies graph of our application that is built with Spring Nested. This graph is derived from the services references by the dependency analyzer and logged in Graph Viz text format on startup.

I smudged the component names at this picture just to avoid NDA aspects.

Further Plans

We want to open its’ source. Please let us know if you are interested in it or leave a valuable feedback and ideas.

Solr Experience: search parent-child relations. Part I

2011-06-06T07:15:00.000-07:00

Typical e-Commerce sites provide a catalog of products with filtration (or search by criteria). We implement such catalog with Solr search server based on Apache Lucene library.

A catalog entries usually have parent-child relations and search must consider these relations. Unfortunately Solr documents are flat and some straightforward approaches e.g. several indexes, denormalization have been rejected because of efficiency and implementation efforts.

In this post I describe the problem and some possible approaches.

E-Commerce site

Product is the model of any goods which are sold on the site. A product has attributes like Brand, Occasion etc. You can’t buy a Product, you can buy an Item - a concrete instance of the Product, which has such attributes like Color, Size, Length, etc. Also every Item shares Product’s attributes.

For example, if you search "Nizza Hi" on www.shopadidas.com site, you see different sneakers of the single model ("Men's Originals Nizza Hi Shoes"). They have different colors, therefore they are Items. Then search the same on Macys.com - you see the only pair of sneakers. If you look to its' page, you see that there are several colors available. The Macys.com search result contains Product, not Items.

Let's have a site, where we sell sweaters with different colors and sizes (Items). There are three brands and each brand is presented as separate Product. The list of all available Items:

Alfani:
- Red - XL
- Red - XXL
- Blue - XL
- Green - L
- Green - XXL
Calvin Klein:
- Red - XL
- Blue - M
- Blue - L
- Green - M
- Green - XL
Style&co.:
- Red - M
- Blue - XXL
- Blue - XL
- Green - S

To provide a better usability we need to show three Products instead of twelve Items, and allow to filter them by color and size i.e. to provide Faceted Navigation.

For example, a client need Red shoes with XL size. Therefore, the site must show Products that contain Items with color:Red and size:XL only:

Alfani:
- Red - XL
- Red - XXL
- Blue - XL
- Green - L
- Green - XXL
Calvin Klein:
- Red - XL
- Blue - M
- Blue - L
- Green - M
- Green - XL
Style&co.:
- Red - M
- Blue - XXL
- Blue - XL
- Green - S

I’ve highlighted with bold font Items and Products passed the filter. The third Product (Style&co.) doesn’t match the filter. Although it has Items with Red color and XL size, but it doesn’t contain any Item with both Red color and XL size we are looking for.

Attemt #1: Two indexes

The first intention is using Solr as database: to create separate indexes for Items and Products. Thus we can get the result with two Solr-queries:

find all Items passed the filter;
find all Products that contain Items we got on the previous step.

The logic seems working, but further investigation has shown two problems:

Pagination - usually we want to show not the whole result, but only 10, 30, 50, or 100 Products. But it's very expensive to query all Items (number of them can be huge) and than pass them inside the second query for getting first N Products;
Faceting - we have to write a lot of custom code around Solr to calculate all facet counts from the result of the first query. Also the first query result must contain product ID, which is expensive.

Thus this approach was rejected.

Attempt #2: Denormalization

Denormalization is creating the only index for Items. It's similar to the previous approach, but there are two differences:

productID field is not stored field, because we need it only for querying;
Faceting requires custom code too, but it can be placed inside Solr within own Components. They use Solr/Lucine internal data structures and much faster than post-processing the result.

There is another problem - result collapsing. We need Products as final result, not Items. Solr’s FiledCollapsing can be used for this but it has a little bit different purpose. We ended up with our own collapser.

Thid approach is not flexible: small requirements changes impact the codebase a lot.
We've got that the document granularity should match the search result.

Attempt #3: Item Attributes' Promotion

Then we came back to the Product index. The question was how to index Items inside Product documents with keeping their identity for filtering. Here we come to the attributes’ promotion approach: item’s attributes (sub-documents) are promoted into the multi-value document fields.
But doing it straightforward we met the false positive multi-value field match problem. Let's look to Lucene documents:

I highlighted with bold font the result for query color:Red and size:XL. The third Product “Style&co.” is presented in the result table, but it must be filtered out (see our example above), because it doesn’t contain any Items with both Red color and XL size we are looking for.

After we realized this problem, we moved to...

Attempt #4: Term Encoding

The solution was quite brave - we solved it with an accurate term encoding. We introduced a composite terms for keeping an Item identity. For example:

Alfani:

Red - XL : RedXL
Red - XXL : RedXXL
Blue - XL : BlueXL
Green - L : GreenL
Green - XXL : GreenXXL

As result you can find a document by an item attributes combination and avoid a false match for absent items like RedXL. We also had to generate all subsets of attribute values. Thus the full list of generated terms for our example looks like:

Red, XL, RedXL, XXL, RedXXL, Blue, BlueXL, Green, L, GreenL, GreenXXL.

The findability is solved (because Style&co. doesn't have term RedXL, which we are looking for), but we still needed a proper faceting. Honestly speaking we faceted such composite fields with Tokenizer and HashMap (a kind of brute force).

We knew that this approach is far from perfect due to performance and scalability issues and the complex processing in query time. That's why we continued to find a more robust approach.

Part I: Conclusion

Keep in mind that the described issue takes place with almost any case, when your search result contains the documents with any form of hierarchy. We actually met two instances of that:

product document has an item sub-documents (which is not presented as a separate documents itself);
product documents are grouped by another documents reside in the index;
document has associations with another entity encoded by a term (it seems like relation sub-document).

In the next part I will introduce the silver bullet - SpanQueries.

Understanding GC pauses in JVM, HotSpot's CMS collector.

2011-06-02T19:09:00.000-07:00

(previous artictle "Understanding GC pauses in JVM, HotSpot's minor GC")

Concurrent Mark Sweep (CMS) is one of HotSpot JVM low pause garbage collectors. CMS can do most of its work for reclaiming memory concurrently with application (without stopping it). But still it requires few stop-the-world pauses to make its work. This article will explain nature of these pauses and how to minimize them.

Basics of concurrent mark sweep

HotSpot’s CMS is a generational collector, it means that heap is separated into young and old (tenured) space and these spaces are collected independently. For young space collection usual HotSpot’s copy collector is use (see previous article about HotSpot’s young space collector). Concurrent Mark Sweep is used only to collect old space. To enable of using CMS collector you have to specify –XX:+UseConcMarkSweepGC in JVM’s command line.

CMS collection cycle has following phases:

Initial mark – this is stop-the-world phase while CMS is collecting root references.
Concurrent mark – this phase is done concurrently with application, garbage collector traverses though object graph in old space marking live objects.
Concurrent pre clean – this is another concurrent phase, basically it is another mark phase which will try to account references changed during previous mark phase. Main reason for this phase is reduce time of stop-the-world remark phase.
Remark – once concurrent mark is finished, garbage collector need one more stop-the-world pause to account references which have been changed during concurrent mark phase.
Concurrent sweep – garbage collector will scan through whole old space and reclaim space occupied by unreachable objects.
Concurrent reset – after CMS cycle is finished, some structures have to be reset before next cycle can start.

Unlike most other garbage collectors, CMS does not do compaction of heap space. Instead of moving objects to make unoccupied space continuous, CMS keeps lists of all fragments of free memory. This way CMS is avoiding cost associated with relocating of live objects (and relocating of objects is expensive operation which require stop-the-world pause), but as down size of this heap space is prone to fragmentation. To minimize risk of fragmentation CMS is doing statistical analysis of object’s sizes and have separate free lists for objects of different sizes.

Length of CMS pauses

CMS itself has only two pauses, but your application will also experience pauses of young space collector which is working in conjunction with CMS. See previous article about pauses of young space collector.

Initial mark

During initial mark CMS should collect all root references to start marking of old space. This includes:

References from thread stacks,
References from young space.

References from stacks are usually collected very quickly (less than 1ms), but time to collect references from young space depends on size of objects in young space. Normally initial mark starts right after young space collection, so Eden space is empty and only live objects are in one of survivor space. Survivor space is usually small and initial mark after young space collection often takes less than millisecond. But if initial mark is started when Eden is full it may take quite long (usually longer than young space collection itself).

Once CMS collection is triggered, JVM may wait some time for young collection to happen before it will start initial marking. JVM configuration option –XX:CMSWaitDuration= can be used to set how long CMS will wait for young space collection before start of initial marking. If you want to avoid long initial marking pauses, you should configure this time to be longer than typical period of young collections in your application.

Remark

Most of marking is done in parallel with application, but it may not be accurate because application may modify object graph during marking. When concurrent marking is finished; garbage collector should stop application and repeat marking to be sure that all reachable objects marked as alive. But collector doesn’t have to traverse through whole object graph; it should traverse only reference modified since start of marking (actually since start pre clean phase). Card table (see card marking write barrier) is used to identify modified portions of memory in old space, but thread stacks and young space should be scanned once again.

Usually most time of remark phase is spent of scanning young space. This time will be much shorter if we collect garbage in young space before starting of remark. We can instruct JVM to always force young space collection before CMS remark. Use JVM parameter –XX:+CMSScavengeBeforeRemark to enable this option.

Even is young space is empty, remark phase still have to scan through modified references in old space, this usually takes time close to normal young collection pause (due scanning of old space done during young collection is similar to scanning required for remark).

When CMS collection starts?

Unlike stop-the-world old space collectors, CMS collection cycle should start before old space become full. CMS collection is triggered when amount of free memory in old space falls below certain threshold (this threshold can be chosen by JVM based of runtime statistics or set via parameters) and actual start of CMS collection cycle may be delayed until next young collection.

Normally objects are allocated in old space only during young space collection (which may promote some objects to old space). So CMS cycle usually starts right after young space collection, which is good because init mark pause will be very small.

But in certain cases object may be allocated directly in old space and CMS cycle could start while Eden has lots of objects. In this case initial mark can be 10-100 times slower which is bad. Usually this is happening due to allocation of very large objects (few megabyte arrays). To avoid these long pauses you should configure reasonable –XX:CMSWaitDuration.

Configuring fixed threshold for CMS start

You can set fixed threshold for olds space occupation for triggering CMS cycle by using JVM options ‑XX:+UseCMSInitiatingOccupancyOnly ‑XX:CMSInitiatingOccupancyFraction=70 (this will force CMS cycle to start when more than 70% of old space is used).

Explicitly invoking CMS cycle

You can also configure JVM to start CMS cycle by invocation of System.gc() by ‑XX:+ExplicitGCInvokesConcurrent command line option.

Full GC with CMS

If CMS cannot free enough in old space, JVM may fallback to compacting collector. Compacting collector will force stop-the-world pause so it can be considered emergency case. Normally you would like to avoid full GC and long stop-the-world pause associated with it. Full GC may happen either if CMS is not fast enough for dealing with garbage (or collection cycle has been started too late) or due to fragmentation of old space (there is no large enough continuous space for object to be allocated). Also it is possible that you just didn’t give JVM enough memory and after full GC it will through OutOfMemoryExpection anyway.

Permanent generation collection

One of reasons why CMS may end up in full GC is garbage in permanent space. By default CMS does not reclaim unused space in permanent space. If your application is using multiple class loaders and/or reflection you may need to enable collecting of garbage in permanent space. JVM option ‑XX:+CMSClassUnloadingEnabled will allow CMS collector to clean permanent space. Remember that objects in permanent space may have references to normal old space thus even if permanent space is not full itself, references from perm to old space may keep some dead objects unreachable for CMS if class unloading is not enabled.

Utilizing multiple cores

CMS has multiple phases. Some of them are concurrent; others are stop-the-world pauses but may be executed in parallel to compressed application freeze time.

‑XX:+CMSConcurrentMTEnabled – allows CMS to use multiple cores for concurrent phase.

‑XX:+ConcGCThreads= – specifies number of thread for concurrent phases.

‑XX:+ParallelGCThreads= – specifies number of thread for parallel work during stop-the-world pauses (by default it equals to number of physical cores).

‑XX:+UseParNewGC – instructs JVM to use parallel collector for young space collections in conjunction with CMS.

Comming soon:

Tuning CMS collector for large IMDG storage nodes

Understanding GC pauses in JVM, HotSpot's minor GC.

2011-06-01T03:04:00.000-07:00

Stop-the-world pauses of JVM due to the work of garbage collector are known foes of java-based applications. HotSpot JVM has a set of very advanced and tunable garbage collectors, but to find optimal configuration it is very important to understand an exact mechanics of garbage collection algorithms. This article is first of the series explaining how exactly GC spends our precious CPU cycles during stop-the-world pauses. An algorithm for young space garbage collection in HotSpot is explained in this post.

Structure of heap
Most of modern GCs are generational. That means that java heap memory is separated into few "spaces". Spaces are usually distinguished by “age” of resident objects. Objects are allocated in young space, then eventually copied to old (or tenured) space, if they survive long enough. This principle is based on hypothesis that most object “die young”, i.e. majority of objects become garbage shortly after being allocated. All HotSpot garbage collectors separate memory into 5 spaces (though for G1 collector spaces may be not continuous).

• Eden are space there objects are allocated,
• Survivor spaces are used to receive object during young (or minor GC),
• Tenured space is for long lived objects,
• Permanent space is for JVM own objects (like classes and JITed code), it is behaves just like tenured space so we will ignore it for rest of article.
Eden and 2 survivor spaces together are called young space.

HotSpot GC algorithms overview

HotSpot JVM is implementing few algorithms for GC which are combined in few possible GC profiles.

• Serial generational collector (-XX:+UseSerialGC).

• Parallel for young space, serial for old space generational collector (-XX:+UseParallelGC).

• Parallel young and old space generational collector (-XX:+UseParallelOldGC).

• Concurrent mark sweep with serial young space collector (-XX:+UseConcMarkSweepGC

–XX:-UseParNewGC).

• Concurrent mark sweep with parallel young space collector (-XX:+UseConcMarkSweepGC
–XX:+UseParNewGC).

• G1 garbage collector (-XX:+UseG1GC).

All profiles except G1 are using almost same young space collection algorithms (with serial vs parallel variations).

Write barrier

Key point of generational GC is what it does need to collect entire heap each time, but just portion of it (e.g. young space). But to achieve this JVM have to implement special machinery called “write barrier”. There 2 types of write barriers implemented in HotSpot: dirty cards and snapshot-at-the-beginning (SATB). SATB write barrier is used in G1 algorithms (which is not covered in this article). All other algorithms are using dirty cards.

Dirty cards write barrier

Principle of dirty card write-barrier is very simple. Each time when program modifies reference in memory, it should mark modified memory page as dirty. There is a special card table in JVM and each 512 byte page of memory has associated byte in card table.

Young space collection algorithm

Almost all new objects (there are few exception when new object can be allocated directly in old space) are allocated in Eden space. To be more effective HotSpot is using thread local allocation blocks (TLAB) for allocation of new objects, but TLAB themselves are allocated in Eden. Once Eden becomes full minor GC is triggered. Goal of minor GC is to clear fresh garbage in Eden space. Copy-collection algorithm is used (live objects are copied to another space, and then whole space is marked as free memory). But before start collecting live objects, JVM should find all root references. Root references for minor GC are references from stack and all references from old space.

Normally collection of all reference from old space will require scanning through all objects in old space. That is why we need write-barrier. All objects in young space have been created (or relocated) since last reset of write-barrier, so non-dirty pages cannot have references into young space. This means we can scan only objects in dirty pages.

Once initial reference set is collected, dirty cards are reset and JVM starts coping of live objects from Eden and one of survivor spaces into other survivor space. JVM only need to spend time on live objects. Relocating of object also requires updating of references pointing to it.

Finally we have Eden and one survivor space clean (and ready for allocation) and one survivor space filled with objects.

Object promotion

If object is not cleared during young GC it will be eventually copied (promoted) to old space. Promotion occurs in following situations:

• -XX:+AlwaysTenure makes JVM to promote objects directly to old space instead of survivor space (survivor spaces are not used in this case).

• once survivor space is full, all remaining live object are relocated directly to old space.

• If object has survived certain number of young space collections, it will be promoted to old space (required number of collections can be adjusted using –XX:MaxTenuringThreshold option and
–XX:TargetSurvivorRatio JVM options).

Allocation of new objects in old space

It would be beneficial if we could possibly allocate long lived objects directly in old space. Unfortunately there is no way to instruct JVM to do this for particular object. But there are few cases when object can be allocated directly in old space.

• Option -XX:PretenureSizeThreshold=n instructs JVM what all objects larger than n bytes should be allocated directly in old space (though if object size fits TLAB, JVM will allocate it in TLAB and thus young space, so you should also limit TLAB size).

• If object is larger than size of Eden space it also will be allocated in old space.

Unlike application objects, system objects are always allocated by JVM directly in permanent space.

Parallel execution

Most of task during young space collection can be done in parallel. If there are several CPUs available, JVM can utilize them to compress duration of stop-the-world pause during collection. Number of threads can be configured in HotSpot JVM by –XX:ParallelGCTreads=n parameter. By default JVM will choose number of thread by number of available CPU. As expected, serial version of collector will ignore this parameter because it can use only one CPU. Using parallel collection reduces time of stop-the-world pause by factor close to number of physical cores.

Measuring stop-the-world pause for young collection

Young space collection happens during stop-the-world pause (all non-GC-related threads in JVM are suspended). Wall clock time of stop-the-world pause is very important factor for applications (especially applications requiring fast response time). Parallel execution affects wall clock time of pause but not work effort to be done.

Let’s summarize components of young GC pause. Total pause time can be written as:

T_young = T_{stack_scan} + T_{card_scan} + T_old__scan+ T_copy ; there T_young is total time of young GC pause, T_{stack_scan} is time to scan root in stacks, T_old__scan is time to scan roots in old space and T_copy is time to copy live objects (1).

Thread stack are usually very small, so major factors affecting time of young GC is T_old__scan and T_copy.

Another important parameter is frequency of young GC. Period between young collections is mainly determined by application allocation rate (bytes per second) and size of Eden space.

P_young = S_eden / R_alloc; there P_young is period between young GC, S_eden is size of Eden and R_alloc is rate of memory allocations (bytes per second) (2).

T_{stack_scan}– can be considered application specific constant.
T_{card_scan}– is proportional to size of old space. Literally, JVM have to check single byte of table for each 512 bytes of heap (e.g. 8G of heap -> 16m to scan).

T_old__scan – is proportional to number of dirty cards in old space at the moment of young GC. If we assume that references to young space are distributed randomly in old space, then we can provide following formula for time of old space scanning.

; there S_old is size of old space and D, k_card and n_card are coefficients specific for application (3).

T_copy– is proportional to number of live objects in heap. We can approximate it by formula:

There k_copyis effort to copy object, R_{long_live} is rate of allocation of long lived objects, k_survive = R_{long_live}/ R_alloc(k_survive usually very small), and k_tenure is a coefficient to approximate aging of object in young space before tenuring (k_tenure ≥ 1) (4).

Now we can analyze how various JVM options may affect time and frequency of young GC.

Size of old space. Size of old space is affecting T_{card_scan} and T_{old_scan} part of young GC pause time according to formulas above. So we as we are increasing size of old space (read total heap size) time of young GC pauses will grow and it can be helped. After certain size of heap (usually 4-8 Gb) time of young collection is dominated by T_{card_scan} (technically T_copy can be even greater than T_{card_scan}, bur it usually can be controlled by tuning of GC options).

HotSpot JVM options: -Xmx=n, -Xms=n

Size of Eden space. Period between young GC is proportional to size of Eden. T_copy is also proportional to size of eden but in practice k_survive can be so small that for some applications we can forget about Tcopy. Unfortunately time between young GC will also affect coefficient D in equation (4). Though dependency between D and P_young is very application specific, increasing P_young will increase D and as a consequence T_{scan_old}.

HotSpot JVM options: -XX:NewSize=n, -XX:MaxNewSize=n

Size of survivor space. Size of survivor space puts hard limit of how much objects can stay in young space between collections. Changing size of survivor space may affect k_tenure(or mat not, e.g. if k_tenureis already 1).

HotSpot JVM options: -XX:SurviorRatio=n

Max tenuring threshold and target survivor ratio. These two JVM options also allow artificially adjust k_tenure.

HotSpot JVM options: -XX:TargetSurviorRatio=n, -XX:MaxTenuringThreshold=n , -XX:+AlwaysTenure, -XX:+NeverTenure

Pretenuring threshold. For some applications using pretenuring threshold could reduce k_survive due to allocation of long lived object directly in old space.

HotSpot JVM options: -XX:PretenureThreshold=n

(Next article "Understanding GC pauses in JVM, HotSpot's CMS collector.")

Comming soon:

Tuning CMS collector for large IMDG storage nodes

Running 200 VM instances on OpenStack Compute

2011-05-31T02:18:00.000-07:00

Introduction

OpenStack Compute is an exciting piece of software. It is open, it is evolving rapidly and the community is terrific. However, as with any software that is new to the market, there’s little understanding and hard data on its performance. It’s scalability, for instance, is something that is only known through anecdotes around the watercooler. This post tells a story of a “baby cloud” running in one of our labs.

Our goal is to see how much we can stretch the software until it rips. In this installment, we’ll start with a very basic test — launching 200 virtual machines in parallel.

Setup and first steps

The test stand consists of three blades (two X5570 and 48G RAM each). All blades are diskless and SAN-backed. The exact hardware configuration is not relevant, since performance of individual VMs is not relevant for this kind of test.

Test image that was used for test is a stripped installation of RHEL 5.6, packed using QCOW2 (153Mb).

All the tests were performed using KVM-based RHEL 6.0 version of OpenStack Compute.

Here is our /etc/nova/nova.conf for cloud controller node:

We used the following script to run the VMs:

We started with a trunk build 1058 of OpenStack Compute.

First try

After firing up the script for the first time, we noticed that 3-5% of all machines were not coming up — that is, showing as down when listing instances. This was not completely unexpected, yet still discouraging. Having ten machines out of two hundred failing on the spot is not a good thing.

The investigation showed that connections to Rabbit are frequently timing out. Turned out that we stumbled upon a known bug in the eventlet library: socket connects are incorrectly reported as timed out. Patch to eventlet and improvements in RPC code helped us to eliminate the Rabbit timeout issue and bring down the error rate significantly. Kudos to Chris Behrens and Soren Hansen for the solution.

Second try

The error rate went down, but it was still above zero. Looked like we’ve stumbled on a race condition. Logs showed traces of exceptions about releasing locks. It turned out that the problem was in synchronization while creating VLANs. Simply adding a lock on the VLAN creation (bzr1097) made the problem go away — kudos to Vish Ishaya.

Third try

Now all instances show up as running, but are they actually usable?

“Usable” may mean many things, depending on how thorough you are. We have some pretty sophisticated test suites that can verify if a cloud instance is healthy for most practical purposes. But in this case, a simple ping of the VMs from outside to see if they respond was sufficiently informative.

The result was somewhat unexpected and bizarre: we could ping only 150 instances on network while API was indicating that all 200 were running. Run 150 instances… all 150 are responding. 180… 150 are responding. 151… again only 150 are responding. Looked like we stumbled on some sort of a limit.

Investigation showed that the problem is in dnsmasq, which has default DHCP lease cap at 150. Took some time to figure this one out, but very straightforward to fix — bzr1100.

Finally, the two hundred nodes were able to run without a hitch. Here’s one for the team!

Some random data

For those who want more data, here go a couple pretty graphs.

We ran two different configurations — one with two compute nodes, another with three (the third one being co-located with CC).

Fig 1. Two compute nodes, dedicated cloud controller.Total startup time: 12 minutes 6 seconds.

Fig 2. Three compute nodes, co-located cloud controller.

Total startup time: 8 minutes 20 seconds.

Though this is hardly conclusive, co-locating cloud controller with one of the compute nodes is not necessary a bad idea, at least from the performance standpoint.

Some random observations

Final success rate is 100%, reproducible;
Most of startup time is spent injecting SSH keys through libguestfs;
To run one hundred VMs per host from the same image, only 24Gb of RAM and 3.4GB in /var/lib/nova is needed. Thanks to KSM and QCOW2!

Next steps

Obviously, we are not settling on two hundred VMs run in parallel. This is still very much a work-in-progress and we’re raising the bar higher every time. Stay tuned for the next report from the trenches.

P.S.

By the time this post was written, we were able to go past 400 VMs. The team is working around another race condition as we speak.

Grid Dynamics Delivers JClouds support for OpenStack

2011-05-06T11:10:00.000-07:00

Overview

Grid Dynamics teamed up with GigaSpaces to deliver JClouds support for the OpenStack Compute (Nova). We have created a new JClouds provider called Nova, which is based on the existing Rackspace cloud servers provider created by Adrian Cole, the main author of JClouds library. OpenStack Compute, also known as Nova, is a community effort to create a cloud infrastructure suitable for both public and private clouds. JClouds is one of the best open-source libraries to manage public and private clouds. It provides an abstraction on top of REST APIs for the cloud. And while there are many cloud platforms supported by JClouds, up until now there was no support for OpenStack Compute.

Details

OpenStack supports two RESTful APIs. The first API is implemented to mimic that of Amazon EC2. The second one is Openstack's native API which is derived from the Rackspace Cloud Servers API. The native API is developed by the community and will eventually allow access to all available features in the most simple and natural way. In contrast, Amazon-like API is not supported by the community and is, therefore, harder to extend. Taking these facts into account, we decided to use Openstack's native API in our JClouds provider.

Our provider supports the 1.1 version of the OpenStack API. It supports all core features of the JClouds compute service. However, OpenStack itself does not support some of the JClouds features on certain hypervisors. You can see a detailed list of supported and unsupported features on this wiki page.

Requirements and Locations

Our provider requires the Cactus release of OpenStack, which supports the new 1.1 version of the API. We are using RHEL builds for our OpenStack installations. The packages for RHEL are available from our yum repository. In addition, our provider should also work with Cactus packages for Ubuntu. You can get the artifact from JClouds Maven snapshot repo or get the JClouds source and build your own copy. The provider code is in snapshot-apis/nova folder.

Conclusion and call to action

We are very excited about what OpenStack has to offer. And now with support from JClouds, it becomes much easier to manage OpenStack from Java applications. Please, give our JClouds provider a try. And of course, any feedback is always welcome! You can find instructions on how to get started here and here. Feel free to ask any questions about the provider in JClouds Google group or here in comments.

Relevant Links

Indexes: RDBMS vs Coherence vs Lucene

2011-04-27T01:19:00.000-07:00

Most people are familiar with concept of “index” in SQL database. But indexes are widely used far beyond relational databases. There are even dedicated products, like search engines, which are specializing in indexing of data stored elsewhere. This article will give a high level comparison of capabilities and performance aspects of typical indexes in three different types of technologies: SQL database, in-memory-data-grid (Oracle Coherence) and full text search engine (Apache Lucene).

Query use-case – multiple attribute filter
While working with large number of objects (e.g. execution messages from stock exchange or product records from e-commerce web site), we often need to select subset of them using values of few attributes. A simple example would be ‘find all red women’s shoes of 9 inch size’. In pseudo SQL this query may look like:
select * from product where type=’shoes’, color=’red’, size=’9”’, gender=’W’
Let’s use this query further to compare different technologies.

Inverted index
I believe most of you are familiar with inverted indexes, but I would like to give a short overview anyway. Idea behind inverted index, which is used in all three technologies covered by this article, is very simple. Imagine we have record with fields A, B and C and we want to execute queries like “select * from records where A=x”. Instead of scanning all records in table, we can create additional data structure (index), which will track for each possible value of attribute A list to all records having this value.

So if we need all record which have A=1 we will find “1” entry point in index (by attribute A) and get list of all records which conforms that criterion. Sounds simple, but devil is in details. How to store index? How to handle queries which have multiple criteria? Different technologies have different answers and these answers have significant impact on performance of queries.

RDBMS – BTree based inverted indexes

BTree is most common implementation of inverted index in RDBMS (world of RDBMS is diverse, so I’m sure there are alterative implementations but let’s stick with BTrees here). BTree is good for disk based storages and has been used in relational databases for decades. Typical SQL database will allow you to create index using one or more attributes and you can create multiple indexes for same table. If you are creating index for multiple attributes, order of attributes is important.

Normally RDBMS can use only one BTree index for particular table during execution of query. That’s an very important point!

How RDBMS will handle our “shoes query” then? RDBMS have to choose single best index (usually using statistics). Then it will use this index to select super set of query result and filter this super set record by record testing remaining query criteria. E.g. RDBMS may choose that index by color field is best, so it will find all red products using index and then check every such product to throw away everything which is not a women’s shoes of 9 inch size. In other words, RDBMS is still required to do a lot of work even if it is using index.

Actually our “shoes query” is very bad for RDBMS to handle. Whatever single-attribute index RDBMS may choose (color, type, size or gender) it will likely have to check few thousands of records to finish query. It is still better than full table scan though.

Is it possible to use index more efficiently for multiple attribute criteria queries in RDBMS? – It is.

We can create compound index, e.g. index by [size, color].Using this index RDBMS will be able to scan only though all red products which have size=9”. This way we have less record to check. We can tune indexes in RDBMS to match our specific queries. But for different query e.g. “find all red women’s shoes of brand X” index [size, color] will be useless.

Update: Though BTree indexes are most common in RDBMS they are not necessary only choice. Some databases, including Oracle, can use bitmap indexes (the way similar to Lucence, see below). Also Oracle can do BTree index intersection (similar to Coherence, see below) though query analyzer rarely choose this strategy.

Indexes in Oracle Coherence (in-memory-data-grid)

In-memory-data-grids also may use indexes to speed up query execution. Instead of BTree grids usually prefer in-memory data structures (either hashtables or sorted trees). Let’s take Oracle Coherence as example. In Coherence (and in data grids in general) data are distributed across cluster nodes (processes participating in grid), and each object has its owner node. Indexes are also distributed across nodes; each node is responsible for indexing objects which belongs to node.

When you query Oracle Coherence, query is broadcasted among cluster nodes, each node is executing query over its local data and sends back a result set. Result sets from all nodes are aggregated to produce final result of query.

Each node holds just a portion of overall data and all this data are in memory (not on disk). This makes grids use slightly different approach for using indexes compared to RDBMS. Unlike RBMS, Oracle Coherence can use multiple indexes to execute simple query.

Let’s go back to our “shoes query”. Assuming what we have indexes for every attribute, Coherence will retrieve [set of all shoes], [set of all red products], [set of all products for women] and [set of all products of size 9”] (each from corresponding attribute index). Once it has all four sets, Coherence will calculate an intersection and that will be result of query (all red women’s shoes of size 9”). Important point here is that Coherence doesn’t have to check objects itself as long as it has indexes for every attribute used in query. Sets of objects in Coherence index are implemented as hash sets (using object’s key to reference object). To intersect two hash sets, Coherence has to loop over one set and test every entry against second set. If one of set is small (few dozen of objects) this is very fast, but if both lists have few thousands of objects intersection may take significant time.

This approach also has its limitations. In our “shoes query” all four product sets produced indexes will be quite long, so set intersection operation will take significant time. Sometime removing index may actually speed up query (e.g. [list of products for women] is probably very long), also order of criteria in query (filter in terms of Coherence) in important because Coherence doesn’t do global query optimization. While this approach is generally better for multiple criteria queries than RDBMS one, “shoes query” is still quite unpleasant use case even for in-memory-data-grid.

Lucene and bit set indexes

Apache Lucene is a full text search framework built in Java. Surprisingly full text search and multiple criteria search are very similar and Lucene can handle our “shoes query” out of box.
Lucene is using compressed bitsets to store inverted index (instead of BTrees or hash tables). Compressed bitsets has two very important features: they are very compact and it’s possible to perform binary operations (e.g. intersection) over compressed bitsets without decompression. Similar to Coherence, Lucene will try to use all four indexes to execute our “shoes query”. But unlike Coherence, using compressed bitsets Lucene can intersect all four product sets mentioned in Coherence examples very fast. Even if individual lists can have thousands of elements, it is a piece of cake for bitsets to handle.
Using compressed bitsets makes Lucene very powerful tool for multiple criteria queries, in practice Lucene can be orders of magnitude faster for that type of queries compared to RDBMS or Coherence.

But using bitsets comes at price. In particular update of compressed bitsets requires rewrite of whole set. Update cost of bitset based index is quite high. Apache Lucene is doing spectacular work to compensate this fundamental limitation (e.g. using segmentation of index and logarithmic merging under hood), but still index update cost for Lucene is generally much higher compared to RDBMS or Coherence.

Best of both worlds: Coherence + Lucence
If you are working with Orcale Coherence and starting filling envy for power of Lucene’s search capabilities, you may want to try grid4search project which allows using Lucene as replacement for built-in Coherence indexes. See more details in project page.

Spring Nested - Part II - What?

2011-04-15T13:47:00.000-07:00

In the previous post I described the problem. This post will describe the solution.

Solution

Spring provides parent context feature. It allows us to build a context hierarchy by declaring the special children context creator bean in the Web application context, and specify the children context config XML paths:

<bean id="root" class="com.griddynamics.springnested.ContextParentBean">
<property name="configLocation"
value="classpath:META-INF/spring/authenticator.context.xml 
            classpath:META-INF/spring/datasource.context.xml"/>
</bean>

No magic at all - it just creates children contexts in the loop, and sets the root application context as the parent for them. Look, we already solved all issues: we have separated and isolated contexts; application can run without any particular context's failure just by ignoring an exception on child context initialization;but it doesn’t allow children contexts to call to each other, nor handle Web requests. Let’s address these too.

That “root” bean (as well as other bean from the root application context) will be accessible from the children contexts. It allows us to implement ServiceLocator based upon the Spring features. Let’s add two methods: one explicitly publishes bean into the root scope, and the second one finds the published reference. Then these methods can be invoked via Spring xml as a factory methods inside of the children contexts:

// the "root" bean implements

interface BeanRegistry{

Void export(ExportRef ref, Class interface);


class ExportRef{
ApplicationContext childCtx;
String beanName;
}

Object lookup(String name, Class interface) ;

The most interesting points of implementation are:

to publish reference (export) you need to store bean name, class and provided application context;
context can export the bean before the actual bean initialized (you can tune it by depends-on= attributes);
ExportRef obtains the holding application context quite easy. Honestly speaking it's the only reason for its' existence;
lookup() returns proxies always (which is quite easy to implement with the Spring support);
we've chosen validate references by the classes provided into the both calls (the interfaces are supported only);
references are names only, which meets the Spring familiar developer expectations;
exports and lookups don't introduce any eager initialization, everything is lazy as possible.

The most advantage is the short learning curve for Spring aware developers. When the team crafts the service we just let the client team know the interface class and reference name. They are able to copy-paste several xml lines of the lookup() factory-method bean (see 3.2.3.2.3) - voilà it’s just a Sping singleton for the regular injection:

<!-- child context, root bean is visible from the parent context -->
<bean id="theService"
    factory-bean="root"

    factory-method="lookup">

<constructor-arg value="sampleService"/>
<constructor-arg value="com.griddynamics.springnested.SampleService"/>
</bean>

I’d like to spot that this is close to the plain Spring. In contrast to this, OSGi references identified by the interfaces mostly, instead of names, and the only additional ‘filter’ feature allows to treat OSGi services as a named beans. The interesting thing is how SpringSource adopts this OSGi “feature” to be more intuitive for Spring users.

The context configs structure

The good practice we've found is keeping the service config and the integration config separated: the first one should be imported into the latter one (by the intrinsic Spring ability).

service-core.xml - contains the service singleton bean definition (container independent):

<bean name="theService" class="com.your.ServiceImpl">
<property name="dao" ref="theDAO"/>
</bean>
<!-- there is no "theDAO" bean declared here -->

service-nested-refences.xml - integrates the service into the application wired by Spring Nested:

<import resource="service-core.xml"/>
<!-- spring nested magic goes here -->
<bean name="theDAO" factory-bean="root" factory-method="lookup">
<!-- the reference name and an interface are specified here -->
</bean>

service-test-references.xml - provides DAO stub for testing:

<import resource="service-core.xml"/>
<bean name="theDAO" class="com.your.StubDAO"/>

You can see the separation of concepts in action - no vendor lock-in! Your components can be integrated into any container, even in unit test by providing a stub.

XML authoring (or sugar)

Initially we deal with the several line xml declarations for the service import and exports (factory-method bean). Then, we introduced fancy xml syntax by XML authoring facility it took about 200 LOC. Now the declarations look like:

<nested:export ref="sampleService" interface="com.griddynamics.springnested.SampleService"/>

<nested:import id="sampleService" interface="com.griddynamics.springnested.SampleService"/>

Quite close to Spring DM's XML. Isn't it? Right! It's worth to mention that there are no changes in the flow: these declarations produce the same factory-method beans which has been shown above.

Competitors

OSGi and EJB were criticized before. Surprisingly the single classloader OSGi has been spotted by Kirill. But it still pushes to additional build steps, and has OSGi API learning curve.

Coda

That’s all for now, forthcoming post will tell you how the Web-requests successfully reache children contexts' beans; and how the tiny container becomes smart enough to handle complex children config dependencies and prevent its' cycles.

Data Grid Pattern - Snowflake data schema

2011-04-14T04:05:00.000-07:00

Evil of distributed joins
Traditionally data grids based solutions are using denormalized data models. Denormalized model allows reducing number of data lookups in storage and achieve low response time. Traditional normalized data models are no good for distributed key/value storages. Main argument against them is requirement to do multiple joins during typical data access operation. If your data are partitioned, joins are becoming prohibitively expensive. You have to join each partition from left join side with each partition on right side thus cost of join grows in quadratic proportion of your number of partitions.

Even if you can narrow record set before actual join, your data is still may be located on different servers. Each join between partitioned tables will force your data to be moved across network, adding network latency to response time. More joins – more latency accumulated and you will find you out of your allowed response time very soon.
I other words, generally, in grid, you cannot use neither joins, nor normalized data model (though there may always be an exception).

Denormalization is evil

But data denormalization approach is far from perfect. It has its downsides and serious ones. Data redundancy is a side effect of denormalization. Redundancy increase memory consumption and cost of updates. Loosing consistency due to redundancy is another foe. It is a constant battle.

Data warehousing experience

Data warehousing industry has similar challenges to solve. They have huge amounts of data and also have to deal with distributed data storages. And they have an answer to this challenge for some time!

Analytic databases are usually using "snowflake" data model. Snowflake schema has single fact table (in center of schema) and several dimension tables. Size of fact table is huge, sizes of dimension tables are relatively small.

Snowflake data model allows implementing joins across distributed data storage in smart way. While we still want to have fact data to be partitioned across cluster, we can have a full copy of dimension tables on each node executing queries. In other words we may use replication strategy for small dimension tables while keeping large fact data partitioned.

Now we can partially execute query using dimensions which are available localy and then issue single query to distributed fact table. As long as you do not need to join facts table with itself of another facts table query execution requires just one network round trip (and if you really need joins between facts wait for next article “Map/Reduce in data grid”).

How can we you utilize snowflake schema with grid middleware?

First we have to decide how we are going to store dimension tables in memory. We can use in-memory data grid itself or we can use some in-memory database like Hypersonic or H2 database. Grid will support replication of dimension data out of box and natural support for java object, but in-memory relational DB will offer better support for queries (many grids do not support joins even in replicated storage).

Next question is cluster topology, we may replicate dimension data over all nodes or just few of them (query nodes). These query nodes can be either grid peer or remote grid client.

Functional separation between query nodes and data nodes (ones storing partitions of facts table) is probably most practical because they have different memory/CPU utilization. By separating them you can tune JVM memory options and scale each tier independently.

Unfortunately this is just a pattern but not a feature of any data grid product I'm aware of. You still have to complete a lot of non trivial engineering to make it work, but this work worth an effort. This pattern allows to use IMDG in cases there denormalization is unable to solve problem. This approach is opening totally new horizons for using of in-memory-data-grid technology.

UPDATE: Illustrations of snowflake schema

From financial industry

and from retail

Spring Nested - Part I - Why?

2011-04-12T12:42:00.000-07:00

Yet Another one (Spring based) Service Locator Framework

It's a well known fact that an every Java developer builds his own framework monthly. We've done it too. Let’s have a look, at least it’s useful as a Spring internals guide.

Context

We deliver highload backend platform for one of the leadering US eCommerse sites. Technically the platform consists of a number of components, deployed to a cluster as a single Web Application (aka WAR-file) builds with Spring Web MVC. Components inside of the platform expose services for the frontend application(s) via remote interfaces (we are lucky to use Hessian for it). So, we have a trivial form of SOA implementation, when the platform's services access each other via Java in-process calls. The components are crafted by the isolated Teams and we are keen on reducing communication efforts between them. We need to avoid conflicts and/or blocking between teams, and try to achieve it via separation and formal collaboration procedure.

Why not OSGi, EJB, etc…?

I know you loudly repeat this question ten times while reading the intro above. Particularly these two technologies were rejected due to:

multiple classloaders machinery, which we weren’t ready to take (i.e. we are happy with the single class loader);
maintenance efforts - we need to maintain descriptors and run additional build steps;
overwhelming and learning curve.

In addition to this we don’t need the partial hot redeploy. Just google “OSGI pain“ and you'll find the most of our arguments. It looks like just we are not mature enough to carry on these expenses, and did micro OSGi ourseves.

Ok. Let’s come back to our attempt to utilize the vanilla Spring MVC.

Tiny Problems Appear

We realized that out-of the box Spring Web MVC doesn’t fit for our needs. Let’s look to a sample component - it’s delivered as two jar files: sample-service-api-1.0.jar and sample-service-impl-1.0.jar. Both of these JARs contains the necessary class files, and the implementation JAR also contain Spring Web Application Context config XML. To allow services to access each other inside of the WAR we have to load all services’ config XMLs into the single Web Application Context instance (there is a possible alteration, but it doesn’t help us). It’s quite similar to using an intrinsic Spring ability:

<beans>

<import resource="services.xml"/>

<import resource="messageSource.xml"/>

<import resource="themeSource.xml"/>

<bean id="bean1" class="..."/>

<bean id="bean2" class="..."/>

</beans>

In this case we have a single flat Spring beans namespace among all services i.e. ServiceA singleton is able to inject ServiceB singleton by a regular Spring injection. It’s something similar to the technique described here. But as soon as we take this approach we find that the platform scarifies from the lack of the application contexts encapsulation. The actual problems are:

internal beans are also exposed into the common beans namespace - so, we have to maintain distinguished bean names to avoid the situation when introducing an internal "dateHelper" bean in your component’s context harms the other vendor’s component which is unlucky to use the same name;
some context-wide policies like aspects, autowiring, lazy-loading, bean postprocessors etc introduced by one component impacts another i.e. we need to keep it separated;
sometimes we need partially working platform WAR i.e. when one of the context are failed to initialize, independent peers can work fine e.g. we have some security related service, which is not able to run without some crypt keys, even fake. In this case all developers have to be aware of this detail: obtain, and install the necessary files (the similar problem has place with the team specific resources like DB or backends, which isn't available for another locations). It can be definitely avoided by an alternate WAR configuration e.g. reduced WAR which has the problematic component excluded, but it doesn’t work in a real life due to organizational and technical issues.

We realized that all these things above impact our productivity. Then we decided to slightly modify the approach presents in Spring, and implement ServiceLocator upon Spring contexts hierarchy. See Part II

Fast Hessian methods leads to java.net.NoRouteToHostException: Cannot assign requested address

2011-04-12T08:51:00.000-07:00

Last week the old issue with the performance testing environment appears again - something happens with the network under high load, and the Hessian invocations start failing with java.net.NoRouteToHostException: Cannot assign requested address

The first clue was googled easily. I checked it by netstat - yes, we have a huge amount of TIME_WAIT sockets, that exceeds the number of available ephemeral ports. How to solve it? Initially I tried to hack Hessian with SocketFactory sets SO_REUSEADDR option for the underlying socket, then consider to provide an alternative http client which is able to cache connections. Who ever know how complex it seems?

The actual resolution is quite simple: Java has everything what ever you need -Dhttp.maxConnections=100 allows us to put much pressure to the fast Hessian service methods. Java runtime already has HTTP1.1 client supports keep-alive, which utilized by Hessian underneath. So, we just need to allocate enough pool to prevent sockets closing and the following TIME_WAIT pollution.

P.S. if you still unaware about Hessian benefits, it's really worth to look at it, or ask me at the comments below.

Grid Dynamics and OpenStack

2011-03-10T21:10:00.000-08:00

This week Rackspace announced the creation of the new Cloud Builders business unit, with the goal to provide commercial services around designing, building and supporting OpenStack installations, as well as provide training and certification services. We think that this is an excellent investment on behalf of Rackspace that could very well take the project to another level.

It is even more exciting that Grid Dynamics is a part of this team, together with other Rackspace partners — Citrix, Dell, Opscode, Equinix and Cloudscaling.

With so many players that had announced their support for OpenStack, I am often asked what Grid Dynamics intends to do in this area. Well, we are obviously continuing doing what we did so far — providing state-of-the-art engineering services for the enterprises. In this perspective, OpenStack becomes the foundation for our solutions in the private cloud space.

Why choose OpenStack? There are several reasons. We've been watching OpenStack development even before the Austin release came out, and here are the highlights:

Development model. True open-source (contrast with "open core" model of other vendors), very fast turnaround on incorporating patches. The progress of the last eight months is amazing.
Community. On the Bexar design summit, there were over two hundred people from about fifty different companies all around the globe. It seemed impressive at that time. Well, look at the size of the community today! If it's not exponential growth, then I do not know what exponential growth is.
Talent. OpenStack really gathered a star team — brilliant engineers from Anso Labs, Citrix, Canonical... Just hanging around with those guys is an experience in its own.
No license fees. This is the thing that is really important for service providers — confidence that the platform can scale with their business, not only from the technical standpoint, but from economical standpoint.

With so many companies around, I am often asked — what exactly are we going to do with OpenStack? Well, from the business development standpoint, we will offer a range of solutions for the enterprises. One of them, OpenStack Jumpstart, is already announced — you can read more about it on our site. But since you are reading this blog, you are probably more interested in the technical aspect.

Let's see:

Support for enterprise hardware and operating systems. We are already actively maintaining RHEL 6 port, which will be the basis for the ports to other versions. Also, our main lab is based on Cisco UCS — and we intend to continue working in this area to leverage unique features of UCS.
Advanced scheduling and placement. Enterprises and service providers often want to take the datacenter and network topology in account when placing virtualized resources — for the purposes of HA, DR, and providing different tiers of service. We started working in this direction by introducing availability zones in Bexar, but a more sophisticated and complete solution is underway.
Network-as-a-service (NaaS). Network is a historically overlooked piece of infrastructure among cloud vendors, and sadly, this is the case with OpenStack today. What we are pushing for is treating network as a first-class citizen, allowing to provision software switches, load balancers, firewalls, tunnels and other gear. We are fixing top-priority holes by implementing features such as multi-nic support (which will hopefully make its way in Cactus), but a complete overhaul is needed to support enterprise requirements to the network layer.
Integration and extendability. For OpenStack to be successful in the enterprise scenario, it has to play nice with its IT and business systems — orchestration, service catalogs, portals, billing, service desk etc. Of course, there's no single recipe to integrate with all the possible flavors of business systems. What can be done, however, is allowing OpenStack to be extended with vendor-specific modules. We are working in this direction; one tiny step is introducing pluggable schedulers in Bexar.

We are confident in OpenStack future. With such an impressive start, brilliant team of developers and stellar portfolio of companies backing it, it's hard not to succeed. It's very exciting to be the part of the Next Big Thing.

P.S. The engineering team keeps a separate blog on purely technical matters that is not syndicated here — http://openstackgd.wordpress.com/. If you are interested in our weekly progress, this is the place to go.

Lessons learnt on a 2000-core cluster

2010-12-14T11:18:00.000-08:00

A while ago we had the chance to test our product (embarassingly
parallel cluster computations in semiconductor industry) on a
2000-core cluster.
This, despite the rather simple-sounding problem statement, turned out
to pose a number of challenges and interesting insights into the
issues of, ehm, decently scaling at such scale.

Below you can find a (rather lengthy) presentation from an internal
tech-talk held on the topic, summarizing these challenges and the
lessons or "morales" we learnt from coping with them.

Data Grid Pattern - Network shared memory

2010-11-08T10:44:00.000-08:00

Using shared memory for inter-process communications is a very popular pattern in UNIX world. Many popular RDBMS (both open source and commercial) are implemented as sets of OS processes communicating via shared memory. Designing a system as a set of functionally specialized processes cooperating with each other helps to keep system more transparent and maintainable; yet, using shared memory as IPC provider keeps communication overhead very low. Shared memory is very efficient compared to other forms of IPC. Several processes can exchange or share information with each other via shared memory without involvement of OS (e.i. no syscalls, no context switching overheads).

Unfortunately, shared memory is useful only if all communicating processes are hosted on same server (e.i. connected to same memory circuits). While we cannot use real shared memory when building a distributed system, idea of building system as a set of cooperating specialized processes still remains attractive.

Key point of shared memory idea is that all processes can access and modify the same data; all processes have consistent view and can do atomic operations on data (e.g. compare-and-set). Data grid technologies allow us to achieve same shared memory features, but in network environment.

Modern data grid products (e.g. Oracle Coherence, GemStone GemFire, etc) provide key features for shared-memory-like communication:

shared data with strong consistency guaranties across processes in a cluster;
atomic operations over single key in a shared storage.

Still, one can not drive this analogy too far. Data grids are very different from "networked shared memory". They have specific tradeoffs and it is impossible to just take an application (like PostgreSQL) and make it clustered using Oracle Coherence. But we can reuse and extend this architectural approach and develop a “network shared memory” pattern as a way to build distributed system.

“Network shared memory” pattern

First, you need to separate data and processes in your mind. Next you should identify operations with data which are happening in your system. You may have operations such as serving request using data, updating, importing data, exporting data, transforming data, etc. Ideally, you should have a separate process dedicated for each operation (though it is not always possible in all cases due to various reasons, so be just reasonable). Once you finished with analyzing your data and designing processes, you can start to work on physical model and design a way the data grid will store your data. Remember, data model for grid should be always modeled with access pattern in mind. Blindly copying data model from RDBMS may produce disastrous results (data modeling for grid is another large topic).

Why using multiple processes is better compared e.g. with multiple threads within a single process? Here are few reasons:

You can distributed your processes between servers (e.g. for optimizing resource utilization)
You can start/stop processes independently
You can upgrade processes independently
You have better failure isolation (e.g. memory leak in one process will not bring down whole system)
You can use less heap per JVM, and have individual memory options for different processes

Obviously, there are some drawbacks as well. A few of them:

You will have more JVMs running and thus more overhead of JVM itself.
All processes have to communicate over network and it is not as fast as using shared memory between threads.
If your data is not partitionable, this approach may not work at all

Advanced usages of “Network shared memory”

Data grids have high availability built in. We can leverage this feature to achieve high availability for our processes. First pattern is a “hot standby”. We may have several instances of same process running (e.g. on different boxes) but only one of them active at a time. In a case of active process going down, grid middleware will detect process failure and one of the standby process will take over. Sounds simple, however even this simple approach requires “death detection”, “peer discovery” and “distributed consensus” features.
Believe me, all of these is not an easy task to implement. Fortunately, data grid already have all of this implemented, you just need to use its API. Data grid also can be used to reliably store internal state of process, thus you can implement failover even for stateful processes.

A next evolutionary step of this approach is a load balancing between processes. Data grid can help coordinating processes by sharing routing table for request, state for stateful operations and/or using distributed locks for controlling access to resources.

Data grid is more than just distributed data store
While data grid technology is primarily designed for working with larger data sets, advanced features of modern data grid products may bring benefits to your system even if all your data fit a memory of single server. Their high availability and distributed coordination features may be invaluable for designing modular distributed saluting.

Data Grid pattern - Data flow mediator

2010-10-13T17:35:00.000-07:00

I'd like to start a series of articles about data grid oriented architectural patterns. First pattern I want to present is a “Data flow mediator”. Let me start with example.
Imagine you have a large ecommerce web application and want to do some real time analysis over user actions. You have a stream of simple events, (e.g clicks). You need to do real time aggregation by various dimensions: by user, by product, etc. At large scale this task is quite challenging: number of writes is enormous, different dimensions make sharding challenging and business want this analytics to happen as close to real time as possible (say few seconds delay). With or without data grid this task will remain challenging, but data grid technology have a strong advantages for this problem.
Let me now introduce “data flow mediator” pattern.

In this pattern, data grid is used as a buffer between systems which produces events (clicks), and systems which consumes information (real time analysis modules).

From producer point of view:

Data grid provides high and scalable throughput,

Data grid provides reasonable balance between durability/performance/cost. With data grid, we can store data in memory only, protected by multiple redundant copies.

From consumers’ (RT analysis modules) point of view:

Advanced data grids (e.g. Coherence, GemFire, etc) provide required queering/aggregation tool (implementing efficient queries by multiple dimensions in grid still an art, but it is doable),

High and scalable read throughput. Different analysis modules may share same “mediator”

From architect's point of view:

Mediator decouples data producer from data consumers, thus localizing impact of changes for each component,

Data grid is self managing. Imagine managing DB with 50 shards + HA replication + dynamic adding/removing servers to cluster and you will treasure this feature of data grid.

I have demonstrated this pattern with ecommerce examples, but there are similar use cases in finance or telecom. Key prerequisites for this pattern are:

Large number of small updates,

Data loss is not fatal (either we can tolerate it or restore data from elsewhere),

Large number of read queries,

Queries are reasonable simple but more complicated than just primary key gets,

Low response time required for both read and write,

Scale is above scalability limits of single RDBMS

I hope this article will help you better understand data grid technology and its usage.
Next time I will present “distributed shared memory” pattern.

Stay tuned.

Coherence, magic trick with cache index

2010-10-11T00:41:00.000-07:00

Oracle Coherence has ability to query data in cache not only by key, but by values (or their attributes). Of course queering by secondary attributes is not as efficient as by primary key, but still can be very useful. Oracle Coherence also supports indexes which improve performance of value based queries significantly.
But some times index may not behave exactly as you expect. Here is an example:

Program output:

So far there are no surprises.

Now let’ s create an index and repeat exactly same query over exactly same data:

Program output:

Magic :)

Now let me demystify this situation.
Coherence use inverted index to speed up filter queries. Inverse index is a map which maps value of indexed attribute to a set of keys. E.g. for query like “give me keys of all entries, where attribute A = ‘B’ ” we can extract result directly from inverse index. Coherence will just retrieve key set associated with “B” from inverse index of attribute A (of cause index should be create explicitly). This is simple case. But what if attribute itself a collection (Coherence is storing objects, their attributes can be anything). Imagine you are storing articles and each article has a list of tags. Now you have a query like “give me all article tagged with ‘java’”. This is quite common case. Without index such query will be executed as full scan which may be terrible inefficient. To allow filters such as ContainsAny/ContainsAll also benefit from index in case when attribute is a collection, Coherence will add all element of collection to inverse index also. Rule is simple: if attribute value is a collection or array its element will be added to index, otherwise attribute value itself will be indexed.
In example above Coherence inverse index will look like:

“true” -> 1, 3

“false” -> 2, 3

When equalsFilter is executed using index it doesn’t check real value (it save a lot on deserialization costs). This explain how we’ve got object which is “true” and “false” at same time (there were no real call to Object.equals(...) if index is used).
In practice, you are not supposed to mix scalar and collection value in same index, so it shouldn’t be a problem. But be aware, using EqualsFilter is not necessary involving Object.equals(...) invocation.

Coherence Production Checklist describes TTL expiry attack

2010-08-19T22:42:00.000-07:00

Recently I've found a curious contradiction between advice given by software experts and a real hardware. Let's have a look to Coherence Production Checklist and TTL tuning for production. Does it make sense? At the first glance, yes it does - you secure production cluster from multicast IP clashes by setting ttl to 1 or even 0.
But there is some objections here. Have a look to Catalyst 6500/6000 Switch High CPU Utilization and TTL Expiry . In two words setting ttl<=1 lead to high load of switch/router device, because router prefers to use hardware matrix to route packets. Instead of this, expired packets should be processed by CPU to issue ICMP Type 11, Code 0 - "Time Exceeded", even such replies is disabled. As result router goes crazy, and you have a really odd network behavior.
That's what you should do instead: implement multicast assignment policy in your datacenter; disable or properly control multicast routing - just kill mcast packets at the network segments boundaries by the routing table; and of course do not follow software experts advice when you deal with the real hardware.

Coherence, ReflectionPofSerializer now supports POF extractor

2010-07-16T14:30:00.000-07:00

A year ago I have implemented and open sourced ReflectionPofSerializer. This class has saved me from implementing thousands of lines of boring serialization code for various Filters, EntryProcessors, Aggregators , Invocables and other mobile objects in Coherence.

Recently I was asked about POF extrator support in ReflectionPofSerializer. My answer was no, it doesn't support extraction from POF directly. But it turns out what ReflectionPofSerializer can be easily extended to support it. A bit of coding and woala, let me introduce ReflectionPofExtrator.

ReflectionPofExtractor is an extractor implementation which can be used in Coherence for filtering and indexing. It’s usage is very similar to standard ReflectionExtractor (except it accessing attributes by field names not by calling methods). It can extract a field of stored object directly from POF binary, also it can parse simple paths to extracted attributes (e.g. path "instrument.exchange" will make extractor first extract filed "instrument" then extract field "exchange" from instrument and return it). If ReflectionPofSerializer have been used to serialize object, ReflectionPofExtractor can read value directly from binary POF format (what is it, without deserialization cost). If another type of serializer was used, object will be deserialized and field will be extracted via reflection.

It did some micro benchmarking, results are below:

For cache filled with 500k objects, a full scan filter is executed, below is filter execution time for different size of objects:

Binary object size: 180 bytes

ReflectionExtractor execution time: 20s
ReflectionPofExtractor execution time: 14s

Binary object size: 604 bytes

ReflectionExtractor execution time: 23s
ReflectionPofExtractor execution time: 24s

Binary object size: 1804 bytes

ReflectionExtractor execution time: 150s
ReflectionPofExtractor execution time: 117s

This result proves 2 things:

Never relay on micro benchmark – do proper performance testing of your application.
PofExtractor is not always a performance win, use advice above.

But anyway now you can estimate effect of using PofExtractors before writing boiler plate of custom PofSerializer/PofNavigator.

War story: optimizing one Hadoop job

2010-07-02T13:13:00.000-07:00

Recently, we faced a problem of categorization of shopping items : applying a complex set of regular expressions to product description, figure out category of product and extract category-dependent fields. Since original data is a result of web crawling, majority of records do not change since previous crawl while 1% of records do actually change. This led to idea of incremental processing when we initially select altered records and then process only those that had changed.

Algorithm

Original algorithm was to run a set of map-reduce jobs:

“Diff”. Compare older data and new data to fetch updated records. Create a 'diff' that contains input for updated / new records and list of deleted keys.
“Extract”. Convert diff to hadoop MapFile format, along with fetching fields (desired processing) for updated & new records. Deleted records are mapped to deletion marker.
“New”. Rip out new records to a separate output which along with output of step 4 forms final result.
“Merge”. Go through previous processing results, using diff as a dictionary, replacing updated records and removing deleted ones.

Optimization

Initial implementation performed really bad, it worked even longer than just processing of all fields on a single machine. So, we started optimization which included following steps:

1. Use custom WritableComparable.

Initially, we used CSV text file as intermediate format. Now, we decided to implement custom Writable class (fig. 1) for values and WritableComparable (fig. 2) class for keys of intermediate data.

figure 1: RecordKeyWritable.java

figure 2: RecordWritable.java

Surprisingly, this optimization produced a little impact, about a 1% of total runtime, mostly by eliminating time for string splitting.

2. Adding combiner

Next, we decided to implement efficient combiner that will combine all records of the same product for both Diff and Merge stages. See figure 3 for example of combiner for Diff stage.

figure 3: RecordCombiner.java

After enabling combiner for Diff we won 15% of time, and Merge time decreased by 20%. Data transfer volume decreased significantly (up to 2x).

3. Upgrade Hadoop 0.18.3 to 0.20.2.

We did not expected that, but just this cut total processing time by 30%, mainly because shuffle performance increased dramatically. At the time of our tests, Hadoop 0.20 could not be deployed on Amazon EC2 very easy, so we made a couple of fixes in scripts. Anyway this should be fixed in Hadoop repository by now.

4. Enable GZIP compression for input, output and intermediate data.

We've tried to compress input data, providing input format compression, output compression and intermediate (sequence file) format compression. However, we could not notice impact of any of these optimizations

Output:
SequenceFileOutputFormat.setCompressOutput(conf, true); SequenceFileOutputFormat.setOutputCompressorClass(conf, GzipCodec.class);

or
FileOutputFormat.setOutputCompressorClass(conf, GzipCodec.class); FileOutputFormat.setCompressOutput(conf, true);

figure 4: Snippets enabling GZIP compression for M/R output

Gzipped input is understood by fileInputFormat & SequenceFileInputFormat transparently

5. Enable LZO compression for intermediate data.

LZO compression achieves medium compression rate at a pretty good speed of compression and outstanding speed of decompresion. This makes LZO codec ideal for compression of intermediate data (Mapper's output) so it allows to highly reduce traffic between mappers and reducers at a low computational price. Additionally, it supports independent block decompression. We tried LZOcodec from hadoop-0.18 pack and it allowed us to save 5% of total processing time. Unfortunately, LZO support was removed from hadoop-0.20 due to license restriction. There're several alternative projects in progress, but at the time of our tests there was no out-of-the-box solution.

conf.setCompressMapOutput(true); conf.setMapOutputCompressorClass(LzoCodec.class);

figure 5: snippets enabling LZO compression for traffic between mappers & reducers

6. Tune number of mappers and reducers

After a set of experiments we found out that manual setting number of mappers and reducers can speed up M/R substantially. First, check that map tasks are small enough so all nodes are equally loaded and they are not too small to suffer from initialization overhead (map task execution time lower than 20 seconds is known to be bad).

Further optimization is achieved by tuning ratio of number of mappers to number of reducers. Just by setting appropriate values via conf.setNumMapTasks() and conf.setNumReduceTasks() we were able to cut total processing time by 30%! Unfortunately, optimal parameter values strongly depends on data size and distribution. It appears that taking 3 * input size / block_size mappers and 1 reducer per slave node is a good point to start optimizing.

7. Map side join

We have also tried map-side join for Merge. In order to do this we have to get data for “Merge” step sorted and partitioned the same way. This allows us to perform merge join – our mapper reads two input simultaneously matching records of the same key. This also required extra-map reduce to merge old results (as we did not merged results of update&delete with new before). Also we could not avoid sort & shuffle since we needed IdentityReducer to keep processing results in single file. So finally we could not achieve significant performance increase (5% of Merge time savings) at a price of much higher complexity.

8. Put diff map to local disk instead of HDFS

The next step was to put diff-file to each merging node in local file system, instead of sharing it by hdfs. This saved us 10% more of merge time. Anyway map-file lookup was still too slow taking 95% of merging time so we looked for keeping it as in-memory distributed hash map (DHT).

9. Replace MapFile with in-memory cache

Instead of hadoop map file (which is intended to keep index in memory) we switched to use of in-memory hash map keeping both index and values in memory. Result was outstanding: Merge step became 70% faster and New step got 60% faster. Since our diff was small enough to fit in memory of slave node, we used our own simple HT:

- it serves as a http-service
- on start it reads the diff file and populates hash table with it
- the service is terminted when Merge join is over

If diff size gets bigger so it could not fit in memory of a single slave node, we will have to look up for more complicated solution. Probably one of available in-memory hash table implementation would fit, so this needs further investigation..

Conclusion

If you would face a problem of optimizing of map-reduce task here is our suggestions:

Start with tuning map-reduce tasks number.
Try latest hadoop version
Avoid map-file as a dictionary data : try to put it in memory, reconsider algorithm, look for third-party solution.
Add combiners and switch to custom record format.
Try replace reduce-side join with map-side one

P.S.

This table uncovers some details of optimizations tried and level of effort for each of those:

Improvement	Effect	Effort
Custom record format	Total time reduced by 1%	2 hours of coding.
Adding combiners	This, along with Custom record formal>, reduced total time by 17%. *Diff* time reduced by 15% and *Merge* time reduced by 20%.	2 hours of coding.
Update hadoop	Total time reduced by 30% by similar speed up of all steps	Had to fix deployment scripts for EC2, this should be already fixed in codebase.
GZip compression of input/output	No visible effect	Two lines in JobDriver code.
LZO compression of intermediate data	Total processing time reduces by 5%	Two lines of code in JobDriver in Hadoop 0.18. Significant effort (2-3 days) of adding LZO-like compression to Hadoop 0.20 (until the compression in incorporated in Hadoop distribution)
Map-reduce task tuning	Total processing reduced by 30%	A week of experiments. Note that this should be revisited when data size or distribution changes (and we know it always grows).
Map side join	Negative: total processing time increased.	A week of coding & experiments.
Put dictionary data in local file system	10% saved on *Merge*, total time reduces by 7%	Couple of lines in running scripts.
Put dictionary data in memory	70% *Merge* speed up, 60% *New* speed up, total time reduced by 50%.	2 days of coding for simple in-memory hash-table. In case when dictionary data does not fit in memory, reconsideration is required. Custom in-memory DHT may take a bliss of time, commercial promises to be pricey, open solutions should be investigated..

Coherence. Read through and operation bundling.

2010-06-30T13:57:00.000-07:00

If you are using read-write-backing-map, you should know what CacheLoader interface has methods load(...) and loadAll(…) for bulk loading. And if your cache loader is loading data from DB, your implementation of loadAll(…) is probably designed to load all objects via single DB query (cause it is usually order of magnitude more efficient than issue DB query per requested key). Lets assume your read through cache is working just fine, and at some point you decided to implement cache preloading – a popular pattern. To implement preloading you have found some way to collect keys (e.g. via query to DB) and wanted to use getAll(…) call to load data into cache. Everything looks ok, unit tests have passed, but on real data it takes forever to preload cache! Let’s investigate this …

Problem quickly reveals itself, Coherence is not calling loadAll(…), and instead it calls load(…) for each key in single thread. Culprit here is partitioned (distributed) cache. Your call to loadAll(…) produces single GetAllRequest (one request per storage member owning part of key set to be precise). While processing this request partitioned service is looping over keys from request at call get(…) for each key from backing map. Partitioned service never calls getAll(…) from backing map and it gives no chance for read-write-backing-map implementation to call loadAll(…). Bummer.

My favorite solution for this issue is to load objects from DB directly and then put them to cache (looks a bit ugly, but as you will see below it is the most clean workaround). Sometimes, it may not be an option. Do we have other alternatives?

cachestore-scheme element has a configuration option named operation-bundling. Let’s try to use it.

No effect. What is a problem?

Let me explain how operation bundling works. Operation bundler intercepts method calls ( load(…) for this case) and delay them a little (1ms in config above). After delay it check if there are pending requests in other threads, and if they are, it gathers all requests for all threads and executes them in single loadAll(…), then dispatch results back to callers in other threads (and finally returns its own result of cause). This means max number of request in bundle is limited by number of threads which are accessing read-write-backing-map. By default we have just one thread per partitioned service, that's why nothing interesting happens.

Let’s add thread pool config to distributed scheme.

Test it. Nothing has changed. What next?

Yes we have thread pool now, but partitioned service have received just one GetAllRequest and each request is processed by single thread, so we still have one thread that sequentially calls to get(…) method. Blast!

Go back to our getAll(…) call site, create thread pool, separate keys between threads, call multiple getAll(…) in parallel.

Finally, loadAll(…) is being called now, but number of keys in single call to getAll(…) is still limited to size of partitioned service thread pool, and we also have to have same number of parallel getAll(…) requests.

So, for me this solution is a way more hacky than just load object for DB without Coherence help. I have notified Coherence team on this problem, hope they will fix it in some future release.

Deploying to Hyper-V: Windows over Windows

2010-06-22T10:56:00.000-07:00

Recently I was able to spend some time to getting acquainted with Microsoft Powershell and Microsoft Hyper-V. The task was to imitate elastic cloud service on existing hardware running Windows Server 2008 R2 with Hyper-V. Main goal was to learn how to create and shutdown running windows-based virtual machines on-demand.

Hyper-V provides WMI management interface which is available for use in Powershell, thanks to PowerShell Management Library for Hyper-V.

While solving my task I faced several issues I would like to share

Backup & Restore

Unfortunatelly, current version of NTFS does not support neither data de-duplication nor copy-on-write. Each of these technologies could dramatically speed-up creation of running instance from snapshot. For now, making a copy of virtual machine snapshot (including disk image snapshot) takes about 95% of instance creation time.

Hence, when you need fast instance creation, it might worth to use some NAS that supports de-duplication.

The other option is to use Hyper-V Differential Disk feature, but it requires to mangle or recreate virtual machine config file to avoid original disk image corruption. The other caveat is that if original image becomes corrupted, all differential disks become corrupted too.

Avoiding duplicate VM names

An obvious way to maintain reusable VM image is to prepare virtual machine with all needed software pre-installed, and then shutdown and 'export' it. When importing VM, Hyper-V uses disk image in-place, and rewrites config files making image unusable. Hence, for getting working copy of VM, we need to make a copy of exported image in some other place and then 'import' it.

But when we restore two such virtual machines, we get two identical instances with the same names. We need to give these instances different names to distinguish them. The easiest way is to generate GUID and use it as a VM name.

The VM name is specified in two places in the exported virtual machine XML configuration file. It would be very easy to use XPath to specify nodes for update, but unfortunately, Hyper-V config file parser is very formatting-sensitive. Hyper-V does not accept MSXML-parsed and saved VM config file, because it is saved slightly re-formatted. Hence, we have to replace original VM name with replacement pattern and consider xml file as plaintext.

Update: It is possible to use XML object OuterXml() method to get config file Hyper-V understands. Hence, config file update script might look like:



$hostname = [System.Guid]::NewGuid().ToString()

$cfgfile = Resolve-Path("$VMsPath\$hostname\Virtual Machines\*.exp")

$xml = [xml](Get-Content $cfgfile)

$xml.SelectSingleNode('DECLARATIONS/DECLGROUP/VALUE.OBJECT/INSTANCE[@CLASSNAME="Msvm_VirtualSystemGlobalSettingData"]/PROPERTY[@NAME="ElementName"]/VALUE').'#text' = $hostname

$xml.SelectSingleNode('DECLARATIONS/DECLGROUP/VALUE.OBJECT/INSTANCE[@CLASSNAME="Msvm_VirtualSystemSettingData"]/PROPERTY[@NAME="ElementName"]/VALUE').'#text' = $hostname

$xml.OuterXml | Set-Content $cfgfile

Avoiding duplicate hostnames and SIDs

After automatic virtual machine creation was implemented, the next issue was machine names and IP addresses. Obviously, when a complete copy of the VM is created, the copy will have the same hostname as original machine and some IP address obtained via DHCP.

Avoiding machine SID and hostname duplication is pretty simple. Microsoft provides freeware system preparation tool in Windows Resource Kit (sysprep.exe). It has different configuration file formats for different windows versions, but setupctl.exe tool from Windows Resource Kit generates valid config files for any windows version.

Sysprep.exe moves Windows back to sealed state (the state in which pre-installed windows exists on new computers), removes itself from disk and shuts down the Windows.

In sealed state all host-specific information is misssing: hostname, SID, Windows Product Key, activation date and so on. On first boot Windows starts post-install process, queries user for hostname, product key, generates system SID and makes several other steps. Setupctl.exe allows to create unattended post-install file. One of important unattended setup features is optional random hostname generation.

Hence, duplicated SIDs and hostnames can be avoided using sysprep and setupctl at the last step of preparing reusable VM image.

Acquiring IP addresses of the VM

IP address of running virtual machine is available in VM key-value pairs. Hyper-V acquires it through Integration Services. Note, that on some early versions of Hyper-V it is impossible to acquire IP address of the running virtual machine. Hence, if you installed Integration Services on VM, type in powershell:

Get-VMKVP "vm-name"

if you don't see IP address among other values, you should upgrade Hyper-V.

Conclusion

Microsoft Hyper-V with some workarounds may be used as local cloud-like infrastructure. Combining such infrastructure with public cloud services like Amazon EC2 and GoGrid may provide almost unlimited infrastructure on demand with reduced cost as a result of local infrastructure usage when demand is moderate.

Have a nice day!