Nati Shalom's Blog

Using SSD As A Foundation For New Generations Of Flash Databases

“You just can't have it all” is a phrase that most of us are accustomed to hearing and that many still believe to be true when discussing the speed, scale and cost of processing data. To reach high speed data processing, it is necessary to utilize more memory resources which increases cost. This occurs because price increases as memory, on average, tends to be more expensive than commodity disk drive. The idea of data systems being unable to reliably provide you with both memory and fast access—not to mention at the right cost—has long been debated, though the idea of such limitations was cemented by computer scientist, Eric Brewer, who introduced us to the CAP theorem.

The CAP Theorem and Limitations for Distributed Computer Systems

Through this theorem Brewer stated that it was impossible for any distributed computer system to be able to provide users with these three following guarantees simultaneously:

• Consistency (Every node will be able to view the same data at the same time)

• Availability (Every request will receive a response)
• Partition Tolerance (The system will continue to operate even if the system faces any arbitrary failures)

 We can evaluate different approaches for data management solution by the above three properties and the tradeoffs that each one of them faces when utilized over the other. For example, what is the tradeoff of putting more emphasis on consistency? The trade off will most likely be less availability or partition tolerance.

 Traditional RDBS solutions offered consistency over partition tolerance and cost. In-Memory caching solution such as memcache offered a different set of tradeoffs, for example speed over some degree of consistency (between the in-memory data and the data held in disk).In Memory Data Grid or In Memory Computing, as it is called today, extended the use of memory as the system of record and thus enabled higher degree of consistency by reducing the dependency on the underlying disk databases. The new generation of NoSQL databases took a different approach and offered speed and scale at low cost over consistency by utilizing distributed commodity storage and relying heavily on asynchronous data flow to speed up the data processing.

 Quite often there is a direct correlation between those tradeoffs and cost - for example high degree of consistency often relies on synchronous data-flow and replication operations which often come at a cost of speed. To overcome the speed limit we often need to use memory or other flavors of high speed storage which often translate to high cost. On the other hand if we can compromise on consistency, we can offer speed and scale using commodity resources (  without relying on high cost resources).

As technology advances and demands for cost effective and scalable, reliable data increase, new solutions are coming to light, with one of the most favored being a combination between two items we are already familiar with: RAM and flash.

 

Why SSD Is a Viable Storage Solution

Many of the underlying assumptions in the previously mentioned alternatives that  was built under the assumption that disk is the bottleneck. If disk is the bottleneck, we needed layers of optimizations to minimize the access to disk. SSD provides a high speed storage device, which gets rid of the idea that disk is the bottleneck. This is a fundamental change in one of core assumptions behind the design of many databases today, let me explain..

 For example if the disk is slow it makes sense to put various filters to minimize the access to disk. As disk gets faster all of those filters becomes an overhead. In many cases it would be faster writing directly to disk without those additional layers. In similar way the use of asynchronous operations need to change as well. When the disk is slow it makes sense to use asynchronous operations to deffer write operations. However, when the disk is no longer the bottleneck it would be faster and vastly simpler to access the disk directly. This also allows avoiding many of the consistency tradeoffs that i mentioned earlier that are often a result of those asynchronous operations.

 

Putting SSD and RAM Together using off Heap Storage

 RAM storage can be as big as necessary and it is incredibly quick—but it costs about ten times more than a flash disk. By putting SSD and RAM together we can optimize the cost/performance ratio. The main challenge in doing so, remain around consistency-i.e. how do we keep the data in-memory and flash in sync so that from an end user perspective this integration would look completely seamless?

One method of such integration is referred to as off-heap storage.  

Off Heap storage is often implemented as a plug-in that is used by the memory data-store to offload its data from the actual RAM into Flash and visa versa.

Traditionally Off Heap was implemented with shared memory which is basically a block storage that provides external access to the same RAM device. By doing so, we bypass the management overhead of managing data in RAM through JAVA. The limitation of that approach is that the capacity is still limited by the size of RAM. It also forces an external data management and garbage collection layer to manage this external heap.

With SSD, users can now have both the memory and the SSD device synchronized and regularly kept up-to-date, meaning storage is scalable and consistent.

Some of the SSD drivers provides a key,value interface which makes the integration vastly simpler as it takes care of its own data  management.  Having said that, most of the In-Memory implementation that uses SSD as off-heap storage provides fairly limited functionality in terms of query and transaction support. This is due to the fact that they( What is they?)  integrate with SSD through a disk storage interface.  

 

Using SSD as a Foundation for Flash DataBase

 To really make the best use of SSD and maximize its potential we need to think of SSD as foundation for database. This requires a more tight and native integration with SSD in order to overcome some of the limitations that exist with many of the current In Memory Data Grid and SSD implementations.

 The specific set of features that are needed for this sort of integration includes: 

• Portable and native Key/Value interface that works against any flash device.
• Using Flash as Durable Storage - Flash can be used as a persistent data-store and not just as extension to RAM. As such we can leverage the durability of flash to speed up the load time of data from the underlying flash device incase of a planned or un-planned recovery process.
• Support for batch operations - batch operations are a common way to speed up the access to flash by minimizing the number of cross boundaries calls between the RAM device and the underlying flash devices.
• Transactional - To support transactional access we need to extend the batch operations to fail, succeed as a single atomic operation.
• Query index support - to speed up the query and access time, the data needs to be indexed in a way that fetching a particular object by its IDcould point directly to its physical location on disk.

 

To make this a practical reality GigaSpaces partnered with SanDisk who through fusion I/O acquisition own the majority of the SSD market. As part of this partnership, we implemented a new version of our In Memory Data Grid, XAP MemoryXtend, which is now integrated with SanDisk ZetaScale interface. ZetaScale implements all of the above features as a general purpose flash disk interface. Through this integration we are now able to utilize the XAP RAM based storage for handling complex queries and analytics as well as transactional support. We can also leverage the XAP cluster support for integrating multiple flash devices that are mounted on various machines and make them work as one big transactional database ( as illustrated in the diagram below).

The XAP MemoryXtend cluster consists of a number of partitions each plugged into a local SSD drive.  Each partition can have at least one backup running on a separate machine to ensure  availability. The XAP client is a smart proxy that abstracts the underlying cluster and exposes all of the physical partitions through a single data-grid interface. The proxy routes the write or query requests to a particular partition in cases where we’re looking for a specific data item. In the case of agregated queries it will invoke a parallel query against all nodes and consolidate the results into a single result-set.

 

Flash DataBase as Service

As Flash devices are now being supported by cloud providers such as Amazon, it only make sense to leverage this capability and offer this sort of Flash based database on-demand.

This is where the XAP Deployment and Orechestration comes handy as it automates the process of deployment and management of XAP clusters across a variety of cloud infrastructures.

For more information, check out The next Big Disruption in Big Data.

The next big thing in big data: fast data

The big data movement was pretty much driven by the demand for scale in velocity, volume, and variety of data. Those three vectors led to the emergence of a new generation of distributed data management platforms such as Hadoop for batch processing and NoSQL databases for interactive data access. Both were inspired by their respective predecessors Google (Hadoop, BigTable) and Amazon (Dynamo DB).

As we move to fast data, there’s more emphasis on processing big data at speed. Getting speed without compromising on scale pushes the limits in the way most of the existing big data solutions work and drives new models and technologies for breaking the current speed boundaries. New advancements on the hardware infrastructure with new flash drives offer great potential for breaking the current speed limit which is bounded mostly by the performance of hard drive devices.

Why using existing RDBMS, NoSQL on top of flash drive is not enough

Many of the existing databases — including more modern solutions such NoSQL and NewSQL — were designed to utilize the standard hard drive devices (HDD). These databases were designed with the assumption that disk access is slow and therefore they use many algorithms such as Bloom filters to save access to disk in case the data doesn’t exist. Another common algorithm is to use asynchronous write to commit-log. A good insight into all the optimizations that are often involved in a single write process on NoSQL databases is provided through the Cassandra Architecture. Let’s have a look at what that entails below:

 

The Cassandra write path

 

What happens when disk speed is no longer the bottleneck?

When the speed of disk is no longer the bottleneck, as in the case of flash devices, then a lot of the optimization turns into overhead. In other words, with flash devices it will be faster and simpler to access the flash device directly for every write or read operation.

Flash is not just a fast disk

Most of the existing use cases use flash devices as faster disk. Using flash as a fast disk was a short path to bypass the disk performance overhead without the need to change much of the software and applications. Having said that, this approach inherits many unnecessary disk driver overhead. So to exploit the full speed potential of flash devices, it would be best to access the flash devices directly from the application and treat flash devices as a key/value store rather than a disk drive.

Why can’t we simply optimize the existing databases?

When we reach a point in which we need to change much of the existing assumptions and architecture of the existing databases to take advantage of new technology and devices such as flash, that’s a clear sign that local optimization isn’t going to cut it. This calls for new disruption.

The next big thing in big data

Given the background above, I believe that in the same way that demand for big data led to the birth of the current generation of data-management systems, the drive for fast data will also lead to new kinds of data management systems. Unlike the current set of databases, I believe that the next generation databases will be written natively to flash and will use direct access to flash rather than the regular disk-based access. In addition to that, those databases will include high performance event streaming capabilities as part of their core API to allow processing of the data as it comes in, and thus allowing real-time data processing.

 

Fast data in the cloud

Many of the existing databases weren’t designed to run in cloud as first-class citizens and quite often require fairly complex setup to run well in a cloud environment.

As cloud infrastructure matures, we now have more options to run big data workload on the cloud. The next generation databases need to be designed to run as a service from the get-go.

To avoid the latency that is often associated with such a setup, the next generation databases will need to run as close as possible to the application. Assuming that many of the applications will run in one cloud or another, it means that those databases will need to have built-in support for different cloud environments. In addition, they will need to leverage dynamic code shipping to pass code with the data, and in this way allow complex processing with minimum network hops.

In-memory databases and data grid are the closest candidates to drive next-gen flash databases

RAM and flash-based devices have much more in common than flash and hard drive. In both RAM and flash, access time is fairly low and wasn’t really the bottleneck; in-memory databases use direct access to the RAM device through key value interface to store and index data.

Those factors makes in-memory based databases more likely to fit into the next generation flash databases.

The combination of in-memory databases and data grid on top of flash devices also allows the system to overcome some of the key capacity and cost per GB limitations of an in-memory based solution. The integration of the two will allow an increase in the capacity per single node to the limit of the underlying flash device rather than to the limit of the size of RAM.

The overall architecture of the integrated solution looks as follows (Source: Gartner):

 

 

As can be seen in the diagram above, the IMC (in-memory computing) layer acts as the front end to the flash device and handles the transactional data access and stream processing, while the flash device is used as an extension of the RAM device from the application perspective.

There are basically two modes of integration between the RAM and flash device.

LRU Mode – In this mode, we use the RAM as a caching layer to the flash device. The RAM device holds the “Hot,” i.e. the most recently used, and the flash device holds the entire set.

Pros: Optimized for maximum capacity.

Cons: Limited query to simple key/value access

Fast Index Mode – In this mode, all the indices in RAM and the data in flash holds the entire set.

Pros: Supports complex querying, including range queries and aggregated functions.

Cons: Capacity is limited to the size of indexes that can be held in RAM.

In both cases, the access to flash is done directly using key/value interface and not through disk drive interface. As both RAM and flash drives are fairly fast, it is simpler to write data to the flash drive synchronously and in this way avoid potential complexity due to inconsistency.

It is also quite common to sync the data from the IMC and flash devices into an external data store that runs on a traditional hard drive. This process is done asynchronously and in batches to minimize the performance overhead.

Direct flash access in real numbers

To put some real numbers behind these statements, I wanted to refer to one of the recent benchmarks that was done using an in-memory data grid based on GigaSpaces XAP and direct flash access using a key/value software API that allows direct access to various flash devices. The benchmark was conducted on several devices as well as on private and public cloud-based services on AWS. The benchmark shows a ten times increase in managed data without compromising on performance.

 

 

 

For full details of the benchmark, see the following white paper:

XAP MemoryXtend -Massive Application Storage Capacity for Real-Time Applications

 

 

 

PaaS shouldn’t be built in Silos

Last year I wrote a post titled Platform as a Service: The Next Generation Application Server?  where I was referring to the shift from  application server into platforms.

On my 2011 predictions  I referred to a new emerging PaaS category targeted for those who be looking to build their own custom PaaS offering.

The demand coming from many SaaS providers to offer their own platform to their ecosystem players (a good example on that regard is Salesforce force.com) as well as the continues pressure of enterprises on their IT to enable faster deployment of new application and improve the utilization of their existing resources will force those organizations to build their own specialized PaaS offering. This will drive a new category of PaaS platform known as Cloud Enabled Application Platform (CEAP) specifically designed to handle multi-tenancy, scalability, and on-demand provisioning but unlike some of the public PaaS it needs to come with significantly higher degree of flexibility and control.

One of the immediate question that came up in various follow-up  discussion was  “how is that different from the Application Servers of 2000?”

I’ll try to answer that question in this post.

The difference between PaaS and Application Servers 

Lori MacVittie wrote an interesting post, "Is PaaS Just Outsourced Application Server Platforms?," which sparked an interesting set of response on twitter.

Lori summarized the aggregated response to this questions as follows:

PaaS and application server platforms are not the same thing, you technological luddite. PaaS is the cloud and scalable while application server platforms are just years of technology pretending to be more that it is..

Now let me try to expand on that:

The founding concept behind application servers was built around the concept of Three Tier architecture where you break your application into presentation, business-logic and data-tier. It was primarily an evolution of Client/Server architecture, which consists of a rich client connected to centralized database.

The main drive was the internet boom that allowed million of users to connect to the same database. Obviously having million of users connected directly to a database through a rich client couldn’t work. The three-tier approach addressed this challenge by breaking the presentation tier into two parts – a Web Browser and Web Server - and by separating the presentation business logic tier into two separate parts.

In this model, Clients are not connected directly into the database except through a remote server.

Why is PaaS any different?

A good analogy to explain the fundamental difference between existing applications servers and PaaS enablement platform is the comparison between Google Docs vs Microsoft office. Google came up with a strong collaboration experience that made sharing of documents simple. I was willing to sacrifice many of the other features that Office provided just because of the ease of sharing.

The same goes with PaaS – when I think of PaaS I think of built-in scaling, elasticity, efficiency (mostly through Multi Tenancy) and ease of deployment as the first priority and I’m willing to trade many other features just for that.

In other words, when we switched from desktop office to its Google/SaaS version we also changed our expectations from that same service. It wasn’t like we took the same thing that we had in our desktop and put it over the web – we expected something different,.  This brings me to the main topic behind this post:

PaaS shouldn’t be built in Silos

If I take the founding concept of Application Server – i.e. Tier based approach - and try to  stretch it into the PaaS world the things that pops up immediately in my mind is the impedance mismatch between the two concepts.

Tiers are Silos – PaaS is about sharing resources.

Tiers were built for fairly static environment – PaaS is elastic.

Tiers are complex to setup – In PaaS everything is completely automated. Or should be.

Now this is an important observation – even if the individual pieces in your Tier based platform is scalable and easy to deploy the thing that matters is how well they can:

1. Scale together as one piece,
2. Dance in an elastic environment as one piece.
3. Run on the same shared environment without interfering with one another... as one piece through multi-tenancy.

Silos are everywhere ..

In the application server world, every tier in our infrastructure is built as a complete silo. By "silo," I mean that they are not just built separately, but that each component comes with its own deployment setup, scaling solution, high availability solution, performance tuning and the list goes on...

All that individual deployment and tuning in the context of a single application, well, if you stretch that into multi-tenant application, it becomes much worse.

Silos are not just the components that make our products. Silos are also a feature of those who build the components.

If you think of Oracle Fusion as an example or IBM for that matter you’ll see silos. Messaging, data, analytics, all broken into a separate development group with different management, individual development groups.

Each component has a different set of goals, different timelines, different egos, and in most cases they are even incentivized to keep those silos in place simply since they are not measured on how well their products work with other pieces but solely on how well their thing does its (siloed) job which leads into even greater focus on silos - functionality that works as an end to itself and not a contributor to general application development.

Oracle Fusion is an interesting attempt to break those silos. In the course of creating Fusion, they even came up with an interesting announcements of bringing Hardware and Software together.

Is "bringing hardware and software together" good enough? It's certainly an attempt but if you look closely you’ll see that most of it is a thin glue that tries to put together pieces that were bought separately through acquisitions and are now packaged together and labeled as one thing.

However most of it is still pretty much the same set of siloed products. (Oracle RAC cluster is still very different than Weblogic cluster, etc.) That doesn’t apply to Oracle only – think of VMware and its recent acquisition now labeled under VFabric.

Now imagine how you can scale together an erlang based messaging system with other pieces of their stack as one piece.. The sad thing about VMware is that it looks like they went and built a new platform targeted specifically for PaaS world but inherited the same pitfalls from the old Tier-based application server world.

Silos are also involved in the process by which we measure new technology.

Time and time again I see where even those who need to pick or build a PaaS stack go through the process of scoring different pieces in the puzzle independently, with different people measuring each piece rather than looking on how much those pieces work together.

A good analogy is comparing Apple with Microsoft: Microsoft would have probably score better on feature by feature comparison (they can put a checkmark in more places) but the thing that brings many people to Apple is how well all of its pieces work together.

Apple even went farther and refused to put in features that might have been great - just because they didn’t fit the overall user experience. They often win with users because of the scarcity of features, even though it may not look as nice on a bullet list of features.

Sharing is everything.

The thing that makes collaboration so easy in the case of Google Docs is the fact that we all share the same application and underlying infrastructure. No more moving parts. No more incompatible parts.

All previous attempts to synchronize two separate documents through versioning highlighting and many other synchronization tools pale compared to the experience that we get with Google Docs. The reason is that sharing and collaboration in Office was an afterthought. If we work in silos, we’ll end up… in silos. Synchronizing two silos to make them look as one is doomed to fail, even if we use things like revision control systems for our documents.

We learned through the Google docs analogy that sharing of the same application and its underlying infrastructure was a critical enabler to make collaboration simple.

There is no reason that that this lesson wouldn’t apply to our Application Platform as well. Imagine how simple our lives could be if our entire stack could use one common clustering mechanism for managing its scaling and high availability. All of a sudden the amount of moving parts in our application would go down significantly, along with lots of complexity and synchronisation overhead.

We can measure the benefits of having a shared cluster based on the fact that it comes with less moving parts compared with those that are based on Tier based model:

• Agility – fewer moving parts means that we can produce more features more quickly without worrying if once piece will break another piece.
• Efficiency – fewer moving parts means that will have less synchronization overhead and network calls associated with the each user transaction.
• Reliability – fewer moving parts means lower chance for partial failure and more predictability in the system behavior.
• Scaling –  fewer moving parts means that scaling is going to be significantly simpler, because scaling would be a factor of a "part" designed for that purpose.
• Cost – fewer moving parts means fewer products, a smaller footprint, less hardware, reduced maintenance cost, even less brainpower dedicated to the application.

First Generation PaaS – Shared platform

If we look at the first generation PaaS platforms such as Heroku, Engine Yard, etc., we see that they did what Oracle Fusion does. They provide a shared platform and carve out a lot of the deployment complexity associated with the deployment of those multi-tiered applications. (This makes is marginally better than Oracle Fusion, but not good enough... certainly not as good as it could or should be.)

I refer to these as “PaaS wrappers”. The approach that most of those early-stage PaaS providers took was to “wrap” the existing tier-based frameworks into a nice and simple package without changing the underlying infrastructure, to provide an extremely simple user experience.

This model works well if you are in the business of creating lots of simple and small applications but it fails to fit the needs of the more “real” applications that need scalability, elasticity and so forth. The main reason it was built this way was that it had to rely on an underlying infrastructure that was designed with the static tier-based approach in mind. An interesting testimonial on some of the current PaaS limitations can be found in Carlos Ble's post, "Goodbye Google App Engine (GAE)."

Toward 2nd Generation PaaS

In 2010, a new class of middleware infrastructure emerged, aimed at solving the impedance mismatch at the infrastructure layer. They were built primarily for a new, elastic world, with NoSQL as the new elastic data-tier, Hadoop for Big data analytics, and the emergence of a new category (DevOps) aimed at simplifying and automating the deployment and provisioning experience.

With all this development, building a PaaS platform in 2011 will - or should - look quite different than those that were built in 2009.

The second-generation PaaS systems would be those who could address real application load and provide full elasticity within the context of a single application.

They need to be designed for continuous deployment from the get-go and allow shorter cycles between development and production.

They should allow changes to the data model and introduction of new features without exposing any downtime.

The second-generation PaaS needs to include built-in support for multi-tenancy but in a way that will enable us to control the various degrees of tradeoffs between Sharing (utilization and cost) and Isolation (Security).

The second-generation PaaS should come with built-in and open DevOps tools that will enable us to gain full automation but without losing control, open enough for customization and allowing us to choose our own OSes and hardware of choice as well as allowing us to install our own set of tools and services, not limited to the services that come out of the box with the platform.

It should also come with fine-grained monitoring that will enable us to track down, trouble-shoot  our system behavior through a single point of access as if the platform was a single machine.

The 2nd generation PaaS systems are not going to rely on virtualization to deliver elasticity and scaling but instead provide a fine grain level control to enable more efficient scaling.

At Structure 2010, Maritz said that “clouds at the infrastructure layer are the new hardware.” The unit of cloud scaling today is the virtual server. When you go to Amazon’s EC2 you buy capacity by the virtual server instance hour. This will change in the next phase of the evolution of cloud computing. We are already starting to see the early signs of this transformation with Google App Engine, which has automatic scaling built in, and Heroku with its notion of dynos and workers as the units of scalability.

Final words

Unlike many of the existing Platforms, in this second-generation phase, its not going to be enough to package and bundle different individual middleware services and products (Web Containers, Messaging, Data, Monitoring, Automation and Control, Provisioning) and brand them under the same name to make them look as one. (Fusion? Fabric? A rose is a rose by any other name - and in this case, it's not a rose.)

The second-generation PaaS needs to come with a holistic approach that couples all those things together and provide a complete holistic experience. By that I mean that if I add a machine into cluster, I need to see that as an increase in capacity on my entire application stack, the monitoring system needs to discover that new machine and start monitoring it without any configuration setup, the load-balancer need to add it to its pool and so forth.

Our challenge as technologists would be to move from our current siloed comfort zone. That applies not just to the way we design our application architecture but to the way we build our development teams, and the way we evaluate new technologies. Those who are going to be successful are those who are going to design and measure how well all their technology pieces work together before anything else, and who look at a solution without reverence for past designs.

References

• Is PaaS Just Outsourced Application Server Platforms?
• PaaS choices today – ZDnet
• Platform as a Service: The Next Generation Application Server?
• VMware Knows the Cloud Doesn’t Need Server Virtualization
• Extreme and extensible applications Gigaspaces XAP andVMware vFabric GemFire –ZDnet
• Goodbye Google App Engine (GAE) 

Written by Nati Shalom and Joseph Ottinger

Google App Engine plus Amazon AWS: Best of both worlds

George Lawton wrote a a good summary of my JavaOne talk in his article titled Google App Engine plus Amazon AWS: Best of both worlds 

Google App Engine (GAE) is focused on making development easy, but limits your options. Amazon Web Services is focused on making development flexible, but complicates the development process. Real enterprise applications require both of these paradigms to achieve success… What we really want is the flexibility and performance of AWS and the simplicity and ease of use of GAE.

This is exactly what we had been working on for the past year, leading us to the launch of our new cloud platform. With this platform we leverage GigaSpaces XAP as the high performance scale-out application server and Amazon as the robust and flexible IaaS. Together they form an alternative Platform as a Service geared for enterprise grade applications. This allows the cloud environment to inherit the extreme performance, latency and scalability of the XAP platform, which in turn enables achieving your performance and scaling target with less machines, implying a lower cost.

Real-life case study: Primatics financial – Risk analysis as a service

Francis de la Cruz and Argyn Kuketayev from Primatics Financial joined me through the presentation. In their part of the session they described their experience in developing a SaaS application for Real Time analytics.

Kuketayev described how Primatics used this approach to create a new automatically scaling cloud version of an existing banking application. Primatics initially developed a mortgage securities application that allows banks to estimate the value of a basket of hundreds of thousands of loans. The value of these loans fluctuates as economic conditions change and some portion of home owners cannot afford to make payments on their loans. Banks normally only need to assess the value of these loans at the end of each month, making them an ideal candidate for cloud services like AWS.

From a scalability perspective the challenge is to be able to provide a highly multi-tenant application that need to serve many firms, many users in that same firm each running many jobs at the same time. Implementing such a model can be fairly complex as you will need to be able to manage the life cycle of each job and each user independently and in isolation from one another.

 

Trying to build such a service directly on Amazon is going to be fairly complex, as you can learn from George’s summary below:

Primatics wrote the first version of EVOLV:Risk as a hosted web application for a regional bank.. The application needed to be fault tolerant so that if one node crashed, they did not have to restart the application over again from the beginning. Kuketayev said that it is not just about the loss of four hours, but the office is trying to close out the month and needs to access data to end the monthly cycle so they can go home.

Using GigaSpaces' toolset they rewrote the entire application infrastructure in about four-months to run on top of AWS. Now they can kick off as many instances as required for different banking customers, and each instance runs significantly faster than before. Kuketayev said that it is important for banks that none of their applications run on the same infrastructure as another bank.

The diagram below shows the specific architecture that Primatics ended up using. Those that are familiar with Space Based Architecture would find it fairly straight forward:

The application is built out of  a set of processing units. Each processing unit contains the compute agents in the form of a polling-container.  The compute agents gets a a reference to a remote Data Grid that is shared by all processing units. Each agent gets the job injected to it by the polling container and gets a reference to the data it required to process the job. Once the job is completed, the result is stored back in the space. The results are flashed out back to a database through a mirror service.

In a case of a failure, other compute agents are able to continue from the exact point of failure and continue the job processing as if nothing happened. This is because the state of the job is kept safe in the data-grid and not in the agent’s memory.

Kuketayev from Primatics nicely summarized thye lesson he learned after going through the experience of trying to build it on his own vs. trying to use GigaSpaces:

Kuketayev said that one of the biggest lessons is that you need to have your infrastructure do the provisioning for you automatically, or otherwise you end up spending a lot of time just turning things on and off. He said they are now using configuration APIs to automate this process, whereas before they were using scripts. This allow for automatically throttling and failover recovery without human intervention.

Kuketayev advised "You need to make sure you use the right tools … You don't want to have to worry about provisioning and reliability. Make sure you have provisioning, failover, monitoring and SLA out of the box."

The full JavaOne presentation is available here:

Alternative to Google App Engine on Amazon

View more OpenOffice presentations from Nati Shalom.

Final words

Fr solution providers the size of Primatics, building a risk analysis application as a service couldn’t be possible without cloud computing. Cloud enabled them to offer their solution as a service without the need to go through major investment of building a data center to support it.

Primatics’ experience is not special. One of the benefits of building Software as a Service is that you have one shared environment for all your customers. At the same time, one of the challenges is that in a shared environment, failure becomes more public and will impact ALL your clients. If the system doesn’t scale well, you’re going to be hit twice as hard as in a standalone application.

Building a robust and scalable SaaS application can be fairly complex. A good cloud infrastructure will get you a first class data center, but it won’t solve your application requirements.What’s interesting with cloud computing is that it forces you to think about the cost and efficiency of your application more than ever before. In the Primatics example, running a simulation of 100 nodes for 3 hours is very likely to fail at some point. A failure during such a simulation will immediately cost you 300 hours, not to mention the fact that you might lose the simulation window for the day and the reputation challenge you’ll will be facing with your customers. In addition, putting the data in-memory and making the application run 3-5 times faster means that you would need 1/5 of the machine power, which saves 80% of the cost of running the application.

I believe that the challenges imposed by cloud computing force us to focus on what we do best and avoid investing in areas which are not core to our business. Because the pay-per-use model significantly lowers the cost barrier, going down the path of writing your own infrastructure, as many have tried to do before, will be much more expensive and risky then ever before.

References:

• Google App Engine plus Amazon AWS: Best of both worlds 
• The Need for Speed - Saving Costs By Improving Application Efficiency
• eBay Subsidiary Uses XAP to Boost Performance - Expects to Meet Same Loads with 95% Less Servers
• The impact of cloud computing on build vs. buy behavior

GigaSpaces Launches a New Version of its Cloud Computing Framework

We have been working for a while on our new Cloud Computing Framework (CCF). We wanted to wait until we had some real customers and partners behind the platform in production before spreading the word on a larger scale. And now, I am very happy to report that we have reached this point sooner than expected and just in time for the JavaOne event.

The new version includes an enhanced integration with the Amazon billing system Devpay designed to enable users to use GigaSpaces as a service on a true pay per use model. All you need is an Amazon account and your done. Unlike other products, that are offered in similar fashion, we decided not to provide just a bundle of our software packed in an Amazon image (AMI) but instead provide a more tight integration with EC2 that will enable our users to use a single GigaSpaces product for all GigaSpaces versions as well as customize the installation to include their own software packages and application code on the fly. The new CCF provides a true end to end experience that will enable our users to deploy and manage their *entire* application from the load-balancer to the database in one single click.  The latest release of CCF includes support for the new GigaSpaces 7.0 release. Users that are looking to try out the new XAP product can now do that for free without the need to download or install anything.  This will enable you to try out any of the built-in examples, test the high availability features and run benchmarks in just one click. The new version includes lots of additional improvements to the user experience as well as new features and bug fixes. To view a detailed list of the new features click here.

Those that will be visiting JavaOne this week will be able to join the hands-on labs session and experience a full deployment of JEE application using CCF. You will also be able to gain insight from one of our customers, Primatics Financial on how they used CCF to build their Risk Management application as a service – See  full details here

In this post I will try to address some of the most common questions that we often receive when introducing this new service:

Who is using GigaSpaces CCF and Why?

The new release is already used by different customers in the following areas:

ISV’s and Service providers using CCF for SaaS enablement:

Many ISV's are looking for ways to move their standalone applications into a SaaS based offering. In doing so they face a few challenges:

1. Scalability – now their application needs to serve not just the users of a certain customer at a certain location but all customers from all locations. This increases the scalability challenge on their product by an order of magnitude.

2. Continuous high availability – one of the challenges that is imposed by having a shared platform for all customers is that the impact of failure can be catastrophic as failure can affect all of your clients at the same time and becomes public fairly quickly.

3. Portability – There are cases where some customers would still require their application be installed in their local data center for security and latency purposes. Today, this often means you will need to develop two seperate branches of your product to serve these two scenarios. With GigaSpaces you can write your application once and deploy it in the cloud or in your customer data center without having to modify your application and without locking your application to a specific cloud provider.

4. Performance – In developing high performance applications such as Trading, Risk Analysis, or SIP server, keeping the latency at sub msec and throughput at a maximum becomes critical. This is where our XAP middleware has a proven record in running numerious production mission critical applications.

The following are a few public references that chose to use XAP:

- Primatics Financial is using GigaSpaces CCF to build a high performance risk management solution

- Orbyte is using GigaSpaces CCF for building state of the art high performance trading applications that can be offered as service.

Other customers include Online Gaming companies that were looking for a simple and cost effective manner to launch new online games and provide a gaming application platform that enables other gaming providers to host their applications easily in a cloud environment. See further details on our online gaming solution here.

Enterprise customers using CCF to improve time to market of new applications and for large scale testing

According to recent research conducted by Forrester, one of the main requirements driving cloud adoption by the enterprise market is that current IT can’t keep up with the business demands of their organization:

• Capacity planning is too difficult – The current method of provisioning applications based on estimated peak loads has become either too costly or impossible. Most systems are non linearly scalable which means that knowing how many resources would be needed to meet certain goal is known only through trial and error.
• Time to market – launching new applications and services often takes months and sometimes year.
• The business wants a simple and cost effective way to prototype – not all applications make it to production but these initiatives still require the business to invest upfront in the purchase and configuraiton of infrastructure to support these prototypes.
• Cost saving – There are sets of applications that have fluctuating demand on resources.  Enabling these applications to “lease” a common set of resources only when under heavy demand can lead to huge cost saving and efficiency.

I discussed the enterprises challenges in my previous talks. The most recent one is the one I gave during the CloudSlam event:  Practical Guide to Developing Enterprise Applications on the Cloud- Online Presentation 

A good example of an enterprise customer who is already using our CCF in production is a Large Telco Service Provider that searched for ways to reduce the time to market for launching new scalable web applications. This customer was able to bring a project that previously took more than 6 months just to finish the development and testing to 4 weeks with only 90$ spent on pre-investment.

Large scale testing, performance benchmarks and demo as a service

Users that are not ready to use the cloud to run their production systems can still receive benefits from the cloud for testing and demonstration purposes. There are a few challenges that we tried to address with CCF to make this type of use simpler:

1. Making cloud testing environments identical to your local testing environment – To effectively test your application on the cloud and deploy a production version of your system locally, your application must be at a point where it behaves fairly the same in both environments.

2. Setting up a large scale environment of hundreds of nodes can be fairly complex

3. Trouble shooting failure in such an environment can be a fairly complex and lengthy process.

If your application is already running on our XAP product then porting it to the cloud and running it locally in your production environment will not require any change to your code or configuration. With CCF setting up a large scale environment of hundreds of nodes with the new CCF takes just a few minutes. We designed the CCF to run not only GigaSpaces but other external services which are essential for tests such as JMmeter, Tomcat servers etc. Users can easily configure their own initialization and command scripts when the machine is launched and define which software packages will be installed on their machines on the fly. We added the ability to capture the logs of multiple machines easily and monitor the system and middleware behavior at runtime using a zero installation environment all through the web.

GigaSpaces and our existing customers and partners have already been using CCF to run live demos and recently GigaSpaces started using CCF for our own internal large scale testing purposes. Using the cloud for demo purposes has been extremely useful to show cases that previously were hard to demonstrate such as fail-over and scalability in a real production environment. You can read more on how you can use CCF for running your own demo on the cloud in a similar way without going through all the challenges we had gone through in one of my recent posts: Demo as a Service . Similarly CCF can be ideal for running large scale Benchmarks.

Partner ecosystem -

We designed our CCF in away that would enable us to run not just GigaSpaces components but almost any application such as JEE or Spring based applications. This enables our Solution provider partners to use CCF to offer their specialized solutions and offerings for both GigaSpaces and their other software partners.

System Integrators are also looking for ways to gain a competitive advantage by offering turn key solutions at a significantly shorter time to market and lower initial investment. I tried to outline below some of the public references on these two categories:

• Solution providers

Real time production monitoring:  Dynatrace offers enhanced integration with CCF for enhanced production monitoring.  Dynatrace users are able to monitor in real time how their application is performing and trace their application bottlenecks in real time – see details on this integration here.

Model Driven Development framework - New Technology/enterprise (NT/e) - GigaSystemBuilder helps Java developers prove applications quickly - in a few days - on a local grid or in the cloud, and then rapidly move on to production development, rather than waste time on finger-trouble or implementation details – see more information here as well as a live demo on their deployment manager solution

• System integrator

In many cases customers are looking for experts that will enable them to gain the benefit of the cloud and bridge some of the skill-set gaps. AAR is one of our leading system integrators in that field who has already delivered several solutions on top of our GigaSpaces CCF for Enterprise customers. For them, using CCF was a tool to gain a competitive edge and provided their customers the ability to launch new applications in a matter of weeks as opposed to months and at a fraction of the setup (development environment, testing environment, staging environment) and hardware costs that are associated with setting up such an environment with traditional projects.

Here is a quote from one of our System Integrators in the UK that describes their experience in using CCF for the Telco Service provider that I mentioned earlier.

“We use GigaSpaces XAP and CCF to deploy a standard JEE web application on EC2. To address security concerns we kept the business logic outside of the cloud on the customer site. We exposed the business logic through a secured web services channel. We use XAP data grid for maintaining the session high availability and for reducing the latency associated with accessing the remote data center. We used the XAP integrated Jetty container for hosting our web application and through that gained the built-in self healing and auto-scaling provided through the CCF. In this way we were able to deal with potential failure or load without human intervention. From a management and monitoring perspective we were able to leverage the integrated ganglia monitoring and integrated visual representation of different metrics, like cpu/memory/network usage of the entire cluster. We used JMX to bind our custom Mbeans to the integrated ganglia UI. Overall GigaSpaces CCF gave us big boost on utilizing ec2”

I must admit that from our end this was probably the smoothest project we had been engaged with. The reason was probably related to the fact that unlike other projects we had full control over the environment. This enabled us to minimize potential miss configuration errors which is one of the major areas for complexity in setting up such a project. Another benefit was the support. When we faced an issue it was very easy to log-in directly to the machine and fix the problem on the spot without the need to go through a complex process of authorization from the IT Security and without the need to create a reproduction package and ship it over.

The customer was able to deploy a new business application quickly without putting all the initial investment upfront. In this way they gained the freedom to decide at any point if the application should continue, be canned or if they should move it entirely to their local IT.

Looking into the future

The future looks very promising as we continue to work on bringing new customers and businesses towards using CCF and the latest XAP 7.0. We are currently working on supporting the Amazon European datacenter and taking advantage of some of the new XAP 7.0 cluster administration API to enable users to provide thier own custom SLA deployment. At this point, I would like to ask for your specific feedback and wish list for our next major CCF release.

GigaSpaces based solution makes it to the finalist of Cisco Developer Contest

I was very pleased to read an email from Leonardo, who was the winner of the OpenSpaces Developer Challenge (a worldwide programming contest using the Gigaspaces application server which was held last year), saying that he is now a finalist in the Cisco developer contest. Here's a bit about him and the application he submitted:

About Leonardo

Leonardo worked for several ISPs in various roles as network administrator and java programmer for IT consulting firms, and finally as software architect in high-performance Java EE based projects. He is passionate about parallel programming, distributed computing and more recently semantic web and its applications on software engineering.

Leonardo was the winner of the OpenSpaces Developer Challenge. He enjoys reading about various technologies in the field of computer science. When he is not developing code, he prefers to spend time with family and friends, walk in the park, or watch a movie.

About the application

Resource Management Platform is a proposal to develop an event based platform that leverages AXP, Services Gateway Initiative (OSGI), Jini and JavaSpaces technologies to enable deployment of IP Multimedia Subsystem (IMS) applications based on Session Initiation Protocol (SIP); more specifically, the Call Section Control Function (CSCF) components. It will have admission control mechanisms to manage Call processing.

This solution improves infrastructure manageability for large scale IMS applications. Such a platform will potentially be useful to enable deployment of high-performance, network-based SaaS (Software as a Service) or Cloud Computing solutions at the network edge by leveraging AXP.

You can find the full details about his project here.

Leonardo's project is interesting, because it shows how you can use Space Based Architecture (SBA) for implementing a scalable Telco application and offer it as SaaS application on the cloud.

Interestingly enough, I got another email the week before from Amin Abbaspour, who presented another case study illustrating how you can build a scalable SMS service using SBA, as shown in this diagram:

 

What the two projects have in common, from an architecture perspective, is that they both represent a highly scalable Event Driven design. The unique thing about Event Driven applications is that they require a combination of messaging, data and service interaction that needs to be tightly orchestrated to meet high performance/low-latency requirements without compromising on consistency, ordering (FIFO) and reliability. This combination of requirements represent one of the hardest challenges in building scalable architectures. Trying to meet this type of challenge in the traditional way by integrating messaging system for event delivery , database or simple caching (like Memcached or TC) for data and a traditional application server for business logic is going to lead to fairly complex architecture. Trying to reach linear scalability and keeping the latency low with so many moving parts is close to impossible. This is what makes SBA such a good fit. The main difference about SBA is that it recognizes there is strong dependency between messaging, data and business logic. The key is to have one shared clustering, high availability and scalability for all three components of the architecture. This makes it possible to reduce the number of moving parts and network hops associated with each business transaction, thereby increasing reliability.

On a personal level, I was very pleased to see that the software we are developing is helping people like Leonardo and Amin to build their own carrier and put themselves in a unique spot in highly competitive market.

Good luck Leonardo and Amin!

References

• Making EDA programming simple with JeeWiz
• Cisco AXP Developer contest
• What's new in GigaSpaces XAP 7.0
• GigaSpaces on the cloud
• OpenSpaces developer portal

Practical Guide to Developing Enterprise Applications on the Cloud- Online Presentation

According to a recent survey, available skill sets is one of the leading decision factors for organizations in determining which application platform to use, while scalability and availability are next. This reveals one of the main obstacles for bringing enterprise applications to the cloud: How do you take something as disruptive as cloud computing, and bring it to an enterprise environment, without forcing a complete re-write?

In my recent talk at the CloudSlam (online) Conference, I tried to summarize the challenges people face when trying to deploy enterprise applications on the cloud. Try to imagine a scenario where you have an existing JEE system, and your system is under load. At this point you would like to add more machines to accommodate that load. What will happen to your application in this case? Will your application be able to take advantage of the additional capacity? How do you know how many machines should be added to meet the load in the first place?

The answers I usually hear to these questions are fairly consistent. If you’re lucky, you’ll only require configuration changes to be at a point where you can utilize the extra resources. Knowing how many machines to add in order to meet a certain load is yet another challenge. Going through the normal capacity planning process could take weeks with the way systems are currently running.

At the same time, if you’re going to deploy our application in a static manner (reserved instances, static IP, …), it would probably make it simpler to deploy your existing application on the cloud, but it probably won’t make much economical sense. So the question I was trying to answer is how we get from the static application deployment most of us are using today, to a point where we can get end-to-end scaling and elasticity of our application from the load-balancer to the database without going through a complete re-write.

In this presentation, I tried to suggest a set of solutions to overcome the various challenges I mentioned earlier and how to apply them in a gradual manner to avoid huge initial investments and high risk. There were also some interesting questions toward the end, centered specifically around one of the hardest problems – getting the data pieces sorted out in a such a dynamic environment. Those who missed this presentation can now view it online:

Note: The voice quality isn’t that great, plus the desktop sharing on Webex didn’t give a clear indication when my desktop was actually shared, so the beginning of the talk is missing the first two or three slides. If you have questions or want to get a copy of the presentation, just shoot me an email.

During the upcoming JavaOne conference there will a hands-on lab session by Daniel Templeton from Sun Microsystems (PetClinic in the Clouds: Scaling a Classic Enterprise Application) where users can go through the steps and deploy a Pet Clinic application on the cloud. 

In addition to that lab, I'm going to have a Technical Session (Alternative to Google Application Engine for Java™ Technology-Based Applications).

For those that are not going to join the JavaOne event and want to get some hands-on experience on how the steps that i outlined in the presentation can be implemented in real application, I’d recommend looking Pet Clinic demo that is available on Amazon cloud using our cloud framework.

I’m happy to say that the number of customers that are already using this approach through our Enterprise Cloud middleware platform is growing quickly and were seeing more and more application that would traditionally considered as not "cloud compatible", being deployed on the cloud. Below are few of the public references that were mentioned just recently on Jim Liddle's blog:

• Telco application running on Amazon EC2 and GigaSpaces
• Using GigaSpaces to Trade in the EC2 Cloud

Designing a Scalable Twitter

Guy Nirpaz, Uri Cohen and Shay Banon came up with an interesting exercise as part of the recent partner training that took place at the GigaSpaces office. In this exercise, the students were asked to come up with a scalable design for Twitter, using Space-Based Architecture.

There are some interesting scalability lessons from this exercise, which are applicable to anyone looking to implement new-style real-time web applications such as the ones used for social networking.

In this post I'll  try to summarize the main patterns to put into place and considerations to make when designing such a scalable architecture.

Background:

For those of you who are not yet familiar with the service, Twitter is sort of a SMS-service meets discussion board.  You can post short messages (up to 140 characters) that can be shared with a group of subscribers that are referred to as "followers". The main difference between twitter and other messaging applications is that both SMS and Instant Messaging (IM) applications were designed primarily for one-on-one communications whereis Twitter was designed primarily for broadcast communications (publish/subscribe, or pub/sub). Another aspect that is special about Twitter is that by default anyone can follow anyone else. In other words, it was designed for open communications, not private, as were IM and SMS.

What are Twitter's scalability challenges?

1. Sending a tweet (a message on Twitter is known as a 'tweet') -– The challenge is how to handle an ever-growing volume of tweets and re-tweets and responses that can lead to a viral "message storm"

2. Reading tweets – The challenge is how to handle a large number of concurrent users that continually “listen” for tweets from users (or topics) they follow.

Designing A Scalable Twitter

Choosing the right scalability patterns

Almost every challenge in software architecture has its roots in one of the existing patterns. So the simplest course is to start by looking for those patterns, and choosing the right patterns to scale the application. Looking at many other scalable architectures, we'll begin with a partitioning pattern as the core design principle. By partitioning our Twitter-like application we'll spread the load across a cluster of servers and scale by simply adding more servers (i.e., partitions).  Another important architectural observation about Twitter is that it doesn’t fit into the classic database-centric design that most web applications do. On the flip side, it doesn’t fit well with a messaging-centric design (pub/sub) either. It is a combination of the two.

A pattern that is suitable for this type of collaborative messaging is known as a blackboard pattern.  In our design, we will use those two design patterns -- partitioning and blackboard -- as the foundation for our scalable Twitter application. With the foundation in place, let’s list the requirements and examine how these patterns can be used to scale the app.

Scalability Requirements

We'll assume a relatively extreme scaling requirement:

• Tweet Volume: 10 billion tweets per day
• Tweet Storage: 100 Gigabytes per day (with 10:1 compression)

Additional assumptions:

• Tweets are limited to 140 characters
• Tweets are immutable, i.e., there are no updates, only inserts
• Twitter limits client applications to 70 requests per hour

Now that we have the foundational patterns and clear requirements, we can design the architecture. We'll start first with the blackboard system.

Using an In-Memory Data Grid (IMDG) as a Blackboard System

The are several approaches to building a blackboard system. To maximize performance and scalability, we'll store the data in memory, thus avoiding disk I/O, which is often the main cause for contention. For years, Java has provided a model for designing blackboard systems known as JavaSpaces. More recently, distributed caching has become popular and can provide similar capabilities to those of JavaSpaces. Let's examine two popular distributed caching approaches for our blackboard system:

1. Simple read-mostly caching using memcached
2. Read/write caching, also known as an In-Memory Data Grid (IMDG)

Choosing between memcached and an IMDG

Memcached enables us to to store the data (tweets) in a distributed memory set and read it in a scalable fashion. Having said that, be aware that memcached is not transactionally-safe and is not designed for reliability (i.e., it doesn’t support fail-over and high availability). That means that if we use memcached or something similar, we will have to use a database as the back-end. Every tweet posted will have to be written to both memcached and the database in a synchronous fashion to ensure that no tweet will be lost. This approach may be good enough for scaling read access, however, for writes and updates it offers limited scalability.

Unlike memcached, which was designed for simple read-mostly caching, In-Memory Data Grids  are designed for handling a read/write scenario, and can therefore act as the system-of-record for both write and read operations. We can still use a database for long-term persistence, but because the IMDG maintains its reliability purely in memory, we can write and update the database asynchronously and avoid hitting the database bottleneck.

Todd Hoff, author of highscalability.com wrote an interesting summary that covers the different products in this space in a recent post:  Are Cloud Based Memory Architectures the Next Big Thing?

Todd provide a clear explanation of how an IMDG works (using GigaSpaces):

 

• A POJO (Plain Old Java Object) is written through a proxy using a hash-based data routing mechanism to be stored in a partition on a Processing Unit. Attributes of the object are used as a key. This is straightforward hash based partitioning like you would use with memcached.
• You are operating through GigaSpace's framework/container so they can automatically handle things like messaging, sending change events, replication, failover, master-worker pattern, map-reduce, transactions, parallel processing, parallel query processing, and write-behind to databases.
• Scaling is accomplished by dividing your objects into more partitions and assigning the partitions to Processing Unit instances which run on nodes-- a scale-out strategy. Objects are kept in RAM and the objects contain both state and behavior. A Service Grid component supports the dynamic creation and termination of Processing Units.

Back to our Twitter app: Given the scalability requirements, we will need to scale both reads and writes, and therefore, an IMDG is a more suitable approach to implementing the blackboard system.

Now let’s examine how the use of an IMDG as the blackboard system enables us to scale both sending and reading tweets. Let's start by designing the partitioned cluster.

Designing a partition architecture

One of the main considerations in designing a partition cluster of any kind is determining the partition key, such as a Customer ID in a CRM application or a Trade ID in a trading application. At first glance, it sounds like a trivial decision, but choosing the right partitioning key requires a deep understanding of the application usage patterns and data model.  In the case of Twitter, we could choose to partition the application by the data-type, the user, the tweet itself or the followers. Our first goal is selecting a key that will that will be granular enough to enable scaling the application just by adding more partitions, while making sure that we don't end up with a key that is too fine-grained -- making it sub-optimal for querying purposes.

If we use the timestamp key, for example, our application will be optimized for “inserts” (writes), however, even a simple query such “retrieve the tweets of a certain user” will force us to execute an aggregated query against all partitions. Alternatively, if we partition the data based on user-id, we'll be able to easily spread the load from different users across partitions. Retrieving the tweets of a certain user is going to be resolved in one call to a single partition. We may encounter a problem if a single user generates a significant higher load than average, however, in the case of Twitter, we can assume that this is not very likely. Partitioning by user-id is a good compromise.

Data capacity analysis

With such extreme requirements it is clear that storing all tweets in memory is going to require huge memory capacity. Very quickly this will become economically prohibitive, so we need to devise a scheme in which the IMDG acts as a buffer for most of the load on the system, and then offloads the data and queries to an underlying persistent storage.  In our Twitter example, it is fair to assume that most real-time queries (those that require fast access to the data) will be resolved in data from the last hour or 24 hours. Queries that require older data will need to hit the database for the initial call. However, subsequent access to fetch new updates should be resolved purely in-memory.

Using this approach, we'll need about 10 servers, each holding 10GB of data in memory to accommodate 24 hours of activity. If we also want to back up the data in memory, we will need double the amount of servers.

Choosing the right eviction policy

It's reasonable to assume that recent data is accessed most and older data is rarely used. To ensure that we get the maximum hit ratio on our memory front-end, let's choose a time-based eviction policy, which always holds the most recent updates in memory. When we will reach our memory capacity limit the oldest data will automatically get evicted from memory. The actual window of time in which we will be able to keep in memory is obviously dependent on the size of the cluster. With an IMDG implementation all tweets are stored in a persistent storage, which means that when tweets are evicted they are not deleted from the system.

Scaling tweet writes:

If we select user-id as the partitioning key, each user tweet will be sent to a specific partition. Multiple users may be routed to the same partition. Usually the algorithm to determine which partition fits a certain user is something like:

routing-key.hashCode() % #of partitions

In GigaSpaces, this is done by marking the routing attribute of our tweet class with an @SpaceRouting annotation.

The web front-end application will call space.write( new Tweet(..),..)  to send the tweets. This way there is nothing in our web client code that exposes the fact that the underlying implementation interacts with a cluster of partitions (spaces in GigaSpaces). Those details are abstracted within the space proxy. When the write method is called on the space proxy it parses the field that matches @SpaceRouting from our Tweet() object and uses this field value to calculate the partition it belongs to. It then uses that value to route the Tweet(..) object to the appropriate partition.

With this approach, the web application can be written in a very simple way and can interact with the entire cluster as if it was a single server.

The data from the memory partitions gets stored asynchronously into a persistent storage. The persistent storage could be a database, but it could also be other things, such as an index search engine based on Compass/Lucene.

Scaling tweet reads:

To those familiar with messaging system, at first glance Twitter looks like a classic publish subscribe application. A closer look, however, reveals that any attempt to implement Twitter with something like a JMS message queue is going to fail in achieving a scalable system. This is especially true if you consider that the system needs to maintain a durable queue for each user. That could easily lead to a scenario in which each tweet is published to thousands of subscribers and every re-tweet can potentially lead to a "message storm".

As I discuss above, the right way to think about this type of application is as a blackboard pattern, just as a blackboard (or these days, a whiteboard) is used by a group of people (followers, in the case of Twitter) to share information and collaborate. When someone writes something on the board, everyone sees it and can choose to react. Unlike messaging (take email for example), we don’t need to send separate messages to each subscriber. Instead everyone is looking at the same board. Everything is also copied from the board to paper. When the board runs out of space, we erase it. And we can always page through the paper copy to access the board history. 

In Twitter, this means that each follower that follows a group of people is basically polling for messages posted by those users from the last time he read them. To make things more tangible we can express this type of query with the following SQL syntax:

SELECT * FROM Post WHERE UserID=<id> AND PostedOn > <from date>.

The <from date> will normally be the last few minutes, if we're constantly looking for new messages.

But there's a caveat. Remember that we partitioned the application by user-id? This means that each user's tweets are stored in a separate partition. How can we read all users' posts? If we poll for each user individually, we will end up with a lot of network calls. The simplest approach would be to execute one call that looks for ALL the users we're following and look for updates (new tweets) from those users. The pattern we'll use to perform such this task is mapreduce. One way to do that with GigaSpaces is through the distributed task API:

The distributed task API is a modern version of the stored procedure. The following snippet shows what such a call would look like:

AsyncFuture<Long> future = gigaSpace.execute(new GetTweetsUpdates());
long result = future.get(); // result will be the number of primary spaces

The GetTweetsUpdates() class contains code that will be injected in each partition and will enable us to look for updates from the users we follow in a single call. Because the call runs in-process, and because the data is stored in-memory, executing such a task is extremely fast compared with the equivalent with database and stored procedure operations. Execution is aggregated to the caller implicitly. The caller can use a reducer to aggregate the results into a single result object.

Scaling the web front-end

Nothing really new here. We'll use a classic web front-end, which is comprised of a load-balancer and a cluster of web servers that act as a front end to our IMDG instances. The web application will use a single cluster-aware IMDG proxy to send new tweet posts. The IMDG proxy will be responsible for mapping the tweet with the actual partition that hosting the tweet. That logic is kept completely out of the application code. This allows us to keep our web tier clean and simple.

Keeping the web layer stateless to avoid session stickiness

One common pattern for keeping the web tier scalable is to use a Shared-Nothing Architecture, which basically means that the web tier will be stateless. This requires keeping the user session state external to the web-tier. As previously demonstrated, the IMDG can be used as high-performance, scalable data store for maintaining shared session state information. This allows us to avoid session stickiness and to scale the web tier without being locked in to a specific server throughout the entire session, in case the server is over-loaded.

For more information on how to scale the web tier, as well as other important capabilities such as self-healing and auto-scaling, see the following tutorial: Scaling Your Web Application.

Making it simple and cost-effective using cloud computing

Twitter is yet another example for a situation in which system load is highly variable and the difference between average load and peak load can be quite significant. In such cases, provisioning our system can be fairly hard and costly. This is where cloud computing and SLA-driven deployments can help us scale on demand and pay only for what we use.

Once we figured out a way to partition the application, it's going to be much simpler to package the application into self-sufficient units (referred to in GigaSpaces as processing-units) and scale the application simply by adding or removing these units on demand. You can learn more about this here

Final words

Scaling a real-time web application such as Twitter or Facebook introduces unique challenges that are are quite different from those of a "classic" database-centric application. The most profound difference is the fact that unlike with traditional sites, Twitter is a heavy read/write application, and not read-mostly. This seemingly minor difference can break most existing models for web application scalability. Using a combination of memcached + MySQL is not going to cut it for this type of application. 

The good news is that with the right patterns and set of tools, building a scalable architecture that meets such challenges isn’t that difficult.  There are already plenty of success stories that demonstrate that, such as the following example from highscalability.com: Handle 1 Billion Events Per Day Using a Memory Grid

The proposed architecture is by no means perfect and can be further optimized to meet even better performance and latency, but that will come at the cost of simplicity. I believe that the proposed architecture should get you pretty far as-is. Avoid going through more advanced optimizations until the point they are an absolute must.

References

• Twitter getting bigger - Scaling becomes more of an issue

• Memory-Based Architectures and the Cloud

• Handle 1 Billion Events Per Day Using a Memory Grid

• Are Cloud-Based Memory Architectures the Next Big Thing?

• Scaling Your Web Application

• Persistence on the Cloud – How and Why?

• Scaling twitter presentation by Guy Nirpaz

Getting ready for the cloud

In the past few months, I have been speaking in various conferences about cloud computing.
In my presentations, I tried to focus primarily on how one can take practical steps to benefit from the cloud today. To illustrate my points, I used GigaSpaces as the scale-out application server and Amazon EC2 as the cloud infrastructure and thus demonstrated how one can deploy EXISTING java applications on the cloud while:

• Not having to re-write your application
• Preventing lock-in to specific cloud provider
• Enabling seamless portability between your local environment to cloud environment 
  • No code or configuration change is required between the two environments
  • Develop local - test on the cloud
  • Built for iterative development

In addition, I demonstrated the use of our new Cloud tools framework which enabled me to fully automate the entire application provisioning process on the cloud. This tool enabled me to:

• Deploy load balancer
• Cluster web containers
• Cluster application processing units
• MySQL database connected to EBS

And all that through a single click deployment.

In the demonstration, I also showed how you can benefit from the cloud:

• Dynamically scale when the load breaches a certain threshold
• New web containers are dynamically linked to the load-balancer (no configuration change is required as all is taken care of on the fly).
• Self healing - if something breaks, scale-down to existing machine while getting a new machine ready.

The entire presentation can be viewed online 

GigaSpaces - Getting Ready For The Cloud

View SlideShare presentation or Upload your own. (tags: amazon gigaspaces)

You can also view the online video presentation that i gave in the Cloud Summit in Tel Aviv where i ran a live demo of our new cloud infrastructure on EC2 demonstrating both dynamic scaling and also what happens when machine fails in front of a live audience!

The video recording of this presentation is available here

If you want to try to run the demo yourself and you don't have an Amazon user - just drop an email to cloud at gigaspaces dot com and will send you a free pass to the early access program.


Enjoy!

Reducing latency with Sun Real Time JVM

Frederic Pariente  Engineering Manager at Sun Microsystems posted an interesting summary of a case study with GigaSpaces on Sun blog: Gigaspaces curbs latency outliers with Java Real Time

In the context of a customer proof-of-concept this summer and in the light of the 2.0 release of Java Real Time System --JRTS 1.0 had the bad prerequisite of source code changes--, Gigaspaces revisited the opportunity for Java Real Time to serve the low-latency requirements of trading applications. Gigaspaces XAP 6.5, Solaris 10 and both Java 5.0 standard and real-time JVMs were used for the benchmark. The test scenario included a trade matching engine and multiple clients injecting messages at extreme speed. The success criteria was to get guaranteed latency per message under 10 msec, with no code modification to the matching engine"

"..The first lesson learned was that msec latency was achievable with the standard JVM, through some advanced tuning of the JVM command-line options. While the customer had reported application freezes up to 20 sec during garbage collection under heavy load --he was running the JVM with no particular flags, unfortunately default JVM options optimize for throughput--, latencies could be brought down to milliseconds by switching to the Concurrent Garbarge Collector"

"..The second lesson learned was that the number of outliers can be reduced by an order of magnitude by using the real-time JVM. At a small cost in terms of application throughput --lower-- and CPU usage --higher-- of course"

You can see the full detailed benchmark and JVM option in the original post.

How Real Time JVM works?

For those who are not faimiliare with Real Time JVM, Fredric points to a very detailed presentation on his post Java Real Time for latency-critical banking applications  which I'd recommend looking at. I took one slide from the presentation which i found useful for understanding the general concept behind Real Time VM.

As could be seen in the above diagram the RT-JVM introduces new type of threads "Critical RT threads". It makes sure that GC will not run while those thread are running and in that case provide better predictable behavior.

Other references:

Latency is Everywhere and it Costs You Sales - How to Crush it - My Take