/dev/websphere

WXSUtils Programming Guide available

I wrote a quick programming guide to using wxsutils.jar in your IBM WebSphere eXtreme Scale applications. WXSUtils is intended as a layer on top of WXS to make programming it much higher level and it provides commonly used function not shipped as part of the product. It will make you more productive developing WXS applications.

The guide can be found on github at this link.

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. 

Solution and explanation for strange swapping on NUMA x86 boxes

This is the best explanation I've read w.r.t. the swapping that occurs on x86 linux boxes with lots of free RAM when using software with very large process sizes.

http://jcole.us/blog/archives/2010/09/28/mysql-swap-insanity-and-the-numa-architecture/

Packt publishes a WebSphere eXtreme Scale book along with other IBM titles

Packt publishes a good introduction to IBM WebSphere eXtreme Scale, it's for V6.0 but is quite applicable for the later versions. They also publish a range of books on IBM topics and are running a 20% off deal for February. If you're interested then head over to their promo page. BTW, I am not affiliated with packt and receive no compensation etc, this is just a heads up to a good deal for books on IBM topics.

UPDATE: Just checked the email informing me of this campaign to double check if they were offering anything and they did say they would provide me a free book of my choice for getting the word out which I'm not taking them up on. So, again, no financial interest in this at all.

"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.

Mobile Applications and FlashLoads, corporations need more modern IT architectures yesterday

I speak all the time to corporate customers with well known and busy web sites. Typically, they see under 200 page views/sec. Maybe, a large one has a million logins per hour. The load curves for those systems usually resemble a repeating bath tub with a peak in the morning then a long trough until lunch time, another peak, then a trough till they leave the office and so on.

The advent of mobile browsers and mobile applications on devices like the iPhone, Android and Blackberry are changing this load profile. The trough isn't so deep any more. There is more average load simply because people on an impulse can bring up a mobile application and check something out. They don't need to be sitting at a desktop anymore.

A higher trough isn't so bad. People who were betting on consolidation will get a slightly smaller payback because as average utilization of servers rise, the potential to consolidate them falls. This is a small negative. The higher trough is small news though compared with whats going to happen to peaks. We have a new phenomenom here, the FlashLoad.

FlashLoads

We've all heard of flash floods. The stars align, the heavens open, and we get severe sudden massive flooding of a piece of land. The advent of wide spread smart phones and the availability of corporate mobile applications to make accessing their infrastructure easier is about to invent a new IT term. The flash load event.

Imagine the following scenario. A bank has a mobile application. It's a big bank, 20 or 30 million customers. They have a successful iPhone application downloaded to 6 million iPhones. Sounds great? But, lets say the stock market crashes. Everyone sees it instantly, CNN's mobile news application sends up real time notifications to their iPhone, they don't even need to be watching TV or listening to a radio. The phone is all they need to be aware of whether Kim Kardashian wore green as soon as she stepped out just now or that there is a stock market crash. What do the people do? Why, they crack open the banks mobile application to check the brokerage application. The problem is all 6 million are doing it at almost the exact same time. They were all synchronized through mobile notification technology and other media if they are observing. It's a flash load.

Corporate IT systems are usually not designed to handle a million concurrent logins let alone 6 million. They are usually not designed to handle 6 million concurrent users pulling up their brokerage account details or entering sell orders from mobile phones at the same time. The velocity of a flash load event will be fierce. The load will ramp up almost vertically if it's triggered by a media event (Cable show, Twitter, Facebook, TV, Radio, mobile notification service). Companies with auto provisioning systems won't be able to keep up. Provisioning more virtual machines takes time, minutes for the virtual machines to start, operating systems to boot, middleware to start up and so on. Meanwhile, you are getting beaten with a massive surge in load and it's likely not going to be pretty. By the time those extra resources are provisioned in, it's already over. Companies that consolidated resources have a lot less headroom than those who don't. You can't expect the headroom when non consolidated resources run at 10% while consolidated ones run at 50-60% load. The headroom just isn't there until the extra provisioned resources arrive and it won't be instantly.

Caching is a possible solution to some of these problems. The closer to the user that you can service a transaction using a cache then the less of the infrastructure you will touch with each 'transaction'. Caches can usually handle a lot of load, tens of thousands of requests/sec even on blades. A little investment here and having the data preloaded or warm in the cache could be the difference between a severe outage and a working site. I don't think it's a question of whether to cache or not, it's a question of cache or crash.

The cloud may not save companies here either. Very few corporate applications can run in a hybrid cloud/private data center mode. The cloud also has limited capacity. If the major banks used Amazon EC2 to host servers, a flash load event may even strain their ability to provision ec2 instances to try keep up. They are all looking for extra capacity at the same time. It's like the recent credit crisis but it's a mips crisis instead.

I think mobile applications will allow customers to check their accounts and details in a much more accessible and convenient way than the Web ever did. But, with the availability of instant information in our society through mobile devices, they also bring the potential for a new phenomenon. The flash load. Be prepared once you start shipping mobile applications for your public facing services. It's great to see a successful smart application getting millions of downloads but remember as your success measured by the number of downloads grows, the potential of a flash load is also rising fast too. Be prepared before you discover the limitations of a traditional multi-tiered architecture. It's not just a matter of adding REST services or similar, you need a flash load strategy also because it's guaranteed to come along eventually. It's cache or crash.

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

Chatty cache syndrome and solutions

There are a lot of applications out there that already use local caches like the many open source ones as well as things like Dynacache. These caches are basically local hash maps in the same address space and effectively have zero access time and cost. Cache misses are zero cost and cache hits/updates are practically zero cost also.

Applications faced with this have been written to expect this speed, i.e. zero cost. This means some applications are written to be extremely 'chatty' with the cache during individual requests. The caches tend to be very fine grained and a single application request may request or update hundreds of key/value pairs from the cache. When the cost of each operation is practically zero then this isn't a problem at all.

The trouble starts when you try to plugin a distributed cache. These caches are typically remote. This means all cache interactions involve an RPC and involved serializing the keys and values to wire format and possibly back again and thus are far from zero cost. Comparing the cost of an RPC with a local hash map lookup, it's a pretty huge difference in code path. Maybe 3 or 4 orders of magnitude. If the application is doing 100 lookups per request then before it was basically 100 x zero ms, now it might be 100 x 1ms even for misses.

This difference can then have a dramatic impact on an applications performance if it interacted with the cache in a very chatty manner. I see this a lot in applications written against a local cost cache because if something is free then developers will use it and then abuse it.

There are a couple of solutions to this.

First, we can modify the application to be less chatty. This typically means the application should store data in the cache in a much coarser form. Rather than have a customer split across 10 maps, a single customer map whose value is a denormalized 'customer' data might be better for example.

Another solution is to use a distributed local cache. WebSphere eXtreme Scale can do this using it's multi-master capabilities (V7.1+). This allows you to set up a single partition grid. The grid is hosted in a pair of container JVMs that we'll call a hub. Each client then instead of just connecting to the hub with our client APIs, they would instead create a local grid with the same objectgrid.xml/deployment.xml but also with a local catalog instance in the client JVM. The client then makes a multi-master link to the hub grid catalog. WXS will then copy everything in the hub grid locally. The client then reads and writes against it's local grid which is now like a spoke in a hub and spoke topology. All changes made to the clients grid are replicated asynchronously to the hub. If other clients did the same thing then the hub will ensure those changes are also copied to the other clients. Any collisions can be handled in the hub using the default arbitrator or an application supplied one.

This approach allows clients to have zero cost reads and writes to a cache bounded by JVM heap limits but it's fully replicated across different clients using a hub and spoke architecture. It would use optimistic versioning through our arbitration logic in the hub. Thus, it's extremely fast but the capacity is limited to the data that fits in a single JVM although this can be quite large these days.

But, the bottom line is that treating a cache as a zero cost access resources is a mistake. There are lots of types of caches these days, the caches can be made to scale out linearly from a capacity and throughput point of view but application developers need to realize this fact and either use coarser grained cache lookups or the local approach with some kind of hub/spoke but with a capacity restriction as it's unlikely a client JVM can host 600GB of data. It is possible in this day and age to have such large JVMs but if there are a lot of clients then cost of memory becomes a deciding factor and moves you back to the remote large cache with a less chatty application/cache interface.