Rajith’s Column

Here are 5 reasons why I found distributed system are hard to program. This is not some sort of thorough analysis, but merely my observations in dealing with such systems. For completeness, here is the definition of “Distributed System” I used.
A distributed system contains of more than one process that runs as a single system. These processes can be on the same computer or multiple computers that are on a local area network or geographically distributed over a wide area network.

Without any further do here are the reasons in no particular order.

1. Difficulty in identifying and dealing with failures.
When communicating between processes failures can happen at many levels. Dealing with them is not trivial. Of course you rely on frameworks based on technologies like RMI, CORBA, COM, SOAP, AMQP, REST(is an architectural style not a standard) etc to handle these. But the fact remains that you still need to clearly think about these cases and handle these situations properly.

For example if we consider a simple interaction between two processes on different computers, the following failures can happen.

• Failures that occur within the process that initiates the communication (sending the message or invoking the RPC call).
• Failures between the time the process hands over the request to the OS and the OS writing it to the network.
• Network failures between the time it takes to transmit the packets from one computer to the other.
• Failures between the time the OS on the receiving end receives the packets and then handing it over to the recipient process.
• Failures that occur when the recipient process tried to process the request/message.

Sometimes the framework you use, is unable to/may not report all these error cases. Sometimes when the error is reported, it may not contain enough information to figure out at which level the error occurred.
Did it reach the remote computer? if so how far up the stack did it go?. If the receiving process got the request or message did the error occur before or after the request/message was processed?
In some cases where idempotency is built into the the receiving application or the framework/protocol (ex a message client that detects duplicate messages, or doing an HTTP GET) a simple retry maybe ok. In some cases Idempotency and retrying maybe expensive or difficult to implement. In such cases careful thought needs to be given on how these different errors are identified and handled.

2. Achieving consistency in data across processes.
One of the hardest problems in programming distributed systems is achieving a consistent view of data across the processes. When one processes updates some data, you need to replicate them across the other processes, so if any other process decides to operate on the same set of data, then it is doing so on the most current copy.
Lets look at two examples.

Assume a global banking application for ABC bank. A customer goes to a branch in New York, US and deposits money to an account. A few moments later his relative in London, UK does a withdraw on that account. Due to latency there is obviously a time lag before the process in London, UK sees the updated amount in the account.

In an online trading system, a user in NY places an item for sale. The transaction is updated on the closest data center which is in Boston. A few moments later another user in LA is searching for the exact same item and is served off a data center in Phoenix. The user in LA may or may not see the item due to the latency involved in replicating the data across

For example 1 strong consistency is required, while for example 2, you could get away with weak consistency, for example by setting an SLA that says data is valid within a 5 min time window.
This is not an easy problem to solve and this area itself is a subject on its own. Wener Vogels wrote a nice peice on this called Eventually Consistent which is worth reading.
Of course there are specialized frameworks/libraries that can handle this for you. But still there is no escape for you and you pretty much need to have an understanding of the pros and cons of various approaches, failure modes etc.

3. Heterogeneous nature of the components involved in the system.
A distributed system may contain components written in a variety of languages deployed across machines with different architectures and operating systems. Needless to say that this poses certain challenges (especially integration, interoperability issues) when implementing the system. A whole range of standards/technologies were presented to solve these issues, including but not limited to CORBA, SOAP, AMQP, REST (is an architectural style not a standard) and RPC based frameworks like ICE, Thrift, Etch etc. Anyone who has worked with these technologies knows that neither of these are trivial to use nor provide a complete solution in every situation.

If anybody has read the recent posts by Steve Vinoski and the discussions around it would realize the issues/challenges surrounding RPC. The following paper discuss the impedance mismatch problems when working with IDL based systems. The issues with type systems and data formats are not limited to RPC only. When using a message oriented approach like SOAP (doc lit style) or AMQP you will end up tunneling data thats not supported by the protocol as a string or a sequence of bytes. When using REST you would need to represent your resource in a format the requesting application understands/supports, which maybe quite different from the native format.

Again not an easy issue to deal with no matter what technology or framework is used. As an architect/developer you need to understand these issues and deal with them accordingly.

4. Testing a distributed system is quite difficult.
This is arguably one of the hardest aspects of developing a distributed system. Verification of the behavior and impact of your code in the system is not easy.
There are many aspects that needs to be tested, and doing so before every checkin is not a fun task at all. Running some of these tests before every checkin is not practical. But its a good idea to run them nightly and some tests during the weekend. Here are some of the areas that needs to be tested (I plan to write another blog entry elaborating on the testing aspects).

• Functionality testing (can be covered with well written unit testing)
• Integration testing - you need to test the distributed system as a whole with all the components involved
• Interoperability testing - this is crucial when heterogeneous components (different languages, OS) are involved, and is quite different from integration testing
• TCK compliance - If your system is based on standards/specifications, then you need to ensure that you haven’t broken anything w.r.t compliance
• Performance testing - to ensure that your changes haven’t accidentally caused a degradation in performance
• Stress testing - to ensure that your checkin hasn’t accidentally caused any stability issues - ex increased chance of deadlocks when the load increases
• Soak testing - to ensure that your checkin hasn’t caused any longevity issues - ex a memory leak thats manifested after a couple hours, days

Most often than not developers cut corners in their testing as running these tests are tedious and time consuming. Also these tests need to be run regularly to catch issues in a timely manner and the best way to tackle this issue is to automate as much testing as possible. There many options with continuous build systems like cruisecontrol or using a plain old cron job.
Functionality testing, TCK compliance, certain types of integration and interoperability tests can be run periodically.
In most organizations test machines are just lying around doing nothing during the night (unless around the clock testing is done with development centers in different time zones.). Instead of wasting computing cycles, you could automate test suites to run during the night. More time consuming integration and interoperability tests, performance, stress and soak testing can be done nightly, while more longer duration soak testing can be scheduled to run during the weekends.

While testing is a tough issue for any type of system, distributed systems have a lot more failure points which adds to the complexity.
Getting these tests right to cover these failure points and executing them needs a lot of careful thought and planning.

5. The technologies involved in distributed systems are not easy to understand .
Distributed system are not easy to understand. Neither are the myriad of technologies used in developing these systems.
Most folks find it difficult to grasp the concepts behind these technologies. If you look into the discussions and misconceptions surrounding REST you can understand what I am trying to get at. CORBA was not an easy spec to understand, so is WS-* or AMQP. While it is true that you don’t need to understand everything to develop using them, you still need at least a reasonable understanding to figure how to tackle some of the above mentioned issues. Frameworks based on these technologies are touted as the cure for these problems. Sure they could help, but it still does not shift the burden away from you.
To compound the issue all sorts of vendors keep touting their technology/framework as the next silver bullet. No matter what vendor you use, at the end of the day you are still responsible for getting it right. And it is not an easy task. You need to face the reality that distributed systems are hard and that you cannot hide every complexity behind some framework.

My friend Matt brought to my attention an article on why MapReduce is a step backward, written by David J. DeWitt and Michael Stonebraker. I also stumbled on this blog entry written by Mark CC from Google. It is sad to see that most people advocate a particular solution to every problem imaginable and then refuse to look at anything else from a neutral pov. I let you draw your own conclusions.

AMQP defines four standard exchange types (routing algorithms) that covers most of the common messaging use cases. All AMQP brokers are required to support each of these exchange types and pre declare an instance of it identified by a standard name. The idea is to provide a simple out-of-the-box solution to most users. Users are free to create more instances of these exchange types with their own names. Also as mentioned in the previous post, users can create different exchange types and instances of them.

It is important to note that with any exchange type, a message can be matched with more than one queue if two or more queues are bound with the same routing criteria.

Direct Exchange
The exchange does a direct match between the routing key provided in the message and the routing criteria used when a queue is bound to this exchange.

(Click on image)

The most common use case is to bind the queue to the exchange using the queue name. However it is important to note that you could use any value for the binding.

A broker is required to provide an instance of this exchange named “amq.direct“. The Nameless Exchange is a special instance of the above exchange type where all queues are bound to this exchange automatically using the queue name as the routing criteria. This exchange instance has no public name, hence messages sent without specifying an exchange name are directed to this exchange.

Topic Exchange
The exchange does a wildcard match between the routing key and the routing pattern specified in the binding. The routing key is treated as zero or more more words, delimited by ‘.’ and supports special wildcard characters. “*” matches a single word and ‘#’ matches zero or more words.

(Click on image)

A broker is required to provide an instance of this exchange named “amq.topic“.

Fanout Exchange
Queues are bound to this exchange with no arguments. Hence any message sent to this exchange will be forwarded to all queues bound to this exchange.

(Click on image)

• One use case, is to use exchange chaining in a tree like hierarchy that can be used to push messages to a large number of subscribers.
• Another use case is where a direct exchange or a topic exchange can do the initial filtering which then forwards the message to a fannout exchange which will push the messages to all it’s queues.

A broker is required to provide an instance of this exchange named “amq.fanout“.

Headers Exchange
Queues are bound to this exchange with a table of arguments containing headers and values (optional). A special argument named “x-match” determines the matching algorithm, where “all” implies an AND (all pairs must match) and “any” implies OR (at least one pair must match).

(Click on image)

A broker is required to provide an instance of this exchange named “amq.match“.

How AMQP Supports Common Messaging Use Cases

The most common messaging use cases are point-to-point (or store and forward) and publisher/subscriber models. These models can be easily built on top of AMQP.

Point-to-Point
routing_key == queue_name

Pub/Sub
routing_key == topic_heirarchy_value

Next Part : Part5 - Reliability

Prev Part : Part3 - Flexible Routing Model

Background
Most pre-AMQP models had several issues with their routing models.

• Opaque routing models that were not explicitly defined.
• Since the semantics are not visible or explicit manipulating the routing model through the protocol was difficult.
• Rigid monolithic routing engines that had limited or no extensibility or compose-ability.

The AMQP Routing Model
One of AMQP ’s primary goals was to define a flexible, extensible and transparent routing model where the semantics are explicitly defined. This permits the definition of management commands to manipulate the routing model. The AMQP model consists of three components

• Exchange
• Queue
• Binding

AMQP defines a set of rules on how to compose these components in to processing chains. The routing model is analogues to how email works. The following diagram illustrates the routing model from a publisher and consumer’s point of view.

(Click on image)

This is analogues to a Mail Transfer Agent. Queues (or other exchanges) are bound to an exchange using a ‘Binding’. A publisher sends a message to an exchange. The exchange will accept the message and routes it to one or more queues (or another exchange) based on the bindings. An exchange completely decouples a publisher from queues and the consumers that consumes from those queues.

An exchange type defines a routing algorithm to match the bindings with a given message. Hence an exchange type represents a class of routing algorithm. An instance of an exchange type can be thought if as an instance of a routing algorithm. A broker can have multiple instances of an exchange type which are identified by there name. An exchange instance can have the following properties.

• Durable/Temporary
• Auto-Delete

Queue
This is analogues to a mail box. A queue will store the messages in memory or disk and deliver them to consumers. A queue binds itself to an exchange using a ‘Binding’ which describes the criteria for the type of messages it is interested in. Queues can have the following properties,

• Durable/Temporary
• Shared/Private (exclusive)
• Auto-Delete

Binding
This is analogues to a Routing Table. A binding defines the relationship between an exchange and a queue. In other words it defines the routing criteria. The most simple case is where the binding equals the queue name. A binding decouples a queue from an exchange. The same queue can be bound to any number of exchanges using the same criteria or different criteria. Different queues can be bound to the same exchange using the same routing criteria as well.

Routing Key
Is a special field (Header) present in the Message Delivery Properties. It can be thought of as a virtual address, analogues to a ‘To’ field in an email. An exchange may use this field to route a message. The standard exchange types defined in AMQP use the routing key in different ways to route messages.

Standard Exchange Types
AMQP defines several standard exchange types that are described in detail in the next blog entry.

Extending The Routing Model
One can define new exchange types with arbitrary routing criteria (routing algorithms). For example one can define an exchange that routes messages based on content (content based routing). Thus AMQP provides a standard way of extending the routing model without impacting interoperability.

Next Part : Part4 - Standard Exchange Types And Supporting Common Messaging Use Cases

Prev Part : Part2 - Achieving Interoperability And Avoiding Vendor Lock-in

Background
One of the key issues with any software is non-interoperability and vendor lock in. Most messaging systems prior to AMQP did not interoperate with each other. For example messages from Tibco’s Rendezvous couldn’t be routed through IBM’s MQSeries. If two messaging systems need to be connected, there are two options.

• Using a message bridge you could convert from one format to the other. However a bridge would be slow as the conversion adds latency. Also you would need to understand the wire format of each of those systems.
• Replacing one system with the other, which is costly and risky. Downtime can have a severe impact on the company’s revenue model.

Therefore once a messaging system is chosen, users are reluctant to change and locked in with the same vendor while spending large sums of money as licensing costs.

What if we have messaging systems(from different vendors) that can understand each other? If so connecting two messaging systems or replacing one system with the other can be done with minimum costs and risk. Since the semantics (behaviour) are the same the chance of something going wrong is relatively low. This is a key goal for the AMQP protocol.

So what does it take to achieve interoperability and avoid vendor lock-in?

• All brokers need to behave the same way
• All clients need to behave the same way
• Use a standard for commands on the wire
• Use a language neutral type System
• Use open standards and permit royalty free usage of such a protocol.

AMQP satisfies the above requirements by

• Defining a network wire-level protocol
• A defined set of messaging capabilities (The AMQP Model)
• A simple language neutral type system
• Using open, existing, unencumbered, widely implemented standards
• Providing royalty free usage of the protocol

Broker semantics are defined explicitly. One can partially imply the semantics from the wire-protocol. However we need to define the semantics explicitly in order to guarantee exact behaviour in each broker/client implementation. So the protocol defines a set of commands to manipulate state in a peer. These commands are grouped by functionality into classes. For example the Queue class has various methods to manipulate state within a broker.
Ex. queueDeclare, queueBind, queuePurge, queueDelete & queueQuerry

More details on the wire protocol will be discussed later on. The next few posts will focus on discussing the semantic model.

Next Part : Part3 - Flexible Routing Model

Prev Part : Part1 - Introduction

Advanced Message Queing Protocol is a an open standard with royalty free use. It has a strong focus on the financial services industry and provides the performance and reliability required by the said industry.

AMQP Working Group
The AMQP working group is responsible for defining and maintaining the sepcification. The website www.amqp.org provides details on licensing, FAQ, JIRA and a list contributors and a download page.

Version Numbering
The AMQP Version Numbering scheme consists of a major and minor revision number. Ex: 0-10.
Major versions > 0 will provide backward compatibility between minor versions. The current version of the protocol is 0-9. The AMQP WG is expected to release the 0-10 version shortly. Currently the specification is in a rapid growth stage.

Interoperability
AMQP provides complete interoperability via

• A network wire-level protocol
• A defined set of messaging capabilities a.k.a AMQP model

One can partially imply the semantics from the wire-protocol. However we need to define the semantics explicitly in order to guarantee exact behaviour in each broker implementation.

Reliability
The 0-10 (in-progress) version has provided functionality to ensure Guaranteed Delivery to satisfy the level of reliability required by the financial services industry. I will cover this topic in detail in a future blog post as part of the AMQP in 10 mins series.

Full Transaction Support
The 0-10 spec also provides protocol level functionality for distributed transactions. This topic will also be covered in detail in a future blog post.

Security
Security is the main focus for 0-11 version of the protocol. Currently there is only rudimentary support provided via TLS. Planned features include but not limited to partial encryption and payload signing.

Known Implementations
AMQP is platform agnostic and language neutral. Currently we have implementations from a wide variety of languages from procedural to object oriented to functional.

The Apache Qpid project has broker implementations in java and c++ and client implementations in java, c++, python, ruby and .NET.

RabbitMQ provides a broker implementation in Erlang and a client implementation in java

Open AMQ provides broker and client implementations in C.

What about JMS?
JMS can be implemented on top of AMQP. The AMQP WG is also working on standardising the AMQP-JMS mapping and is scoped for 0-11 version of the protocol. This will enable two different JMS over AMQP implementations to behave exactly the same way.
The Apache Qpid project provides an implementation of JMS over AMQP.

Next Part : Part2 - Achieving Interoperability And Avoiding Vendor Lock-in

Prev Part : Part0

From time to time people ask me what is AMQP? and how it is different from JMS? or why do we need another standard ..etc On the other hand the AMQP spec groups is finalizing it’s work on the 0-10 version.

So I decided to write a series of blog posts titled “AMQP in 10 mins”, where each part can be read and understood in 10 mins. I hope to write a part every working day, or more depending on my mood. Comments and criticisms are equally encouraged.

Next Part, Part1 - Introduction.