Pragmatic Dictator

The difficulty of constructing remote services is often not in writing them but testing and debugging whilst ensuring that some of the nastier types of failure (e.g. packet loss or machine failure) are adequately handled.

The norm for these kinds of testing scenarios is to have a full, mocked-up test environment with a bunch of servers. Such a setup needs sysadmin and repeated deployment steps which for most organisations are slow, ponderous things. Incremental test cycles in such an environment become costly which leads to onerous, last-minute testing and the late discovery of difficult to fix bugs that introduce endless release delays.

Over the years I’ve developed an approach for pushing all these testing scenarios back toward the unit level so they can be run regularly per build as they take mere minutes to complete. The core philosophy is to design the software in such a fashion that it runs on a single machine using all the network protocols it would use when deployed across many servers (ah, the power of localhost/127.0.0.1).

Preliminaries

Putting this philosophy into practice requires that we adopt certain design practices:

1. Clean separation between the transport/remote layers and the core service logic. This makes it easy to develop tests that verify the core logic without any remoteness concerns and a second set of tests that perform the more heavyweight remote tests. The benefit is that we can more easily isolate issues when they occur. For example, if the core logic tests pass but the remote tests fail we can be pretty confident the issue is in the remote layers.
2. Clean separation of configuration source from core service and transport/remote layer. This ensures all our software requests configuration using a consistent API which could then be implemented via LDAP, flat-files, in-memory etc. Such a setup allows us to easily build up configuration inside of our tests and make it available to the services we’re building.
3. Runtime discovery of endpoints. To allow us to dynamically allocate port/ip combinations and make them available to whichever services require them. One can achieve this via the abstracted configuration source but it’s often cleaner to have a dynamic lookup/discovery mechanism.
4. Configurable log file locations. So that we can avoid path clashes between services.

Once these things are in place, unit tests can construct transports, endpoints and configuration dynamically at run time in whatever combination is required for a test. It is thus possible to instantiate a collection of services inside of a single process and have them talk to each other as if they were all running remotely. This is somewhat at odds with other design practices where we typically look to remove remoteness when running services locally for purposes of performance.

Failure Scenarios

By virtue of the unit tests having control of all the services and their transports/endpoints it becomes possible to stop or disable services thus simulating machine failures but it’s also possible to extend the approach to cover problems such as packet loss, corruption or increased latency.

These more advanced scenarios are more readily handled with server construction toolkits such as Netty which allows tight control of packet processing and protocol. Using Netty, one can build up the protocol stack per service exactly as required and introduce Decoder/Encoder pairs, Handlers or wrappers around core service implementation that can randomly (and silently without severing the connection) lose messages or packets, break connections etc.

Example

I’ve been working on a Paxos implementation which breaks down into:

• State machines – Leader, Acceptor and Learner and associated elements such as leader election and failure detection.
• Persistent storage layer – as various state must be remembered across Paxos instances.
• Remote communications layer – including cluster membership and remote communications.

The state machines accept messages, make appropriate state transitions and produce messages. These are then passed around between participants via the remote communications layer. The persistent storage layer allows for specification of file locations at construction time which allows test code to allocate separate directories on a single-disk to hold respective state.

The remote layer is built such that none of the members need static/well-known ports to operate off. There is one exception which is a fixed multicast address that is used to do initial cluster discovery. It is implemented using Netty and consists of some codecs for the various messages and a handler that passes messages to and from the state machines.

There are several different implementations of the handler. There is the normal version that dispatches messages reliably and several others that randomly drop messages or lose them at critical moments in an instance of Paxos. The exact behaviour of these handlers is configured at runtime which allows unit tests to construct random or specific failure scenarios and ensure the state machines behave appropriately.

All these elements together allow unit tests to construct, in a single-process, fully remote services that communicate via TCP and UDP/Multicast as if they were running on a network and simulate failure scenarios. Alongside these tests are a collection to verify correct behaviour of the state machines and a set that validates their failure handling via timeouts, leader election behaviours etc. The entire suite including the failure scenarios runs in less than five minutes. That leaves one long-running test that exercises a collection of state machines concurrently for long periods, a necessary soak test run separately.

Alternative Implementations

A similar testing approach is possible with the likes of Jetty 7 as the lower IO layers are open enough to be customised to support these test scenarios. This can be a better option than Netty if services are Servlet based.

More challenging are the RPC-based services as these tend to run atop closed stacks that limit the amount of customisation possible and often have horrid configuration methods. However Thrift, by virtue of it’s Processor/Protocol abstraction can be readily modified to support such testing.

Sidenotes:

1. Applications that use databases for state storage can make this sort of testing tricky but it’s not impossible. One solution to the problem is to make use of virtual machines where one instantiates an image containing a pre-defined database and shuts it down afterwards alongside some scripts to prepare and tear down data within the database
2. I’ve recently applied this approach to several other systems including a trade management system written in Clojure, a trading platform written in Scala and a gossip-based directory service also written in Scala

Queues and Threads

An actor is essentially a queue of requests fed into some piece of single-threaded logic. Only one thread at any moment in time can be dispatching a request through the actor.

In essence this is a SingleThreadExecutor as found in java.util.concurrent.

The model is wonderfully simple at the surface but is hiding a number of challenging issues which aren’t easily solved without assistance from the actor/app developer.

Flow Control

The speed at which an actor’s queue is drained is related to the time taken for the actor to dispatch a request taken from the queue. There are also a number of runtime factors that can cause the queue to empty more slowly than might be expected including:

• Limited size thread of actor thread pool.
• The work being performed within the actor takes a long time generally.
• An actor doing I/O is slowed down due to contention on the device or because of a fault (e.g. a RAID enclosure slowing down whilst it restores a disk).

If actor queues fill faster than they can be emptied we will eventually exhaust resources and fail. It must be possible to do explicit flow-control and generate back-pressure on other components in a system which requires an actor infrastructure to expose metrics about queue sizes or provide other feedback about the state of the system (e.g. by posting events to a listener).

Partitioning

If we partition our actors by function then only one actor can perform some set of actions against some set of state owned by that function which represents a throughput limitation. An alternative is to partition the state across multiple actors each capable of performing the function. Or we can pass all state to be worked upon into an actor as part of the request such that we can run many stateless actors. The resultant and prior state must then be stored somewhere else (database or global memory or some other actor). In summary, we must either:

Ensure state is partitioned inside of actors in sufficiently granular fashion that we can provide enough throughput of operations to prevent overload.

or:

Ensure that we have sufficient actors capable of performing some function and can spread state across requests without contention.

[Sidenote: Achieving this efficiently in a system with dynamic load is not easy, one can of course take the easy option of over-provisioning ]

Network

The kind of network used for actor communication has significant effects on system characteristics. For the purposes of this conversation we’ll consider two kinds of network:

1. Processor/Memory bus – essentially guaranteed delivery, high throughput and low latency. Actors and requesters communicating via a bus are local to each other.
2. Ethernet – no delivery guarantee (a machine failure now doesn’t stop the entire system and packets can be lost silently for variable periods of time), low throughput, higher and more variable latency as compared to processor/memory bus. Actors and requesters communicating via a network are remote from each other.

[Sidenote: A lot of the literature considers both communication over buses and networks to be styles of distributed system. Generally, algorithms for the bus-based style are more plentiful, considerably less complex and easier to construct than for the network-based style see e.g. Lynch ]

An actor that is remote from an entity passing it requests requires two queues. One at the originator and one at the actor. The outbound queue in particular has substantially different characteristics from local actors and requesters:

1. Throughput – maximum throughput of the queue is closely related to the throughput of the ethernet between originating and remote machines. This is considerably lower than the throughput for a local queue.
2. Latency – the remote queue has the latency of the underlying ethernet and transport stack. Again, substantially different from the local queue.

Why does this matter? Because the number of requests one can dispatch through a remote queue per unit time is substantially different from a local queue. One cannot write a naive algorithm to dispatch requests evenly across a set of actors and assume they make progress at the same rate. Care must be taken to control flow in and out of queues to prevent resource exhaustion. Actor infrastructures that encourage these flow-naive algorithms are likely to exhibit difficult to debug failures in the face of unexpected/unanticipated load.

If one chooses an approach of passing state into an actor, then one must ensure the state is not large because it will take significant time to pull it across the network from storage (or another machine) and put it back. Further whilst the data is being transferred it cannot be modified by some other entity for fear of modifying old-state and losing some changes thus some form of conflict resolution is required (locks, vector clock based merge etc).

Availability

By virtue of machine and network failures (packet loss, outright failure etc), remote actors may become unavailable temporarily or permanently. A stateless actor can be respawned on another machine with little effort but a stateful actor requires some storage that is available across a number of nodes. Note that use of stateless actors will only eradicate the need for storage in the case where nothing else is required to hold state and pass it to actors for processing.

[Sidenote: The need for storage and the costly disk-syncing/checkpointing required to look after state so it can be made available to relocated actors impacts the relevance of actor benchmarks that concentrate on request throughput particularly in the multi-machine scenario. This is because storage and network become the dominant performance factors rather than the speed of request dispatch in real-world scenarios. Such a change may require us to adjust our actor/state partitioning approach ]

Should a machine be running a large number of actors and become unavailable, there is now a need to do significant and costly work to get these actors onto another machine. State must be moved, actors must be re-started and potentially re-balanced across remaining machines. Over the course of a re-balancing, various actor queues must be temporarily suspended which can lead to an increased backlog of requests which as above can cause resource exhaustion. Note also that when a machine becomes unavailable all queues on remote machines that were feeding it must be paused until new resource becomes available.

[Sidenote: Once one is running actors across multiple machines, it becomes necessary to have some form of directory service which can be used to find an actor. Without this feature it becomes difficult to relocate an actor (how do you find it's new address?). We must now keep this service up to date, make it tolerant of failures and scalable to potentially large numbers of actors ]

Summary

In a single-machine/multi-core environment, actor models can work well subject to being able to achieve an appropriate partitioning of function and/or state. This mode of operation also permits substantial simplicity in APIs but flow-control cannot be ignored.

Actor models will only work well in the multi-machine environment if they trade their simplicity for more complicated APIs and heightened developer awareness (see e.g. A Note on Distributed Computing) or are deployed into a very strictly controlled environment (increasingly difficult as a system grows).

Actors like many other concurrency models have their uses (though I’d claim they aren’t anything new and can be built in e.g. Java using an executor) but are not a panacea and their much-vaunted simplicity can be a double-edged sword.

My current company has for obvious business reasons got a serious interest in delivering a quality website experience during the World Cup and thus I’ve been spending a lot of time focused on our own performance and capacity management of late.

P&C is one of those 80/20 tradeoffs. There’s always more one can do or measure or test, equally getting the basics in place will deliver substantial benefit. I’d go further and argue that without a solid grasp of the basics, one cannot easily determine what else beyond that might be required. Here then are the basics that I’ve found myself repeating over and over:

• Have an enquiring mind – anomalies are not to be ignored or dismissed on the basis of pure speculation. Determining root cause is essential to prevent surprises in production. Some recent examples:
  1. In one test we noticed that every so often we’d get a substantial blip in disk I/O on servers that should be processing entirely out of memory. Along with that blip there’d be a corresponding reduction in throughput, we could have ignored it, after all things sorted themselves out relatively quickly but we chose to investigate. All these servers were periodically running a cleanup job the developers were unaware of and had not factored into their capacity calculations. The implications for production would have been a regularly overloaded, badly performing website. We’ve since tuned the jobs, adjusted their schedules and increased our capacity to ensure we can always spread the load around enough to accommodate them.
  2. An examination of the distribution of load on the boxes behind our load-balancers revealed a higher than expected amount of variance in CPU and connections. A review of the application revealed that any particular user’s traffic is sticky to one box, unfortunate as it’s stateless, time for a code change. We also spent time looking at the monitoring infrastructure and discovered that in certain cases we’d get false reports of 100% CPU utilisation, that one will be fixed with an OS patch.
• Gather the right data – there’s no value in allowing oneself to be limited by what is easily available via some set of tools people are comfortable with. One tool we were using had an unreasonably low ceiling on the number and rate of samples it could handle such that any graphs it produced showed hardly anything of the true profile of e.g. CPU utilisation, memory consumption or I/O. Forming any opinion about system behaviour in respect of load was going to be an exercise in speculation. We junked the tool and are looking for a replacement, in the meantime we’ve fallen back to making use of low level performance counters which we can sample local to the machine and whack onto disk for later analysis via scripts, opensource tools etc.
• Design tests that support reasoning – One should indeed try and replicate production load behaviours to judge overall system behaviour. The challenge of such testing is that it can be difficult to relate performance data back to exactly what was going on during some period of a test and make a diagnosis or be confident of an improvement. There are a number of things we can do to improve the situation:
  1. Ensure tests are deterministic such that any given run can be compared against other runs. This isn’t as simple as it looks when e.g. you wish to gradually increase load at a fixed rate that is being produced by more than one box.
  2. Have tests produce sufficient logging that one can easily identify what was going on at particular points in the sampled data. Logging of course can actually affect test behaviour and that isn’t always desirable.
  3. Build additional tests that target particular user journey’s through the system. Doing this for all possible journey’s can be costly so it makes sense to focus on testing those which are most popular with users. These kinds of tests restrict the reasoning tree making analysis, diagnosis and solution identification much easier.
• Measure what customers care about – they don’t care about CPUs, I/O or memory, they worry about things like response times. It is important to focus on maintaining a quality user experience not endlessly improving system efficiency. Considering user factors such as response times stops us expending huge effort on CPU utilisation when we should be focusing on say, network I/O, browser performance or reducing the amount of data we push to the browser before a page can render.
• Beware of averages – it is very tempting to combine datasets via the use of averaging unfortunately such a practice can easily hide spikes that might be indicative of a problem. On more than one occasion an engineer has presented a graph that tracks the average CPU and a table that summarises min, avg and max. After which they’ve pronounced load testing was a success and yet they have no explanation for why the average is never more than 50% but the max is 100% and whether or not this is good or bad.




• More than load – excessive focus on measuring the effect of a particular load can make us blind to another important metric, resource cost per unit of work – these are the collection of tests and analysis that help us understand what to tune and how much to keep our appetite for boxes and bandwidth reasonable. One simple thing teams can do per sprint (assuming you’re agile, why wouldn’t you be?) is point a profiler at each component and look for the low hanging fruit that is poor algorithm selection or inefficient code (e.g. repeated scanning of lists where a hashmap would be better or repeatedly computing something that could be cached).

Prioritisation is a solution that can be used in a few situations:

• Messaging – where some class of messages needs to be processed before one or more other classes.
• Job execution – where the results of some set of jobs need to be available before others.
• Levelling – where satisfying peak demand would require lots of hardware that in other periods would be significantly under-utilised.

It’s a very useful pattern but there are a few dark corners to think about:

1. Even low priority items have some importance, otherwise they wouldn’t exist at all. If there are too many high priority items passing through the system there is significant risk the low priority items will not be processed in an acceptable time period.
2. If there are too many high priority items passing through the system, the low priority items might not get processed at all leading to huge backlogs that take an age to process.
3. If the high priority items begin taking a large amount of time to process, low priority items are delayed with resulting in a huge backlog as above.

In essence, a certain workload mix can mean that one must wait infinitely for low priority items to be processed and that is rarely acceptable. Making prioritisation work effectively means ensuring that there is sufficient capacity to process all work within their respective acceptable time periods.

For some applications there is a convenient “quiet” period overnight where low priority items can be cleared out of the system as there’s a dearth of high priority items to process. In other cases processing of priority classes must be interleaved e.g. process 100 high priority items, then 5 low priority items and repeat. Alternatively one can dedicate varying sized pools of resource (partitioning) to processing priority classes with each pool scaled according to their timeliness requirements.

Some technical staff naively use priority to solve a throughput problem where capacity is insufficient to cope with all work in parallel. This can appear to work for a while if there are lulls in demand as mentioned above but ultimately, as workload increases such an approach will fail unless care is taken in profiling the workload and ensuring there is sufficient capacity to satisfy all priorities.

A programming language is a tool. These days in fact it’s more a toolbox as there’s an entire ecosystem associated with a language that makes it more or less suitable for a particular discipline (e.g. website development). There are many other tools beyond languages of course: CORBA, J2EE, SOAP, AJAX, Visual Studio .NET, Emacs etc

The obsession we have with our tools is verging on the sexual. We worship them, we endlessly compare them, we get excited about this or that extension. It drives much conversation in corridors and at conferences but it’s largely worthless because there’s no context.

Does a carpenter get excited about a saw, a power-drill or the latest hammer? Not really, because long ago they realised that whilst one must know how to make effective use of a tool and how to maintain it whilst it goes unused, what really matters is figuring out what the job itself actually is. This is the context that dictates which tools are appropriate.

We speculate about concurrency, we speculate about building websites, we speculate about writing this or that application but it’s all pointless until we actually set about a specific task with intent.

The smart techie has a good grasp of a wide range of tools, knows when to use them and ensures they have meaningful escape plans (that may never be implemented) in case the day comes when those tools turn out to be the wrong choice or need replacing. Most of all a smart techie puts thinking and planning well before worrying about tools.

In simple terms, we need to stop playing with our tools and focus on the real challenge, tackling real-world problems with elegant, simple, well thought out, maintainable, cost-effective solutions. Tools help you build such things but they aren’t the essence of it.