Nati Shalom's Blog

Xtreme Transaction Processing under SBA terminology

In previous post I reviewed Pat Helland analysis on transaction processing in distributed scalable environment as outlined in his document "Life beyond Distributed Transactions: an Apostate’s Opinion" We touched about the implicit contradiction between scalability and distributed transactions and suggested a new model which IMO becomes the third generation of TP systems.
Massimo Pezzini from Gartner named that third generation TP eXtreme Transaction Processing or XTP. Both referred to this model as an emerging trend. Beyond those two analysis I'm aware of other research in this area claiming pretty much the same thing.

In this post I'll focus on how GigaSpaces XAP and SBA implements the principles as was laid out in the previous post:

• Create/rearrange your object model using entities (objects with IDs), in a such a way that most of your state changes will happen inside the boundaries of a single entity. a change to a single entity is atomic.

In an SBA  model this special case of objects is referred to as  Tuples. The Tuple model is very different from a Tree model that other caching systems supports since a tuple doesn't hold any distributed reference to other objects. Holding references to other objects would end up with tight coupling between those objects. This tight coupling will eventually determine the locking/transaction boundaries and therefore the scaling granularity.

• For those changes that exceed the boundaries of one entity, changes will not be atomic and the final desired change will be achieved using a sequence of atomic changes, every change request (activity) MUST be implemented to be idempotent because of the chance of retransmitting due to failures.

 

In an SBA  world this is achieved through Master/Worker pattern which if you would like is a distributed command pattern. In this model transactions are executed through a chain of commands - the submitter is responsible for triggering the first command which start a chain of events (workflow) between a set of workers each responsible for s specific step in the transaction flow(validation, matching, routing..), a failure in one of those steps lead to a rollback of the transaction associated with that step. At this stage another worker will be able to pick it up and continue the processing of the failed transaction from the same point it failed. Unlike the current transaction model the submitter only blocks till the first command is committed, the rest happens in the background. The later is probably the key difference between today's model in which the submitter blocks till the entire transaction with all its various resources is completed. Another difference with XTP is that the submitter remains loosely coupled from the other transaction participants. This is another key aspect that makes SBA  much more aligned with SOA/EDA and Grid type of architecture.

Executing transaction that span between multiple entities/tuples:

In real life situations there are situations where a single transaction need to reference multiple data items under the same transaction scope. This could potentially lead to huge overhead. To minimize that overhead we often partition the data based on affinity in away that each partition will become a self sufficient unit. The affinity key is the common key that is used to associate between a set of entities/tuples. A typlical example is csutomer-id, protfolio etc. In this case transaction scope could span across multiple data items but wouldn’t need to exceed a partition scope.

• To accomplish idem-potency of your activities (actions/methods) the entity itself SHOULD hold a history of committed activities per source(entity).

This steps assumes that there is no hot backup that can continue the processing of a transaction from the exact point where it failed. It is therefore assumed that the way to overcome this limitation is re-issue the transaction in case of a failure. This can obviously lead to scenario where the same transaction or part of that transaction could be processed more then once hence why idempotency handling is required. In SBA architecture we assume that there is a hot (in-memory) backup which can continue the processing of a transaction from the exact point it failed. In this way we minimize the cases in which you would need to deal with idem-potency since we will deal with it implicitly. Obviously this is can be achieved as long as the entire state associated with a transaction resides within the space.

The forth rule:

So far we discussed the common principles which lead to XTP as Pat Helland and Massimo Pezzini layed out. At this stage I would like to introduce a forth principle: Minimize the amount of moving parts associated with a business transaction:

In the current Tier Based Approach we build our system from different set of building blocks that are separated into tiers such as messaging-tier, business-tier, data-tier etc. Often the implementation of those tiers is done in cylos i.e. each tier maintain its own clustering, high availability model and often even data model. While I would agree that it make sense to keep those different components logically separated it is also pretty evident that XTP implies strong runtime dependency between those units. For example - the messaging system need to be aware of how we partition the data in order that transactions will be routed to the appropriate partition, from a latency perspective any network hop is going to effect the latency with direct proportion to the amount of hops. When we apply high availability at each tier separately we by definition add redundant network hops and disk I/O since each system maintain its own high availability either through replication or through persisting the data into the disk. In SBA approach we address the contradiction between the loose coupling required at the application Service level and tight coupling at the runtime level through a concept that has been widely used in distributed technology - virtualization. In our specific scenario we separate between the API and the runtime implementation of that API. We create a common runtime model which address the clustering, high availability, load balancing, fail-over semantics across all tiers and then create API façade that expose the concrete set of functionality i.e. messaging/data-sharing/parallel-processing to the application level on top of that common runtime. In this way we can achieve minimum latency since we eliminate the cylos and redundant network hop associated with them, In addition to that we maintain the workflow associated with a business transaction in-memory under a context of a single partition and therefore reduce all the network overhead associated with it. High availability of the entire application is maintain without writing to disk through replication etc.

Another method in which we can reduce the amount of moving parts with SBA  is through a new component that is refer to as processing-unit. A processing-unit is built as an application container (In our case it extends the Spring container). It contains all the individual component/services that have tight runtime(latency, fail-over, scaling) dependency. As such a processing-unit defines the unit of scale and failover of our application.  In SBA  the transaction is scoped within the boundaries of a processing-unit. The fact that all the components involved in a transaction are collocated eliminate the chances for partial failure and therefore the need for handling distributed transactions.

Summary:

Through the set of two last posts I tried to provide a summary of the rational behind the third generation Transaction Processing (TP model) referred to as XTP - eXtreme Transaction Processing . I later tried to outline based on Pat Helland analysis (Amazon), and Massimo Pazini (Gartner) the core principles behind XTP. In this post I described how Space Based Architecture SBA  follows and extend those principles in a more concrete aspects.

At this stage I would like to summarize what is see as the common principles for achieving XTP :

• Rule #1 - Build your data model from a set of independent data items (Tuples)
• Rule #2 - Partition your different data items based on your data affinity (Customer-id, account-id,..)
• Rule #3 - Execute transaction as a set of individual steps scoped within a single partition.
• Rule #4 – Minimize the amount of moving parts involved under the context of a transaction
  

A full working example that illustrate how this pattern works using Spring and OpenSpaces abstraction is provided here.

This example illustrates a typical scenario of an Order Management System (OMS) where orders submitted to the system becomes a command that triggers a sequence of POJO Driven Bean each dealing with a specific state of that transaction (Parsing/Validation -> Execution/Matching -> Routing).

Julian Browne has previously posted success story of a real production Order Management Systems based on this model.

In The third part of this series of posts I'll try to suggest a simple migration path between the current J2EE transaction model and XTP.

Lessons from Pat Helland: Life Beyond Distributed Transactions

Distributed Transactions are a common pattern used to ensure ACID properties are met between distributed resources. Since the late 70s, the fIrst generation of this model has been widely used in many large-scale applications that struggle with the difficulties of multiplexing many online terminals. This led to the emergence of the 1st generation TP Monitors (TPMs), such as Tuxedo (Now owned by BEA). The emergence of web based applications in the late 90s drove the creation of 2nd generation TPMs, in the form of JEE application servers, to address similar needs using more open and modern technologies as described in the following article by Subbu Allamaraju: Nuts and Bolts of Transaction Processing. The diagram below illustrates a typical transaction flow in a JEE environment:

       

Transactions in EJB Application Server
(Source: Subbu Allamaraju, Nuts and Bolts of Transaction Processing)

The increased business demand for greater scalability led many to the realization that the current transaction model is a bottleneck due to its inherit centralized approach. It also makes our systems quite brittle and complex due to the tight coupling that it introduces, as described in a paper by Pat Helland of Amazon.com: Life beyond Distributed Transactions: an Apostate’s Opinion:

Today, we see new design pressures foisted onto programmers that simply want to solve business problems. Their realities are taking them into a world of almost-infinite scaling and forcing them into design problems largely unrelated to the real business at hand. Unfortunately, programmers striving to solve business goals like eCommerce, supply-chain-management, financial, and health-care applications increasingly need to think about scaling without distributed transactions. They do this because attempts to use distributed transactions are too fragile and perform poorly.

This challenge leads us to the emergence of a third generation of TPMs,  or what Gartner calls Extreme Transaction Processing (XTP).

What approaches are people taking today to overcome the limitations of previous generations? From Pat:

"..Because the performance costs and fragility make distributed transaction impractical. Natural selection kicks in,... applications are built using different techniques which do not provide the same transactional guarantees but still meet the needs of their businesses"

So the question is how to achieve high transactional throughput and performance without compromising consistency and reliability?

Pat Helland suggest the following principles:

• Instead of global transactional serializability we assume multiple disjoint scopes of transactional serializability
• Scalable Apps Use Uniquely Identified “Entities” [Nati: Entities represent data elements that are equivalent to Tuples in Space-Based Architecture terminology]
• Atomic Transactions Cannot Span Entities
• From the programmer’s perspective, the uniquely identified entity is the scope of serializability
• Messages Are Addressed to Entities [Nati: in other words, we use messages to manage the workflow between the disjointed transactions - this is pretty much the equivalent of a transaction coordinator].
• Entities Manage Per-Partner State (“Activities”) [Nati: in GigaSpaces terminology: each activity is executed within a collocated partition].

In order to avoid distributed transactions you must:

• Create/rearrange your object model using entities (objects with IDs), in a such a way that most of your state changes will happen inside the boundaries of a single entity. a change to a single entity is atomic.
• For those changes that exceed the boundaries of one entity, changes will not be atomic and the final desired change will be achieved using a sequence of atomic changes, every change request (activity) MUST be implemented to be idempotent because of the chance of retransmitting due to failures.
• To accomplish idem-potency of your activities (actions/methods) the entity itself SHOULD hold a history of committed activities per source(entity).

Combining these three principles allows you to safely change the state of the application without distributed transactions, while allowing for almost infinite scaling.

How does this apply to SOA? 

A core principle in SOA, event-driven architecture (EDA) and grid computing is the notion of loose coupling of services. The distributed transaction model leads to tight coupling by definition, as it requires intimate integration and runtime dependencies among all of the services that are associated with a transaction. The suggested model of "disjoint transactions" fits in nicely with this type of new architecture. Here's Pat again:

Everything discussed in this paper is supportive of SOA. Most SOA implementations embrace independent transaction scopes across services. The major enhancement to SOA presented here is the notion that each service may confront almost-infinite scaling within itself and some observations about what that means. These observations apply across services in a SOA and within those individual services where they are designed to independently scale.

Pat ends with:

In a few years we are likely to see the development of new middleware or platforms which provide automated management of these applications and eliminate the scaling challenges for applications developed within a stylized programming paradigm. This is strongly parallel to the emergence of TP-monitors in the 1970s.

Guess what, Pat, the future is now!

In a the next post i'll discuss how Space Based Architecture (SBA) and GigaSpaces XAP as the implementation of SBA follow similar principles to those described by Pat Helland in a very elegant way that fits perfectly with SOA, EDA and Grid.

Additional References:

• Shay Banon on Scalability Mythbusters
• InfoQ discussion on New patterns and middleware architecture needed for true linear scalability?

 

Julian Browne on Space Based Architecture

Julian Browne, formerly the head of architecture at Virgin Mobile, wrote a series of three posts about his experience with and views on Space Based Architecture:  

• Space Based Architecture Illustration  - In this post Julian provides a general introduction to SBA, and describes how at Virgin Mobile they re-designed a centralized J2EE Web-based Order Management System (OMS) into a scalable SOA using Space-Based Architecture: "we could make ordering a reliable and predictable process, and we could reuse our new OMS for other channels (why stop at the web? Once it works and works well, why not re-use it for telesales and high street stores too?)"
  
  
• Space Based Agility looks at how SBA supports Agile because of their complementary nature to how business people think.
  
  
• Space-based Archetypes discusses some of the subtleties of SBA as it relates to non-functional requirements such as Availability, Capacity, Concurrency, Enhance/Extend-ability, Interoperability, Latency, Maintain/Manage-ability & Monitoring, Recoverability, Reliability & Resilience, Security and Scalability.

Here are a  few my favorite highlights from Julian's view and definition of Space Based Architecture. Julian's take is quite interesting since it goes beyond the technical definition to a level that can be understood by almost any human being (including my wife:) )

* It's easy to talk to stakeholders about what a space is. Everyone gets the idea of a blackboard. There are hundreds of corporate examples of shared-data, that is accessed by more than one person/role, you can refer to. And having special spaces for certain activities is akin to a departmental separation of concerns.

* Entries in spaces are also easy to discuss without boring them to death with code walk-throughs and technical details. In the non-software world, passing a 'thing' (most commonly a form) between people and maintaining that as the basis of communication is entirely logical. Often forms contain instructions for the reader on how to interpret or use the form's data (just think of your credit card statement. Wouldn't it be weird if payment instructions were kept somewhere else, and you had to go and look them up?)

* Workers make sense too, not only because they represent real 'workers', but because as sponsors change their minds (if there's one key attribute of Agile, it's that it transparently reacts to changing requirements better than waterfall) the extent of change to workers and entries is linear - big change to model, big change to workers/data.

* Because the entries are code and data, they are, like my heated floor, self-contained capsules of all that's needed to perform particular steps of a business operation. Easy to isolate and change because they are naturally loosely coupled.

Julian on when should one consider using SBA:

..my summary of all this is that if you have a simple application, with predictable growth and reasonably contained business logic a tired approach will do fine. If it's a database-driven web site that's not going to get heavy throughput, or generating lots of business transactions, then one of the MVC frameworks will probably cover it. But once you get into distributing that logical architecture across nodes, and the words scalability and transactional safety start to become important to the business, you really do need to look into better ways to break away from some of the heaviness of JEE, you'll probably have to do a bit more work to get some operational aspects to the same level as for the big application servers, but it's more than likely to be worth it in the long run.

Beyond the fact that it is an excellent piece, it is an indication of something that I felt happening to everyone exposed to the SBA concept . Once people get it, they feel this approach makes so much more sense than the traditional approaches they've been exposed to.

People that were burnt in the past by N-Tier architecture and J2EE are most likely to be enthusiastic advocates of SBA, as I've witnessed in our recent customer conference. I'm hoping that Julian is only the first one to speak about it publicly.

Avoiding vendor lock-in in a non-standard environment

Vendor lock-in is a common concern for many customers when they need to choose the software platform for their application. One way to address this concern is to agree on a standard API/Interface. However, as we have all seen on many occasions, that model has many drawbacks. I'll name a few of them:

1. Standards usually lag behind market needs
2. The market requires a certain maturity, which can sometime takes years to get to the point where standard can even be discussed
3. Standards tend to "kill" innovation, as we've witnessed with J2EE, where everyone wanted to stick to the least common denominator (not necessarily the vendors, but the customers)
4. Standards need standards bodies, which tend to be highly politicized and consequently do a poor job of representing the best interests of the end user.

As a result, we need to accept the fact that most new technologies will be non-standard (at least up to a certain degree). That means that  we need to manage the risk of lock-in without relying on standard APIs to do the job for us. Interestingly enough, this question is less common with open source frameworks. A good example is Hibernate and the Spring Framework. Both frameworks were very rapidly adopted although they were not open standards. Hibernate later evolved to become a standard (JPA). Many ideas from Spring made their way through to EJB3 in the same way. What can we learn from this?

We can't rely entirely on standards to manage our lock-in risks. Open source can reduce the risk from a different angle: the fact that it's open and used by others through an open community makes it very close to a standard effort, except that it's done by a specific group that we "choose" to rely on to do a good job representing the interests of the larger community. Open source projects that will do a good job of that, will most likely become a de-facto standard.

Abstractions - the new "standards"
Abstractions are a well-known approach for separation between our application and the underlying implementation. Having said that, achieving the right level of abstraction, without scarifying functionality, is an art form. Spring taught us several lessons on how to do abstractions right through the introduction of the dependency injection and declarative approach. Using this model we can handle the logic of how to get an event and which method to call when an event arrives outside our code through a declarative set of configurations, and keep our code completely independent of the underlying event delivery implementation, whether it is JMS, JavaSpaces, a Data Grid and so on. One of the nice things about abstraction is that it is loosely coupled to the API, as it happens outside of the code, i.e., as part of our configuration. This means that our code can work with different messaging systems, even if they don't implement the same API. Mule is a very good example of how this can be achieved with different and many messaging protocols.

In summary:

1. You can't entirely avoid lock-in risk; you can only manage it
2. You can reduce and control lock-in risk in the following ways:
   • Through abstraction
   • Using open source

The ideal solution is to use open source as the abstraction framework -- and then you are free to use sophisticated commercial platforms underneath, without lock-in. That's what GigaSpaces' OpenSpaces framework is all about.

You can read more about it here:

• Ignorance Is Bliss: Abstractions Vs. Standards
• Wouldn't your SBA approach lock me into GigaSpaces? -

Scalabilty patterns - InfoQ Interview

Johan Strandler from InfoQ did a good job of describing the architecture implications of scaling stateful applications in his article New patterns and middleware architecture needed for true linear scalability?

Is there a need for new patterns and middleware-architectures when building linearly scalable applications? Nati Shalom, CTO at GigaSpaces, believes that existing middleware designed for a tier-based approach cannot be used for a true linearly scalable architecture. Instead he proposes a new middleware stack based on self-sufficient processing units that supports a partitioned/scale-out model.  While Shalom proposes a new middleware stack, Pat Helland at Microsoft, some years ago proposed some new transactional patterns and formalizations to be used in what he calls almost-infinite scalable systems.

This is an encouraging sign that awareness around what's needed to achieve true scalability is now center stage.

Testability – the orphan child in distributed systems – Part II

In Part I of this post-series I discussed the importance of testability in distributed systems and outlined a set of key principles for addressing the issue.

The truth is that testability is often constrained by the products that we're using. In this post I wanted to outline how we at GigaSpaces address these issue with OpenSpaces, our new Spring-based development infrastructure:

Start with the architecture

There are a few components in OpenSpaces and Space-Based Architecture that significantly simplify the distributed architecture complexity:

Processing-Unit: unifies the deployment, scaling and fail-over behavior semantics of both the middleware and business logic components.

In-Memory Middleware: OpenSpaces takes advantage of the fact that space-based middleware can address both messaging and data using the same underlying technology. That in itself reduces a large part of the complexity, because we reduce the amount of moving parts in our architecture.

The fact that "spaces" can be collocated in-memory reduces the runtime footprint and provides a simple way for running the entire application in a single process environment.

Spring Abstraction – The Spring Framework was primarily designed to address the complexity of J2EE applications. OpenSpaces utilizes and extends many of the principles introduced by Spring, such as dependency injection and separation between the business logic and the underlying implementation through abstraction, remoting, etc. This simplifies the architecture and configuration of an OpenSpaces-based distributed application.

Measure scalability and fail-over as part of the unit-test environment

Because OpenSpaces is a relatively lightweight framework, it is easy to start an entire cluster within a single Eclipse session. This allows to very easily measure the behavior of the application in fail-over and scaling scenarios.

Enable fast iterative development 

The concept of the Processing Unit allows developing all of the business logic in a single VM and running it locally.  Moving it to a cluster, and scaling it, only requires to run that *same* processing-unit multiple times. This makes the production and unit-testing environments pretty much one and the same, and therefore, enables testing 90% of the functionality in a single environment. OpenSpaces helps you to take advantage of that and enable the developer to deploy the application from his or her testing environment to the production system by a click of a button, knowing that it's going to behave the same as in the local environment.

Keep the configuration consistent

In this respect, OpenSpaces provides a standalone container that is tuned to run within an Eclipse environment. Using this approach, a user changes only the container used between the development and production testing environment, while the rest of the code and configuration remains the same.

Testability – the orphan child in distributed systems

Designers of distributed systems are forced to deal with many complex issues such as high availability, scalability, security, performance and latency.  We often address these issues with brute force development work, thereby creating a complex development process. As a result, testability is often ignored or at least it is considered a low priority issue.

Testing complexity grows exponentially when we add high availability and scalability to the picture simply because it is very hard to set up an environment in which we can test our code and apply our changes. Consequently, high availability and scalability tests are often performed at a late stage in the development cycle. Not surprisingly, we often find that scalability and high-availability has a significant effect on our application behavior, especially when it comes to performance and latency. Because this issues is only discovered at a late stage of the development process, projects get delayed. In many cases, we need to get back to the whiteboard and redesign our application based on the new findings and the cycle starts all over again.   

This cumbersome process directly affects application quality, stability and reliability. I've witnessed too many cases in which a simple bug brought an entire system down and led to a loss thousands of dollars. The fact that many transactional systems today are online systems makes those failures extremely visible, sometimes showing up in the 6 o'clock news.

Testability should be equally important to the other "ilities" in a new design of a distributed application.

How can we improve the testability of distributed applications?

I've listed below a set of key principles based on our experience at GigaSpaces in internal development and when working on customer implementations:

Start with the architecture:

If the architecture is complex, everything else will be complex -- including testing, scaling and fail-over.

Measure scalability and fail-over as part of the early test environment

Our goal should be to test our application behavior under fail-over and scaling events as early in the development cycle as possible. One way to achieve this is by running a small-scale of my application in a single machine. By small scale I simply mean small volumes of data and concurrent users. All other aspects of my application (clustering, high-availability) should be the same as what I plan for my deployment environment.

Enable fast iterative development

The cycle of applying changes to our code and testing it in our production environment should be very short. As we had witnessed with J2EE, if that cycle becomes too long we simply start to avoid it, and as mentioned earlier, that results in problems being identified late in the game, resulting in project delays. 

Keep the configuration consistent

Developers and architects often tend to think that if they only change the configuration and keep the code the same between the testing and production environments all will be well. As we all know, changing the configuration affects the way applications behave. It is, therefore, quite common to hear: "oh this was a configuration issue.." after we several days were wasted on trying to figure out what went wrong.

In Part II of this topic I'll discuss how we're following these principles and addressing this issue with the OpenSpaces framework we released to GA recently as part of GigaSpaces XAP 6.0 (download here).