/dev/websphere

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.

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.

How to do online runtime upgrades for IBM websphere extreme scale

Ibm websphere extreme scale (wxs) has many high availability features designed to allow data grids built using it to stay up 24/7 indefinitely. One of those features is the capability to upgrade and apply maintenance to the product without shutting it down. IBM WebSphere eXtreme Scale has had this capability since V6.1.

The basics behind doing this are as follows. WXS has 3 logical components, the catalog service and the containers and the clients. These need to be upgraded in order.

Upgrade Catalog service JVMs

First, stop one catalog server JVM at a time, apply the service and then restarting it. Repeat this step for all the catalog server JVMs, we usually recommend 3 in production. You should complete this step quickly but it's important to wait for replication to finish before moving from one catalog JVM to the next JVM. For example, if you are running 3 JVMs as catalog JVMs (and I hope you are...) then you have basically one primary and two sync replicas. If you stop the primary first then you should see a replica become the primary and the other enter peer mode. Now restart the old primary JVM and wait for it to enter peer mode as a replica. Now, stop the next JVM and again wait for a replica to become a primary for the other to be in peer mode. Repeat this procedure for the last JVM. These checks ensure that at any point in time there are two fully replicated copies. You should finish the catalog upgrade once you start. Don't start and complete the next day, for example.

Upgrade the containers

The next step is to do the same with the containers. Typically, customers run multiple container jvms per machine on top of a single install directory. All jvms on a single directory should be upgraded at once. This means shutting them down, applying the service/maintenance and then restarting them. Customers should use the xsadmin command teardown to stop all jvms on a single box. This is the most efficient approach. It let the catalog service know that all jvms for a certain ip will be stopping and it does it in the most efficient manner. This step is then repeated for every box running container jvms. When moving to the next machine, ensure the grid has settled from shutting down the previous box and all replicas are up to date. You don't want to shutdown your only complete copy of the data in a shard by accident. This especially applies to customers using large 64 bit JVMs as containers as creating a new replica JVM can take some time when you use large shard sizes.

Finally, upgrade the clients

Now that the catalog service and containers have been upgraded, you can upgrade the clients at your leisure.

Things to be careful of w.r.t. install procedures

One issue that we have seen is during the catalog step. Customers sometimes will install wxs once on a server/machine and then run a catalog and containers from that single install. Don't do that. It's worth saying again, DO NOT DO THAT.

If you do then here is what will happen. You will need to stop containers and the catalog jvm using that installed folder because of file locks on jars/libraries and other files etc that the JVMs running the catalog and containers are using. You will upgrade the WXS installation and then restart the catalog. You will now discover that you cannot restart the containers because all the catalog jvms have not been all upgraded yet. If you have a small grid then this can be fatal. Imagine a two box scenario for example. There would be no way to proceed as to upgrade the second box, you'd need to stop the catalog and containers there which means you now have no containers running and you lose data. Simply, using two install folders on a box that hosts both a catalog jvm and container jvms solves this problem. This applies whether you install WXS on top of J2SE or on top of WebSphere ND.

Summary: WXS allows 24/7 operation and online runtime service without a restart

So, there's it is. You can apply service or maintenance to IBM websphere extreme scale without a restart by following the above steps and precautions.

State of the art multi-master replication in IBM WebSphere eXtreme Scale 7.1

IBM WebSphere eXtreme Scale already supported a single grid running in multiple data centers, even over a WAN. This was an implementation of CP (see the CAP theorem for details). We've just released V7.1 and are now the only product on the market implementing both single grid across multiple data centers as well as now supporting mirroring independent grids over WAN links.

Multi-master Replication

This new feature allows a customer to host a grid in multiple locations connected through user defined links. Each grid is fully independant and runs it's own catalog service. The locations need to have the same grid defined with the same number of partitions and map/template definitions. The customer can create a link between two locations and from that point forward, WXS will try to make both locations identical. You may also see locations referred to as domains in the documentation. This isn't just a trigger forwarding changes from one side to the other like the competition, this is true replication. If you create a link between a location with data and another empty location then WXS will copy the data from the non empty location to the empty location automatically.

These replication links are bi-directional. Changes made on either side of the link will be propagated to the other side. Besides, the simple topology of two locations, more complex topologies can be constructed. For example, a 'line' topology is easy to implement. If we have 3 locations, A, B and C then make a link between A and B and B and C and all three locations will be replicating to each other. C will get changes from A through B and A gets changes from C through B. A ring topology can be readily constructed also by closing the loop by creating another link between A and C. A ring has the advantage over a line in that changes can be pulled either way around the ring (we do both ways simultaneously actually). This means a link can fail and data can still be pulled the other direction. Rings are more durable than lines for this reason. Changes are sent around the ring in both directions and the latency before changes from a node A appears on another node is dependent on how many links are between them.

Hub and spoke topologies can also be constructed using the same approach. Imagine a location H as the hub. Locations around the hub can simply create a link between themselves and the hub. For example, three locations A, B and C. We would make links A-H,B-H,C-H and this gives us a simple hub/spoke topology. You can use links to create almost any topology you want. If a link goes down then replication is suspended along that link until it restarts.

This is not queue based replication

WXS does not use a queue for replication. We use an approach using constant memory independent of how far back a location becomes out of sync as well as the number of links defined to other locations. This basically means you don't need to worry if a link is down for a while, you won't run out of memory buffering changes like some of the competitors would.

No gateways for replication

Each container JVM communicates directly with like shards in the other location. There are no gateways to define. If you want to define a concentrator then just make a 'fake' location with a link to the left grid and the right grid and presto, it's a gateway but you don't need to make one.

Collision Handling

Collisions are a way of life with a distributed system using AP characteristics. WXS uses a arbitrator to resolve collisions. It has a default arbitration algorithm which simply picks the change from the lexically lower location name. Very simple but will likely be enough for many customers. WXS can also use an application provided arbitrator that makes a new version reconciling the conflicting versions received. You should designate one location as the principal arbitration location so there is a single 'truth' resulting from any collisions.

Summary

The multi-master support in IBM WebSphere eXtreme Scale V7.1 complements the existing state of the art single grid multi-data center support already proven in the product. It allows data centers to be more independent and still provides application level consistency through arbitration. It's implementation is very scalable and it's designed to use constant resources regardless of the topology or reliability of the links between locations. It's also built in to the product.

I'm looking forward to working with customers to exploit both styles of multi-data center replication.

You can read more about multi-master replication in the V7.1 info center.

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.

Update: RedHat 4.0 includes NFS v4

An update for me anyway, didn't realise this but its there. So, excellent news for linux WebSphere 6.0 customers looking for easy high availability.

http://www.redhat.com/software/rhel/features/

NFS v4 on Linux almost here

While the linux kernels have had NFS v4 support in for some time now, the major distros are looking at adding for this soon. LinuxWorld is doing a cover story this month on the testing effort.

This is a big deal for us as WebSphere 6.0 can take advantage of this technology to deliver very low cost highly available OLTP clusters. NFS v4 is currently only supported on AIX 5.3 and Solaris 10. Customers can use third party file systems with lease based locking and force write semantics but I think Linux support for this by the major distributions will help enormously.

http://www.linuxworld.com/read/86018.htm

Single box high availability is possible

Most people that see the usual "This is how you make a system highly available" see a bunch of boxes arranged in pairs and then everybody relaxes. But, this is a very complex setup and, as most math majors know, 2 of anything is twice as likely to fail, so whats 14 of something?

Disk virtualization

We've had this for a while now. My disk system in the lab is a Network Appliance F940 with 28 disks on it. I use iSCSI over Gb ethernet to hook it up to my Linux and Windows boxes that need high speed disks. I basically don't worry about how many disks I need anymore. I just made a large 500GB RAID volume on the NetApp and then parcel it up in to logical disks (LUNs) when I need them. For example, if a database box needs 20GB of disk for data and 5GB for logs then I create two more LUNs from the 500Gb and then assign them using iSCSI to the db box. The database doesn't care how many physical disks etc the 20GB/5GB disks use, the NetApp virtualizes that and it basically looks like a kick ass disk subsystem. We can now start to do the same stuff with operating system images as we'll see. Imagine, I had a box with a collection of processors, memory, network cards, and disk controllers. Why couldn't I do the same thing with this? You now can.

Virtualization to the rescue

High end SMP (and soon Intel boxes) have virtualized hardware now. The machine comes with whats called a HyperVisor whose job is to manage the hardware on behalf of the operating systems running on the box. Many people are familar with VMWare. This virtualizes a PC and allows you to then install Linux or Windows in to that virtual PC. You can then have 4 virtual PCs each running Linux running on a single real PC. Hypervisors take this to a new level.

IBM Power 5 and virtualization

Lets look at an IBM power 5 box. I have one in my lab. It's got 4 CECs. A CEC is a building block. Each CEC has 4 CPUs (mine are 1.9Ghz), 16GB of memory, disk controllers, network cards etc. The CECs are connected together to form a single 16 way SMP with 64GB of RAM. More CECs build bigger boxes. There is another box (a small PC) on the side called a HMC. This controls the hardware. The HMC lets you create a virtual PC or in IBM speak, an LPAR (Logical Partition). You can then assign it memory, CPUs, network cards, disk controllers. Then you can boot that LPAR and install AIX in it or Linux. The resources assigned can only be used exclusively by a single LPAR at a time. But, and here is the cool bit, you can change the number of CPUs assigned to an LPAR when ever you want.

HA with LPARs

So, lets say we wanted a highly available database, IBM DB2 UDB for example. Normally, you'd have SAN based disk and two boxes with HA software like HACMP installed. This is wasteful because the second box does nothing until the first box fails. We can do better with the LPARed system.

Lets use a single 4 way power box. Define two LPARs. Give each LPAR half the memory. Call one LPAR P and the other S. Give P 3 CPU, give S one CPU. Setup HACMP just like there were two real boxes.

Once you're done, you've got a HA setup that probably better than two boxes. The single p5 box has redundant power, redundant LAN, and redundant fiber cards. The memory has fault tolerant features. CPU failure can be detected and then masked as bad. One processor is being used to run the backup system/LPAR so we're still 'wasting' 1/4 of the box for the back. But, it's possible to assign fractions of a processor to an LPAR. For example, we could assign 3 3/4 CPUs to the primary and a 1/4 processor to the backup. Now, we're only wasting 1/16 of the box. The minimum LPAR size depends on how much CPU the monitoring takes. The fractional CPU deal is called micro partitioning. But, you need to be careful with micro partitioning, it can be a performance killer because it can increase the amount of CPU cache thrashing which can hurt performance on that processor.

Now scale it up

This pattern can be expanded to more complex configurations. Lets say we have a 16 way box. I'd make 3 AIX 5.3 LPARs (A, B and C) of 4 processors to run WAS 6 ND. I'd make another 2 AIX 5.3 LPARs (X, Y) of 3 and 1 to run DB2 and NFS v4. I'd use HACMP to make X and Y fault tolerant and use HACMP scripts to tell the HMC that when X fails over to Y, dynamically change the CPUs for X to 1 and Y to 3. So, the backup LPARs use minimum resources until they become the primary and then we give them the primaries resources (at least in terms of still functioning CPUs).

This kind of flexibility with LPARs and being able to dynamically change the number of processors assigned to an LPAR time slice by time slice makes these larger boxes much more economical than before. The price of 4 ways versus 2 ways has collapsed to be basically linear, 8 and 4 ways is almost there. 16s will likely follow.

Summary

So, LPARs bring new options to the high availability table. These options existed in the mainframe world before and in limited forms on the older generations of Sun, HP and IBM machines. The latest virtualization technologies are much more flexible and will increasingly advance on the 390 full blown virtualize everything model. Probably worth doing a google on now to learn more about it.