/dev/websphere

Distributed grids as backing store for Lucene Indexes - possible solution

I'd tried to use a distributed grid with Lucene before. I was at a customer with the competitors doing a face off type engagement. We (the competitors and myself) both implemented Directory implementations for Lucene on top of our respective caching products. The implementation I wrote is on github here. The customer had a very large lucene index in memory, abouit 30GB and they were seeing very large GC pauses (minutes) using the RAMDirectory supplied with Lucene. The plan was to use a grid based directory to move the indexes in to a shared remote grid.

This proved much slower than the RAMDirectory the customer was already using. Lucene was too 'chatty' with the Directory and this means lots of RPCs to the grid. The Directory abstraction is too low level. We tried solutions like near caches but depending on the query, the near cache could need to be very big thus making the solution not viable. This ultimately meant that the customer didn't go forward with either grid as a directory store.

However, I was reading about Solandra recently and they have what looks like a better approach. The problem we had was that just implementing a Directory is too low-level. There ends up being too many RPCs between the Lucene engine and the grid. Solandra implements a Cassandra store underneath Lucene. They did not implement a Cassandra directory which would have had the same issues as I just discussed. Instead, they implemented a higher level plugin. They moved up to implement IndexReader and IndexWriter and this allows them to push some of the search down in to Cassandra's native index/query support rather than just use it as a block store. This seems to reduce the number of RPCs between the store and the lucene engine considerably.

This looks like a much more performant approach and one that should work well with a data grid also. If I get time then I may try to implement this with a datagrid and post results.

Extreme write behind with JMS

I've been working with one of our biggest customers and they have a high quality problem in that under extreme load, the grid (WebSphere eXtreme Scale) is holding up just fine. but there is a problem. The issue is the database the grid runs on top of. They are using write behind to asynchronously write changes made to the grid to the database. This basically will only write changes to the database every 5 minutes or 1000 entries depending on what happens first.

The issue is even with this batching occurring, the database cannot keep up with the write load. They could buy a bigger database but all they really use their database for is a disk based log for whats in the grid. They load the grid from the database and then all applications use the grid as the system of record for that data. The updates made to the data in the grid and written back to the database asynchronously.

If the database cannot keep up even with the batching then it starts to fall behind and more importantly, the buffered changes not yet flushed from the grid also build up in memory.

Buying a bigger database box is also kind of pointless as the whole point of the grid was to avoid having to buy a large SMP box to run the database. Scaling up is expensive and database boxes with their SANs that can keep up with extreme loads are not really what we want to purchase. We want the grid to act as a shock absorber so that a smaller database can be used. But, this is proving a problem at this customer just using normal write behind techniques.

The real problem here is databases are not designed to be used like this. WebSphere eXtreme Scale is basically using a SQL database as a write only disk store and frankly, databases are not very good at this.

So, what is designed as a write only disk based store? Believe it or not, a messaging system is. JMS queues are designed to implement two operations. PUT to the queue and GET from the queue and they can do this very efficiently. More efficiently than a database. This is all the grid really needs here. Flushing the writes from memory to disk through a database doesn't make a lot of sense.

The idea here is to put the JMS queue between the grid and the database. This way, the grid can flush the changes to 'disk' faster than before and then we can apply the changes from the queue at our leisure to the database. This decouples the database and the grid. The grid can now deal with flash loads or tsunamis of load when they occur and flush the changes to disk using the messaging system. The database can apply those changes at it's own slower rate because the JMS queue is buffering the changes in a durable way to disk. Thus, we don't need to spend a lot of money on an expensive vertical or horizontal scaling DBMS.

This is very like a seat belt in a car. We're trying to absorb a shock by stretching it out over a longer period of time. The JMS queue allows this to happen. While the flash load is happening, the queue is able to keep up with the grid and everything stays stable. The database is emptying the queue as fast as it can which is slower than the grid. Once the flash load is gone then the grid goes back to normal load while the DBMS can keep applying changes for another couple of hours if need be.

The implementation of this could be a Loader that writes the changes to the JMS queue. A JMS listener can then pick up the changes are apply them to the database. This is basically like the write behind built in to WXS but implemented instead with an external JMS queue.

Best practices for setting up an IBM WebSphere eXtreme Scale Grid

I'm maintaining a live google document with the best that we're seeing. You can see the document at this link http://bit.ly/wxsbestpractice.

There's more to configuring ThreadPools than thread pool size

We just found a customer issue and it's worth talking about. The customer was preloading data into WebSphere eXtreme Scale. They were multi-threading the preload to speed it up. They were using a default fixed size ExecutorService as the pool. They were fetching records from the database and then doing a submit call on the pool to hit the grid.

They were running out of memory on the preload client. The reason is as follows. The thread pool size was fixed. Once those threads are busy then queueing additional items just puts them in the queue. The queue was getting longer and as a result holding more items. Items in a preload case might be 10k records to push in to the grid. Items are big, they can take a lot of space. A lot of space if you are able to grab stuff from the database very quickly.

This means if the thread pool is undersized then the pool will queue up lots of items and when the items are big, your preload client can run out of memory pretty easily. There are a couple of things to do here. But, the first is to fix that thread pool so that this can't happen. If you initialize the pool like this then you'll get a better behavior:

// this means that once there are 3x numThreads jobs queued waiting for
// a thread then it will start running jobs on the submitting thread.
LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(numThreads * 3);

// Once the queue reports that it is full, the CallerRunPolicy will run job
// on the submitter thread.

ExecutorService p = new ThreadPoolExecutor(numThreads, numThreads, 2L,
                TimeUnit.MINUTES, queue,
                new ThreadPoolExecutor.CallerRunsPolicy());

This shows us creating a LinkedBlockingQueue with a specific maximum capacity. I used 3 times the thread pool size but you need to figure out what works best. The major change here is the CallerRunsPolicy parameter. Now, that the thread pool uses a fixed size queue then if the queue fills then normally the pool will throw exceptions when this happens. We'd have N threads + M items on the queue at that point. CallerRunsPolicy means that instead of throwing an exception, the job is instead executed on the submitter thread. This will prevent the memory runaway that was happening in the case I described earlier and will throttle the preload operation itself rather than let it go crazy.

The other things to watch here would be to use larger batch sizes. If you are using WXSUtils.putAll_noLoader to bulk load the items in to the grid then make sure you have around 1000 key/value pairs PER partition. This is a good rule of thumb to start from when tuning things. If you have 200 partitions and a batch size for PutAll of 1000 items then thats only 5 per partition which isn't enough to really start to get the benefits of batching the puts. A number per partition in the hundreds is probably better.

If you have a grid with 200 partitions then a thread pool size of 32 isn't great either. You should at least have the same number of threads otherwise you'd overrun the pool instantly when doing any bulk operation. The default pool size for WXSUtils is indeed 32 and is too low for production. You can specify the size of the pool using a constructor parameter or by editing wxsutils.properties if thats how you are connecting to the grid.

I've already updated wxsutils to configure the threadpool as described above given recent experience. I have not changed the default thread pool size from 32 but thats something I'm considering also. Maybe, I'll default it to one or two times the number of partitions in the grid but bounded by some 'reasonable' number.

Anyway, it's worth pointing out the problem and how it's resolved just for every bodies education. An unbounded thread pool happens not just because of a growable thread pool for example but they can be memory unbounded pretty easily as in this case when the pool cannot service the pool queue fast enough.

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.

Keeping data in IBM WebSphere eXtreme Scale grids in sync with a database

This is a very common question. A customer wants to keep a copy of important data in a DataGrid, like IBM WebSphere eXtreme Scale. The data is sourced from a database typically. The data is changed by legacy applications in the source database only. The customer wants to know whats the best way to keep the DataGrid in sync as data changes in the database.

A great solution for this is the IBM Infostreams Change Data Capture product. This product installs on the same box as the database and it can watch the database for changes. It watches for changes usually by watching the transaction logs. This is very lightweight and typically has little or no impact on the database process. It can also watch using triggers but this is going to have an impact on the database.

Once CDC spots a problem then it can send a message to a JMS queue to record the fact that a record in a table that is being watched has been modified. The customer then writes an JMS message listener to listen to that queue and apply the change described by the message to the data in the DataGrid.

This is also a great way for buffering changes made to very large datasets when the grid is down for maintenance. We can stop the JMS listener. CDC will still buffer changes on the queue when the listener is disabled. Then we would write a disk snapshot of the grid using wxsutils. Do the maintenance. Restart the grid, reload the snapshot and then enable the JMS listener. The JMS listener will then bring the grid up to date with the changes buffered during the maintenance interval. This will be much faster than a full reload from the source database when the dataset is large or the source database is a smaller database (i.e. it's not that there is a lot of data, the database is just slow getting it all out quickly).

Anyway, this is something I'm going to start recommending every time moving forward. I think it's a great solution, not very invasive and is pretty simple to get working.

Creating disk snapshots of an IBM WebSphere eXtreme Scale grid and why they are really useful

I've been working with a few customers who are using WXS to load large quantities of data in to the grid from a database. This can be quite time consuming as fetching the data from the backend is usually the bottleneck. Loading hundreds of gigabytes of data can take well over an hour. This means that a full grid restart is not something such a customer wants to do a lot given the cost of reloading the data.

The bottleneck here isn't WXS, it's the source data system. Many of the customer scenarios I see use the grid as the system of record. If the grid is down then the data doesn't change. We'll discuss when it does change later.

I've written a utility as part of wxsutils (if you're not using this with wxs then you are missing out...). It will make a disk based snapshot of a grid map. It writes each partition for each map to a file by writing the key/value pairs as JSON strings. Each physical box hosting the grid container JVMs will store all the partitions hosted on that box in files in a particular folder on that boxes file system. The current code does it for the maps you specify. So, just execute the code once for each map you want 'snapshotted'.

Later, the customer can execute the read snapshot code to reload that data set from those files in to the grid. This will happen at grid speed. Each box in the grid will read the local files and load them in to the grid as fast as possible. The bottleneck here will likely be network depending on the boxes.

This approach can save a lot of time in many scenarios:

• The customer wants to cycle the whole grid for maintenance on the WXS runtime or the application artifacts. The customer should snapshot the grid before stopping it and then read the snapshot after restarting the grid post maintenance. This is much faster than a reload from the backend.
• The customers wants during testing to have a grid with a well known start data set. Load it once, snap shot it and then use it to reset the grid every time you want to run a test case from a well known start state.

If the data is changing all the time even when the grid is down then we can deal with that also. Probably the 'best' (and best is a loose word here) way to synchronize a grid with a database is by using a change data capture type product. For example, here is IBMs link. These products watch the transaction logs of a database and allow an application to intercept changes in the data without slowing the database down or using triggers etc. The application listener would then typically make a message about the change and send it to the grid where the grid will process the message and apply the change. This keeps the staleness between the database and the grid to a minimum.

If this listener stores these messages in a JMS queue or MQ series queue then even if the grid is down, these changes are tracked. Thus, when the grid is restarted, do not start the msg listener straight away. Instead do the read snapshot and then start the listener which will down apply all changes since we wrote the snapshot. This allows a delta update of the data in this case keeping the snapshot useful.

You can see how to use this snapshot capability on github at this link.

You will need to add the PerContainerGrid definition to your objectgrid.xml and deployment.xml files and put wxsutils.jar and the jackson json jars also on your container classpaths.

Prototyping criteria API in wxsutils on github

I uploaded a draft of the beginnings of a JPA criteria style API to filter data in WXS to github today. I searched for an implementation that allowed someone to just check a POJO against a Filter and it doesn't exist as far as I can tell.

So, this is a first stab at it. Next step is to look at ways to use it in wxsutils for queries, continuous queries and other stuff. It's in the wxsutils 1.10 level jar.

It's pretty easy to play with right now, just use a FilterBuilder to construct a filter using the conventional dot notation. I'll work on integrating it in for use in Map/Agent operations next.

Adding conditional put to wxsutils library for better concurrency

Updating a shared key/value pair in a concurrent environment is usually done in WXS in a pessimistic manner. The flow looks like this:

Begin

    V = m.getForUpdate(K)

    V++

    m.update(K,V)

commit

This gets a lock, reads the value, modifies it and the updates it. The commit releases the lock. This ensures a serializable operation in the sense only one client thread at a time will be able to modify the value and ensures correct operation.

The common problem with this approach though in a contentious environment is threading. A client will use a server JVM thread to do the getForUpdate and it's blocking. We have lock timeouts to time bound the blocking time but lets ignore that for now. Once the lock is acquired, the server thread is released and the client gets control back. The client then sends a request to do the commit which uses another thread temporarily to commit the transaction releasing the lock and allowing others to get the lock in the future.

The problem here is if a lot of clients go after a small group of key locks then the thread pool can become drained because all threads are currently blocked waiting to acquire some shared lock. Those threads can only proceed if they get lock timeouts OR the client holding the lock can commit their transaction thus unlocking the key. The problem is if the client can't get a server thread because all server threads are used waiting for the lock then we end up with a problem because we can't get a thread to execute the commit for the client holding the lock so the lock never gets released. This is why lock time out is important to prevent server threads waiting forever for a lock.

Now, there is another way to do this and it's an optimistic approach. The problem above is that server locks are being held across client requests. Another approach would be for the client to use code like the following:

Loop

    V = m.get(K)

    V2 = f(V)

    if(m.cond_put(K, V, V2) == T) exit loop

End Loop

The code snippet shows an optimistic version. Here, the client gets the current value and remembers it in V. The modified value in kept in V2. There is a new primitive on wxsutils#WXSMap now called cond_put. This takes the key, the old expected value V and the new value V2. The value for K is only updated to V2 if the current value is still V. A flag is returned to indicate whether the entry was set to V2 or not. In the above case, we use a loop to repeat the operation in the case.

This approach results in locks on the server being held for much smaller periods and never across client RPCs. The client needs to use a loop but i think this approach scales better in a contentious environment than the usual approach.

It's also similar to the concurrency techniques used for highly concurrent data structures now. Most of these are based on a compare and swap algorithm which is very similar to cond_put above.

The implementation for a cond_putAll is now in my github wxsutils library if you want to try it out. 

"Penny wise and dollar foolish" applies to distributed caching too often

Many times I've had people compare the performance of WebSphere eXtreme Scale to a local hash map and comment that it's slower. Basically, they are comparing a remote hash map call to a local hash map call and yes, the remote one is much slower. But, this is an apples to oranges comparison. Try caching 600GB in the local hash map on every JVM requiring a cache? But, hey, it's faster, right?

Wrong. If you just need a local hash map then you should use one. A distributed cache like extreme scale brings expandible capacity, high availability of data in the cache, a network accessible remote very large cache, cache stays up when the application jvms are cycled and so on. It's different and thats why the fact that a local hash map is faster isn't really valid.

For a real world example, lets look at WebSphere Commerce Server. It can cache HTML page fragments in a local map (Dynacache) and in extreme scale also. If the cached data in present in both caches then the local map is faster but, that's not why we sell extreme scale to commerce customers.

An extreme scale cache is remote. You can cycle the commerce JVMs and when they restart, they are working with a hot cache straight away. This isn't the case with the local cache. It has to be regenerated and this costs of lot of CPU. This regeneration also hits the database during this time. So, we have extra costs of CPU, longer response times and database load if we cycle a commerce JVM and have to rewarm up its page cache. None of these problems exist with extreme scale as the page fragment cache because the cache contents isn't kept in the commerce JVM at all and therefore it survives a commerce JVM restart.

Now, imagine bringing down a box with 10 JVMs on it. With the local cache, the cost is 10x when you restart the box. With extreme scale, you has the cost of restarting the JVMs but no cache warmup costs at all as the cache survives the JVM/box cycle. It's ten times faster because those ten JVMs do not have to warm up their ten independent page fragment caches again or hit the database and the customers using that box see "normal" response times pretty much as soon as they are online versus waiting for the JVM caches to warm up in the conventional dynacache scenario.

Next, lets look at invalidation. If a large site had 100 JVMs and I update the price of an item, then all 100 JVMs need to regenerate the page fragments for that item every time this happens. They all burn CPU, the response time for serving that item is longer and there is database load. With extreme scale, the page fragments are invalidated but are only regenerated once, the first commerce server to request the page notices that it's not there, generates it once, puts it in the cache and now the other 99 JVMs have the new page even though they didn't generate it. This is a massive load reduction when prices or inventory levels are updated or promotions for product categories become active potentially impacting tens of thousands of skus in the system and the page fragments for those impacted skus. This means you can invalidate prices to match competitive prices in a more timely manner without worrying about high CPU spikes or database spikes. This is a major advantage for extreme scale with commerce server over the dynacache implementation.

Lastly, if each JVM has its own independent cache then potentially depending on which JVM a request is routed to, a different cached page fragment could be served, if pages are being invalidated aggressively then the site becomes a bit of a mess in terms of consistency of what users see across the cluster. WebSphere eXtreme Scale is a single logical cache shared by all commerce JVMs, they don't have a private local cache, they always go remote to extreme scale for page fragments. This means they are always consistent and a user will see the same cached page fragments no matter which JVM they are routed to.

So, when examining whether a customer should use websphere extreme scale with commerce server, it's important to focus on the massive response time, cpu savings and database savings benefits around edge cases, not steady state performance. The steady state CPU load of a commerce cluster with no JVM cycling, additions or data invalidations is likely to be a little higher with extreme scale versus a local dynacache. This is expected. If you have a completely static commerce site then dynacache with disk offload is probably a better solution. But, and I expect this is most customers, the content of the site changes and you want:

• users to see consistent page contents
• you want consistent response times if you need to ripple commerce JVMs or add JVMs
• large cpu and database load reductions when prices/inventory levels or promotions change
• more consistent response times means more satisfied customers versus a web site where depending on whats going on (edge cases), customers see widely variable response times.

Then extreme scale is the way to go, hands down.