/dev/websphere

Implementing global indexes on WebSphere eXtreme Scale

We get asked a lot how to create global indexes with WebSphere eXtreme Scale (Scale). Scale usually keeps K/V pairs in a large distributed hash table (DHT). Customer want to index attributes of the value and then run queries to find all the Keys with values match some criteria based on the index.

You can, of course, index the attribute and then run a parallel agent to run a query on each partition and then union the results. That works but isn't great from a throughput scaling point of view as every server in the grid is doing a part of the total query. The throughput is then limited to the throughput of the slowest box in the grid. It'll work but it's not great. What it does well is to speed searching over the records, as the larger the grid, the more processors are running the search.

But, there is a better way. It's not so difficult to implement a global index. This is a grid wide index for a specific attribute. This is implementable on top of a DHT (like WebSphere eXtreme Scale) and makes it possible to find all matching keys in a single operation using this type of index. Basically, we make a Map whose key is the search term and the value is the list of keys containing an attribute with that search term. Thus, an index lookup results in a Map.get(value) returning the list of keys. The list of keys can then use WXSUtils.getAll to bulk fetch the results quickly and now we have a better solution. The throughput of this system is much better than a parallel search implementation. Each index lookup results in one RPC to exactly one server. Thus, one server can do N of those look ups. M servers can do M x N lookups and so now we have a better response time as well as linear scaling on index look ups from a throughput point of view.

The code for this is available on my github repository in the wxsutils jar with full source code. It allows three types of index to be created. An exact match one (==), an index which matches any attribute prefixed by a symbol (LIKE XXXX%) and finally one which matches any records whose attribute contains the query anywhere (LIKE %XXX%).

I'll post a blog entry shortly showing some code on how to use these indexes. The Test cases on github also include code showing how it works.

Why would you want a global index? Lets suppose you wanted to find all matching products containing a substring anywhere in the product name. The SubStringIndex in this code allows you to easily index that attribute in a Map containing these records and then find all matching products in low single digit milliseconds or so (depending on the processor), end to end. WebSphere eXtreme Scale can maintain that response time as you scale up the grid by adding boxes. More boxes means larger indexes are allowed and there is more CPU/network for requests, linear scaling. Remember, the query throughput is M x N which M is the number of servers in the grid.

The classes to look at in github are in the package "com.devwebsphere.wxssearch". The IndexManager is the main class from which you obtain an Index instance to search an individual attribute or you can run multiple attribute searches directly using the IndexManager. There are annotations to annotate which attribute on the value object you want to index. The annotations specify the kind of index as well as index configuration properties.

SearchSOA article on WebSphere eXtreme Scale using Microsoft OData for REST

Read the article here at this link.

Video of my QCON SF 2009 Session is online

This session is my talk on the Redis style of key/value programming, whats good and whats not so good. I built a redis like front end for WebSphere eXtreme Scale which I'm improving over time as a prototype with which to develop this kind of ideas.

Maybe the multicore crisis is overblown for now but problems remain

3 years ago, the processor design trends were worrying. Designs were moving to lower clock speed cores with more cores. This meant that software needs to become more parallel to maintain response times. I've already seen customers complain that software runs slower on newer hardware. They are looking at response times, not throughput when they complain.

However, the other question is what will those cores be used for. At the time, people still ran applications on dedicated boxes. The virtualization trend was beginning but is becoming more entrenched now. People today are not using these boxes to run a single application. They are either virtualizing the boxes or running multiple applications side by side on them so save memory.

So, the question is: Has the #cores changed per application?

Maybe not is the answer. 3 years ago, 2 way or maybe 4 way boxes were common. Intel ships 8 core boxes now and 12/16 cores are just around the corner with the next set of chips. Large SMP boxes like Power/Sun are running virtual machines or the equivalent of Solaris Zones/AIX WPARs. I've a feeling that the reality is people are using fewer cores per application than 3 years ago.

This bodes well for existing applications on multi-core boxes. The crisis thought by many to be coming may be delayed or even mitigated as the extra cores are used for virtualized configurations not for running a single application. Of course, this means response times will be longer as applications will not be tuned for all those cores but this may simply be the price of more throughput per server through ever more cores/socket.

Memory will likely constrain this as this application consolidation is only really possible if there is enough memory on the box.

Is it possible the market for high socket multi-core servers will only be high end OLTP databases and the like? Will farms of two socket servers running automated virtualization layers be dominant for virtualizing applications? If the applications are not optimized for lots of cores then a large SMP box may not be the way to go. Farms of 2 socket boxes look attractive if the cost of managing them is reasonable compared to the cost savings versus a large SMP box. The cost per core will likely stay higher on a large SMP versus a blade type system.

Customers willing to optimise applications for multi-core will face a new challenge. I expect large SMP servers to have lots and lots of cores but slower cores. If you can't scale out your application then it'll be the only game. However, if your application does scale out then I think farms of blade type systems will be very cheap relatively speaking and be cheaper to develop as the blade systems will be faster per core than the large SMPs or at least thats my expectation. Databases are in a rock and a hard place here. The best response time will be blade type systems, probably x86 or similar. Throughput will be highend Unix systems but I don't expect these to match the response times of the blade machines but throughput will be better. 

However, the cost of developing a single process to scale on those high end machines (maybe 160+ cores) will be substantial because of the difficulty to achieve very good vertical scaling. Blade systems will have much fewer cores and as a result, software is cheaper to develop to saturate the server. Thus, blade systems allow scale out applications to be cheaply developed, run on inexpensive boxes and have good response times. Software that can scale vertically only will remain expensive to develop, will run on high end expensive boxes and it'll be difficult to match response times depending on the application of the blade systems although if it only scales vertically then at the high end of throughput blades just can't do it. But, even 2 socket x86 blades are going to be extremely powerful so the realm where the high end SMPs boxes are ultimately required is pushed significantly upwards and a lot of people will find the performance of the blade systems meets a lot of needs currently met by SMP boxes.

I guess for CTOs staring as the costs of rewriting software, he/she may be able to relax a little but for architects, life moving forward is getting more complex.

Bloom Filters for efficient fraud detection with IBM WebSphere eXtreme Scale

I've written and uploaded a Bloom Filter implementation and uploaded it as part of the wxsutils library I keep on github. You can look at the source code here. If you want to use it with your WebSphere eXtreme Scale project just grab the jar from here and add it to your class path.

A bloom filter is best explained elsewhere in terms of how it works. The use cases for where it's useful are worth going over again here though.

Cassandra uses a bloom filter to track all the keys where are stored on disk in a space efficient data structure, i.e. the bloom filter. It uses this to avoid disk I/Os that would otherwise be used looking up keys definitely not on the disk.

Another more customer friendly example is a fraud detection scenario. Lets say you're a credit card company that has the spending history over 6 months of all your customers. You want to catch purchases to vendors that are not in that history. One way would be to keep all the customer vendor history in a database or in a datagrid and check if a new transaction is for an existing relationship. If not then you'd escalate it for scrutiny. For a large credit card company, this can easily be billions of rows. Lets say 200 million credit cards with 200 vendors each? That could be a lot of memory.

A better approach would be to create a custom bloom filter for each customer. The bloom filter is initialized for the number of actual vendors the customer deals with, maybe 200 vendors on average. Each vendor is then added to the bloom filter. This filter is a byte array when this is done. It's about 256 bytes. Pretty small to store 200 vendor strings, no? This filter can then be loaded in to a DataGrid and when a new transaction comes in, the filter for that customer can check with a 97% accuracy whether the vendor was used before by the customer. This is very accurate and was achieved with a huge memory saving and we can improve the accuracy further at the expense of memory if needed. This is the beauty of bloom filters. This example can be used for:

• Tracking unusual telephone numbers for cell phone fraud
• Cities/zip codes that the card was used in
• Categories of vendor the customer deals with (i.e. electronics is common but not motorsports)
• and so on.

Basically any time you want a check a list of things for whether something is present or not present then it's a very memory efficient way to do this.

Another use case would be a security system that wants to track every password every used by a user and try to prevent them using one every again or prevent a customer from using banned passwords. A bloom filter could be used to track existing passwords or the list of banned passwords and allow those lists to be checked quickly without taking a lot of memory or disk space. It has huge advantages in the existing password case as the passwords are not physically stored, the bloom filter is like a fixed size set of password hashes.

IBM WebSphere eXtreme Scale customers can create Maps keyed by customer name whose value is a Bloom Filter. A WXS agent can be used to quickly check if a vendor or some related entity probably not in the 'list' of valid things for that customer. A batch job can generate the bloom filters from the actual vendor data and the grid can be preloaded with the resulting bloom filters periodically. The actual vendor data stays out of the grid.

Security is also improved with a Bloom filter. The grid is not storing any sensitive information. The actual vendors the customer deals with are NOT there. The bloom filter is just a bit array that can be used to check if a vendor is not there, it's not a list of vendors that could be hacked.

Bloom Filters are a very useful data structure to save memory as well as CPU path when checking lists of things for exclusion. Applications can use DataGrids such as IBM WebSphere eXtreme Scale to build extremely scalable, high performance systems exploiting techniques like BloomFilters for fraud detection as well as other scenarios.

GitHub for WebSphere eXtreme Code samples from my blog

I've been working on adding samples to github over the holidays. The wxsutils package has a fair bit of sample code around using agents and exposes some simplified primitives such as bulk get/put/remove operations. Just grab the jar and put on the classpath and then create a WXSUtils instance providing your client grid as the constructor parameter. Then you can call the getAll, putAll or removeAll methods for bulk operations.

The redis-wxs package is an implementation of a redis like API for WebSphere eXtreme Scale. It has a database (mysql or db2) with a fixed schema. There is a WXS grid running in front of the database. This is a collection of JVMs which act as a read-through write-behind cache on the state in the database. Client applications interact with this grid.

The clients can work with the grid using a very simple API. The grid basically allows the storage of:

• string/string pairs
• string/long pairs
• long/string pairs
• long/long pairs

It also allows lists and sets of strings or longs with string keys. The lists and sets have efficient primitives to make working with lists and sets easy. Lists have push operations which add new data to the left or right of the list. Pop operations remove items from the left or right. Trim operations reduce the size of the list. The range operation fetches a consecutive series of entries between two indexes from the list.

Set operations are also supported in that you can have a named set of Long values.

All data is stored ultimately in the database through the grid.

You can find the code at my github page

Tip of the century: Instrument your code with JMX

We got a lot of questions from customers on performance. The application is slow and we want IBM to help. Usually the first question is whats slow and nobody knows. The application was written with no builtin monitoring at all. The product monitoring can be useful but sometimes application specific metrics are much more useful.

IBM WebSphere eXtreme Scale includes monitoring but I met with Tom Lubinski from SL.com at QCon and he said that all samples should create MBeans and thus be monitorable. Many customers simply copy samples so this would mean more customer applications would have monitoring. So, here is the PureQuery Loader for WebSphere eXtreme Scale with monitoring.

I create a single MBean for each Map in a JVM. This MBean tracks statistics on the Loader for that Map. The purequery Loader uses classes from the purequery.jmx package but you can modify your Loader to make the same calls and get the same monitoring. You can do this with your own Loader pretty easily. Typically, add 3 calls for the start time, log exceptions and log the time in the operation.

I used JConsole to just connect to a container or test JVM and then I can see the beans for each Map, click on attributes and there are all the statistics. Here is an example of what it looks like in JConsole

 

You can see how many of each type of operation happened, how long the min/avg/max time is in milliseconds. How many exceptions? The string of the last exception and so on. This is very useful and adds almost no overhead to the running application. This kind of instrumentation in your application will save you a lot of time when performance isn't what you expected. It's easy to add, adds almost no performance overhead (when done with fast code, no sync blocks and so on). Just do it.

I'm in the process now of modifying the Chirp sample application to have MBeans for the client side/agent calls the application uses and the Loader code above. It makes the application much more demo-able as well as easier to find issue when they arise.

The source code for the instrumented PureQuery Loader with some framework classes are attached here.

New version which is thread safe: Download JMXLoaderSampleV2

Old version which had thread safety issues, sorry I was lazy :) (remains to compare with updated one)

Download JMXLoaderSample

IBM DB2 Purescale is very cool, IBM WebSphere eXtreme Scale makes it better

DB2 Purescale is announced now and it's pretty cool, they have worked for a long time on this and it looks like a great achievement by the folks in DB2 land. It's a scale out SQL database based on p-series hardware for now and the new cluster accelerator component. The cluster accelerator component is very similar to the coupling facility on z-series. As a result, DB2 Purescale is very similar to the design of DB2/390.

However, it's primary competitor is Oracle RAC. It's a very different design than Oracle RAC in that it does actually scale out well. RAC uses a peer to peer model for coordinating between its nodes that leads to performance bottlenecks under non application partitioned read/write transactional workloads. As a result, it appears most Oracle customers use RAC simply for 2 node hot failover. It's a solution for high availability, not scale out...

DB2 Purescale is different. Instead of trying to have each node coordinate with every other node, they centralized elements of this communication to the cluster accelerator. Centralized sounds slower but its not here. The peer to peer model results in more operations for coordination, all nodes are involved in the coordination function. The centralized one is much less as there is only one node handling this. Big difference, reduces coordination latencies significantly and this means better scaling. Of course, the cluster accelerator will slow down gradually as the cluster scales out but you end up being able to scale significantly more with this approach than the peer to peer approach. It's unlikely in practice customers will hit the limits.

Customers should also choose Purescale over previous solutions like DB2 HADR or queue replication for HA because the second box is actually contributing to the workload. 2 boxes gives higher performance than a single box with a passive replica. But, replication between data centers will still require technologies like queue replication and so on as all purescale nodes need to be on the same high performance network and SAN.

So customers can, of course, use PureScale for 2 node hot failover like Oracle RAC customers do but they will also be able to use it to build larger scale out databases and unlike Oracle RAC it will perform and scale up well. This will let customers build larger N node database clusters with hot failover and get very reasonable scaling. This may mean customers that previously looked at sharded databases might view purescale as an alternative to sharding as it will still look like a single database but operationally a sharded database (i.e. a manually partitioned set of independent databases) may be better from a risk point of view because the shards are independent and scale forever where as there is a limit for purescale but for most customers the limit will not be a factor. Purescale should give many customers looking at sharding a great alternative for scale out without the hassles of partitioning. Plus, there is no reason why a pair of Purescale nodes would be a shard. Each shard still needs to be highly available and if a shard gets really hot then you can add a node to the shard which is impossible with a conventional database. DB2 Purescale solves the hot shard problem nicely.

Does this mean there is no need for caching products like WebSphere eXtreme Scale? Of course not. Lets examine why:

• It's still an SQL database. You needed caches before for SQL databases, you still need caches for DB2 Purescale databases.
• IBM is not giving away DB2 purescale for free. It runs on Enterprise class servers with a SAN and very high speed networks. Companies will get significant value from buying Purescale and as a result they will want to extract the best value from it also. This means they will want to cache data in front of it just like they would have using an SMP database. Nothing has changed.
• Data is served up in SQL form and for many applications then needs to be object relationally mapped to objects. This means complex SQL to the database which is slow and lots of path length doing the actual mapping to objects in the application. Products like IBM WebSphere eXtreme Scale cache data in the native application form, this makes it significantly faster to fetch from the cache rather than fetching from a database and reduces the path on the application side because there is no mapping.
• You can't collocate your application with your data. The data will still be in a separate tier just like todays databases. 
• It's a single data center technology. The communication between the nodes and the cluster accelerator require a very high speed, low latency network and access to a SAN for best performance. You will only be able to use a purescale cluster within a single data center or possibly within a single network switch for best performance. Products like IBM WebSphere eXtreme Scale have deployed/proven out of the box multi-data center (more than 2) active/active capabilities.
• Customers used caches to reduce response times to access data. While DB2 purescale absolutely scales out horizontally extremely well, each operation still does a small amount of communication with the cluster accelerator. This is work that isn't needed with a normal single node database. This means slightly slower response times so if you needed to cache before for response times then nothing has changed with purescale.
• It's not a cost effective place to store short lived persistent data like application/http sessions or data related to application-customer conversations. Products like IBM WebSphere eXtreme Scale are designed to meet these needs in a more cost effective manner.

Most of this is not specific to purescale. These points apply to all databases, even mysql. These are the reasons customers used caches with databases or replaced databases with caches already. So, I hope we sell a truckload of DB2 purescale with a truck load of IBM WebSphere eXtreme Scale in front of it.

Memory usage/sizing in IBM WebSphere eXtreme Scale

UPDATED to include compressed 64 bit numbers.

I figured this was worth doing quickly so here are some numbers for the overhead in WebSphere eXtreme Scale for storing 1m objects in a grid. I'm using a single JVM grid in this case but this applies no matter how many JVMs.

I'm storing 1 million Person objects with the following structure:

    String firstName;

    String surname;

    String street;

    String city;

    String zipcode;

I'm populating these strings with some small sample strings. I'm using a key which is just a person# string.

I did the test with 32 bit and 64 bit Sun Java 6 server JVMs on my macbook. I pulled heap usage using the jmap utility supplied with the JDK (jmap -histo:live pid).

WebSphere eXtreme Scale can store object values in two ways, as POJOs or as a compact serialized form (more efficient than normal serialization, like a byte packed version of the data in the value, only available in 7.0+). The byte mode is enabled by having a copyMode="COPY_TO_BYTES" on the backingmap definition in objectgrid.xml. Lets look at the heap size after a system GC in 64 bit JVM mode:

• 64 bit POJO    782.7MB
• 64 bit BYTE    381.0MB

So, you can see the BYTE representation is much more compact than the POJO version, about double the density.

Lets look at when we use -XX:+UseCompressedOops

• 64 bit POJO 592.0M
• 64 bit BYTE 299.4M

You can see that compressed mode significantly reduces the 64 bit overhead by about 25%.

If we do the same thing with 32 bit JVM (-d32 on Mac) then we see

• 32 bit POJO    523.6MB
• 32 bit BYTE    261.0MB

You can see 64 bit has quite a large overhead on Sun. Compressed 64 bit isn't so bad but still isn't as good as straight 32 bit. This is why we recommend 32 bit JVMs with heap sizes under 2.5Gb. It's just most efficient from a memory point of view.

In 32 bit mode, here is a break down for the POJO mode:

•  num     #instances         #bytes  class name
• ----------------------------------------------
•    1:       6016578      230933320  [C
•    2:       6017049      144409176  java.lang.String
•    3:       1000000       80000000  com.ibm.ws.objectgrid.plugins.CacheEntryHeapFactory$HeapEntry
•    4:       1000000       32000000  sample.bench.Person
•    5:        342796       19860816  [Lcom.ibm.ws.xs.util.AbstractMapEntry;

The WebSphere eXtreme Scale runtime uses around 80 bytes for each cache entry and we use a optimized hybrid multi-level hash map taking 19MB total or 19 bytes per entry. So, thats 99 bytes per entry.

The BYTES model looks like:

•  num     #instances         #bytes  class name
• ----------------------------------------------
•    1:       1005960      108283256  [B
•    2:       1000000       80000000  com.ibm.ws.objectgrid.plugins.ByteArrayCacheEntryFactory$ByteArrayEntry
•    3:       1016742       25359464  [C
•    4:       1017199       24412776  java.lang.String
•    5:        342802       19861344  [Lcom.ibm.ws.xs.util.AbstractMapEntry;

So, our overhead here is 80 bytes per byte array entry, the values are stored as byte[]s in the 108MB chunk and the hybrid hash map takes again 19MB. So, the overhead is still 99 bytes per entry but each entry is stored more efficiently than in POJO mode and it is this saving which makes the big difference in the heap sizes seen above.

Applications interacting with the grid remotely typically use byte mode. Applications running collocated with the grid for highest performance typically use POJO mode in combination with other features but thats another blog entry.

The WXS code about to be released drops the overhead down further to 64 bytes rather than 80 and that will be available soon.

Each partition primary/replica shard is kept in its own area and we've had that design from the beginning because if we kept all data in all shards in a JVM in the same hash map then splitting them is expensive so we never did it that way.

Hopefully, this helps customers size heaps and memory. Obviously, we are continuously working to improve memory density so we can change this when ever we choose to so individual class names we use are not guaranteed at any time.

KV Programming with a column per entry versus entity per entry

It looks like memcached and redis users typically store data in column oriented style. This means that rather than store a Person object using the person key, they instead store each attribute of the person as a seperate key, for example:

<P:123:FirstName,Billy>

<P:123:Surname,Newport>

<P:123:Country,USA>

So, if the object had 3 attributes then there are 3 entries. This is flexible in that it's easy to add attributes when they are needed. The store object approach makes that more difficult as the value has a type and it's not easy to add a new attribute. You could store a HashMap as the value which means adding an attribute is easy.

So, you get flexibility with the entry per attribute approach but you pay a heavy performance cost. If a page uses a Person data entry then you may need to make several remote calls to the cache to get this one person. If there is a list of friends then you'd do this for each person. The number of RPCs per page starts to add up pretty fast. I'm pretty sure this is why you see blog entries etc on why memcached needs crazy fast response times, it's simply because one page calls it so often. So, while for environments where extreme schema flexibility is needed, it's an option. You pay a heavy price in terms of performance.

An entity based approach where you store the object at the key uses less memory on the server simply because there are fewer entries per object, i.e. there is one versus one per attribute. Fetching a single entity is cheaper because it's one RPC and you have it all rather than one per attribute. This means that if a page has ten Person objects to be rendered and each Person has 10 attributes then the column approach uses 100 RPCs while the object approach uses 10.

The other issue here is memcached does not offer transactions. Attributes for a single 'object' may hash to different memcached servers and with UDP especially, updates are not guaranteed to work. So, consistency is definitely a problem. Updates may half work. This isn't acceptable to many and the entry per object approach avoids this issue even without transactions because sometimes, all or nothing is better than half way.

I don't think there is a one is better than the other here. For extreme flexibility on schema the attribute per entry model OR the hashmap per entry model is most flexible but it definitely has a cost in terms of performance and consistency. Fetching only exactly the subset of the attributes that you need may be a benefit but the RPC cost cannot be ignored.