/dev/websphere

Configuring DataSources with DataGrids that front end Databases

I'm seeing this a lot these days. A customer is using IBM WebSphere eXtreme Scale in front of a database. The DBAs are concerned about how many JDBC connections that the grid is requiring. If it's a larger grid then the problem is especially acute. For example, lets say we have a grid with 200 JVMs in it. All of the grids that I have sized recently are in this range. Lets look at different aspects of this problem

1) WXS as side cache

Here, there are no connections from the grid to the database. The application server JVMs do all the database interaction and hence, the grid adds no additional demands on the DBMS.

2) Preloading WXS with data from a DBMS

Here, preloading it can be interesting. The main issue is usually the time to load the grid with the DBMS data. We usually use parallelism when we do this but that means multiple JDBC connections per preload client. The main issue I see with DBMS preload is simply query complexity slowing down the preload process. Simplifying the query and moving join logic in to the grid side preload process seems to be the key to solve this problem. Once you can achieve a high preload speed per single preload thread then you don't need so many JDBC connections to achieve a specific preload speed.

3) Write through WXS to a DBMS

This is probably the worst offender here. Typically, a WXS Loader consists of a TransactionCallback (like this JDBXTxCallback example) and a Loader. The TransactionCallback makes sure a JDBC connection pool is available for the Loader and the TransactionCallback manages JDBC.begin and JDBC.commit and rollback operations. The Loaders share the connection provided by the TransactionCallback for a specific transaction.

Clearly, preloading the grid will enormously reduce the usage of JDBC connections for grid read operations. However, write operations will still consume one JDBC connection per concurrent WXS transaction. Given, the potential throughput performance of such a grid, clearly, the load is usually too much for most databases. A grid server with 20 JVMs on a modern blade can still achieve 20k transactions a second per blade. Lets start with a grid with a 100% write rate for some peaks. Thats 1000 write transactions per JVM. If each write transaction costs 40ms (remember, the DBMS is directly involved with writes here, this isn't the grid, it's the DBMS) then we could do 25 transactions per thread per JVM (1000 / 40ms). Now, how many threads is that? We'd need 40 (1000 / 25 ops per thread) threads per JVM to achieve this amount of throughput in a single JVM. So, the JVM needs a JDBC connection pool of 40 connections (one per thread) too.

If there are 200 JVMs then we need 8000 (200 x 40) JDBC connections for the grid! Now, we can see why the DBAs may be concerned. A write through grid thats write intensive is a great idea for a grid. But, only with write behind. Write through makes NO SENSE AT ALL for this situation. The DBMS will remain the bottleneck and the customer will likely question why use a grid at all if such a solution with write through was proposed. Remember a 10 blade grid with 20 JVMs per blade and a 100% write rate for some workloads would get you here.

So, write through can still use a lot of DBMS connections if the write transaction percentage is high. If it's lower then these numbers can all be revised downwards. For example, if it was 10% then lets do the math again. One box does 20k tx/sec. 10% is 2000/sec. There are 200 JVMs then thats 10/sec per JVM. One or two connections in the DataSource would likely be enough here depending on response time requirements. This gives us 200 JVMs x 2 connections or 400.

It's very important to share a DataSource within a JVM, do not give each JDBCTxCallback it's own DataSource. The example JDBCTxCallback provide in this blog entry (see the link above) accepts a DataSource reference when it's initialized. You can use Spring to wire it in or change the JDBCTxCallback to get one from JNDI when there is an application server hosting the grid containers. This way, even though there is one JDBCTxCallback PER primary shard in a JVM, at least they will all share the same DataSource instance. This would reduce the connection usage dramatically. This is usually only a problem when used a J2SE hosted WXS container. Customers using WXS inside WebSphere ND will likely be using WebSphere's DataSource implementation and this will naturally result in one DataSource per JVM.

4) Write behind WXS to a DBMS

This approach gives the best opportunity to cut JDBC connection usage. Assuming, we preloaded the grid then the connections for reads can likely be ignored. This leaves writes. Write behind decouples the writes to the database from the grid writes. This means that one WXS write does not mean one DBMS write. This was the major problem with write through.

WXS allow DBMS writes to be triggered every X 'dirty' entries in a Map or when 'dirty' entries in a Map are older than a certain time (5 minutes for example). This means we would see much fewer database writes than grid writes. If we have a grid JVM with 5 primary partitions in it (the normal amount when sized correctly) then there are only 5 potential concurrent writes per second in such a JVM. Each partition will send its changes to the DBMS using a single thread. Therefore, the maximum JDBC connections used for write behind is the number of primary partitions in the JVM which in this case would be 5 PER JVM.

Note, that when boxes fail then the # of primary partitions can be more. Thus, size everything with what you expect to be your minimum # WXS container JVMS to figure this out. It's more likely to be 8-10 primaries in this case. Thus, we'd expect 10 JDBC connections per container if the system was extremely write intensive. However, a very high write rate is unusual. Normally, the write rate is much lower (10-20% might even be high). But, a security system built with a grid that updates last login time for example might result in a WXS write per login so be careful sizing this.

We expect in the worse case scenario with write behind to need one JDBC connection per partition. However, this is only the case when every partition writes at the same time. The big factor here is how long does a write behind transaction take against the database or the 'duty cycle'. If it takes 1 second and we're doing one write behind every 5 minutes then we can likely use one shared JDBC connection in a DataSource per JVM, thats a duty cycle of 1 : 300 (300 seconds in 5 minutes). We are unlikely to get much contention for that one connection as there is a 1:300 change of writes from different partitions clashing within a single JVM. If each write behind tx takes 2 minutes then this clearly changes things given there are 5-10 primaries per JVM. 2 minutes would mean a duty cycle of 120:500 or about 25%. You'd need probably one per primary partition here or 10 per JVM.

But, regardless, it's clear that write behind can reduce the DBMS connection usage significantly when the write behind configuration is 'optimal'. Optimal depends on the situation but as a starting point, 5 minutes or 1000 records (T300;C1000) isn't a bad place to start. The size of the DataSource pool for each JVM depends on the duty cycle of the write behind within a single JVM.

The points on sharing a single DataSource within a WXS JVM equally apply here as to write through.

Summary

I hope this blog entry opens peoples eyes in terms of JDBC/WXS interactions. Most applications elect to use a grid because the DBMS is unable to provide adequate response times under peak loads. If the write rate is very low then there is no problem. But, when the write rate is high then write behind is usually the best scenario here. A high write rate application using write through is likely going to still have performance issues as the DBMS will prevent the grid from achieving linear scaling and good response times.

A quality DataSource implementation is a big deal here. The DataSource thats provided with WebSphere Single Server/Base and WebSphere ND is about as fully developed as exists out there. It is a significant improvement (in my humble opinion) over C3P0 or Apache dbcp. Thus, running WXS inside a single server WAS install is something you could consider instead of a J2SE based WXS implementation for this reason alone, the DataSource implementation.

Video of my StrangeLoop 2010 keynote on nosql posted

Infoq just posted a video recording of my keynote on Enterprise NoSQL at StrangeLoop 2010.

http://www.infoq.com/presentations/Enterprise-NoSQL

Easily run queries across all data in IBM WebSphere eXtreme Scale

I just uploaded a small sample showing how to run a query across all data in an IBM WebSphere eXtreme Scale grid. I'll use a simple example to show how it works. Lets assume we have a Grid with a map called Person containing a String/Person pair. The key is a String, the value is the Person object. The Person object looks like this. The objectgrid.xml file needs to have a schema defined so we can run queries against it. The reason for this is so WXS can do reflection to type check the queries before execution and so on.

Here is a code snippet showing how to use a GridQuery to get all the rows in the grid:

GridQuery q = new GridQuery(ogclient, "select p from Person p", 100);
ArrayList<Object> block = q.getNextResult();
while(block != null)
{
    for(Object r : block)
    {
        Person p = (Person)r;
        System.out.println(p.toString());
        Assert.assertEquals(true, personSet.remove(p.getFirstName()));
    }
    block = q.getNextResult();
}

This shows a client using its ogclient connection to run the query and print out all the answers. The query runs across all data returning entries in blocks of at most 100 at a time. A junit test case exercising it is contained in this sample.

The code to achieve this is contained in the package 'com.ibm.devwebsphere.wxsutils.multijob" and its subpackage. You can just include it in your application if you'd like but make sure to add the RoutingI32Key map which is just used for routing not for any storage. Make sure you have defined a schema also.

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.

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.

memcached versus DataGrid, dumb server versus smart server...

Memcached is a pretty popular free open source distributed cache. It's pretty unique compared with the commercial data grids in that it has a completely unique design.

Memcached is basically a dumb server. This means it's a process running on a Linux box typically which accepts client requests for keys and returns the data. Updates and deletes are also supported. This is ALL the function in the server, hence, dumb server. The client is the only piece which is 'distributed' or even knows about other servers. The clients typically get a collection of host:ip addresses (they all NEED to use the same list) and the clients hash the key over the set of online servers. If there were 4 servers running then all clients will hash data for a key over these 4 servers. Clients will write data for a key K to a particular server using this hash algorithm. One server.

Now, thats what memcached does out of the box. Note, there is no replication on the server side. If a server fails then the clients will hash keys over 3 servers but this changes everything from a hashing point of view so it's likely you'll get a very high miss rate as it's unlikely keys still hash to the same server as before given the modulo part of the hashing is now 3 instead of 4.

Clients work around this by adding a consistent hashing layer on top of the normal client. This works better and writes may be written to a server and the next server in the hash ring. This slows down writes by 2x but allows some recoverability when a server fails. The next server in the ring has the data from the dead server so it's not as bad as before BUT if this server also fails then now the cache has lost the data. It's not replicated remember, it's just written twice. After the first server failed, the clients write changes to the next server and the following server in the hash ring. The new next server in the ring does not have a copy of all the data. It was never written there. It will just have new changes since the failure. There is no active replication going on here.

Most of the materiel on the web shows various workarounds to try to make memcached 'reliable'. A lot of the facebook engineering blog entries focus on this aspect. What they had to do to make it reliable. This works for facebook but won't work in general because I don't think the majority of customers are structured like they are from a build your own mentality. I was talking to someone from one of the big internet shops last week and they always run two memcached clusters for HA. If a server in cluster A fails then they stop using cluster A and switch to B. This works but thats a 50% loss in hardware because of a single failure.

Smart servers

Commercial data grids like IBM WebSphere eXtreme Scale are very different. They handle making the data fault tolerant on the server side with built-in replication and elastic scale out and scale in. Their clients understand the server side data layout or routing. If a server fails with WebSphere eXtreme Scale then no data is lost. The software actively maintains replicas on other servers. Clients that stored an entry with a key K before will still find it. It just works. You don't need consistent hashing on the client, our client does it using routing tables instead. Adding more server processes causes the product to automatically redistribute the minimum amount of data to leverage the new memory/CPU and network available through the new JVMs.

The smart servers allow proper replication which means that the cache isn't as brittle as the memcached cache when servers come and go. If nothing ever fails then the memcached approach works but the potential cache hit rates over time if multiple failures occur mean it can be a liability. The last facebook entry I read on this showed them crawling the logs to try to recover lost data to make the 'next server' more complete. They get this issue and are jumping through hoops to handle it but again, most commercial customers won't do this. They expect the product to do this sort of stuff like replication and they view it as table steaks.

So, memcached for sure has its uses but it's got some limitations which need a fair bit of work by its users to handle correctly if its not exactly what you want or risk some bad scenarios as the cache hit rate climbs. It's very, very fast but you gotta remember, it's not doing much either... No transactions, no consistency, no replication, no isolation and so on. If you don't need these features then it's pretty fast but if you do need this stuff then it starts to slow as you add code on the client side to try and compensate for the limitations.