Single Core Architecture & Optimizations

The conceptual architecture of a single core system is deceivingly simple: single CPU, which is connected to a block of memory and a collection of other I/O devices. Turns out, simple is not practical. Even with modern architectures, the latency of a main memory reference (~100ns roundtrip) is prohibitively high, which combined with highly unpredictable control flow has led CPU manufacturers to introduce multi-level caches directly onto the chip: Level 1 (L1) cache reference: ~0.5 ns; Level 2 (L2) cache reference: ~7ns, and so on.

However, even that is not enough. To keep the CPU busy, most manufacturers have also introduced some cache prefetching and management schemes (ex: Intel's SmartCache), as well as invested billions of dollars into branch prediction, instruction pipelining, and other tricks to squeeze every ounce of performance. After all, if the CPU has a separate floating point and an integer unit, then there is no reason why two threads of execution could not simultaneously run on the same chip - see SMT. Remember Intel's Hyperthreading? As another point of reference, Sun's Niagara chips are designed to run four execution threads per core.

But wait, how did threads get in here? Turns out, threads are a way to expose the potential (and desired) hardware parallelism to the rest of the system. Put another way, threads are a low-level hardware and operating system feature, which we need to take full advantage of the underlying capabilities of our hardware.

Architecting for the Multi-core World

Since the manufacturers could no longer continue scaling the single core (power, density, communication), the designs have shifted to the next logical architecture: multiple cores on a single chip. After all, hardware parallelism existed all along, so the conceptual shift wasn't that large - shared memory, multiple cores, more concurrent threads of execution. Only one gotcha, remember those L1, L2 caches we introduced earlier? Turns out, they may well be the Achilles' heel for multi-core.

If you were to design a multi-core chip, would you allow your cores to share the L1, or L2 cache, or should they all be independent? Unfortunately, there is one answer to this question. Shared caches can allow higher utilization, which may lead to power savings (ex: great for laptops), as well as higher hit rates in certain scenarios. However, that same shared cache can easily create resource contention if one is not careful (DMA is a known offender). Intel's Core Duo and Xeon processors use a shared L2, whereas AMD's Opteron, Athlon, and Intel's Pentium D opted out for independent L1's and L2's. Even more interestingly, Intel's recent Itanium 2 gives each core an independent L1, L2, and an L3 cache! Different workloads benefit from different layouts.

As Phil Karlton once famously said: "There are only two hard things in Computer Science: cache invalidation and naming things," and as someone cleverly added later, "and off by one errors". Turns out, cache coherency is a major problem for all multi-core systems: if we prefetch the same block of data into an L1, L2, or L3 of each core, and one of the cores happens to make a modification to its cache, then we have a problem - the data is now in an inconsistent state across the different cores. We can't afford to go back to main memory to verify if the data is valid on each reference (as that would defeat the purpose of the cache), and a shared mutex is the very anti-pattern of independent caches!

To address this problem, hardware designers have iterated over a number of data invalidation and propagation schemes, but the key point is simple: the cores share a bus or an interconnect over which messages are propagated to keep all of the caches in sync (coherent), and therein lies the problem. While, the numbers vary, the overall consensus is that after approximately 32 cores on a single chip, the amount of required communication to support the shared memory model leads to diminished performance. Put another way, shared memory systems have limited scalability.

Turtles all the way down: Distributed Memory

So if cache coherence puts an upper bound on the number of cores we can support within the shared memory model, then lets drop the shared memory requirement! What if, instead of a monolithic view of the memory, each core instead had its own, albeit much smaller main memory? Distributed memory model has the advantage of avoiding all of the cache coherency problems we listed above. However, it is also easy to imagine a number of workloads where the distributed memory will underperform the shared memory model.

There doesn't appear to be any consensus in the industry yet, but if one had to guess, then a hybrid model seems likely: push the shared memory model as far as you can, and then stamp it out multiple times on a chip, with a distributed memory interconnect - it is cache and interconnect turtles all the way down. In other words, while message passing may be a choice today, in the future, it may well be a requirement if we want to extract the full capabilities of the hardware.

Turtles all the way up: Web Architecture

Most interesting of all, we can find the exact same architecture patterns and their associated problems in the web world. We start with a single machine running the app server and the database (CPU and main memory), which we later split into separate instances (multiple app servers share a remote DB, aka 'multi-core'), and eventually we shard the database (distributed memory) to achieve the required throughput. The similarity of the challenges and the approaches seems hardly like a coincidence. It is turtles all the way down, and it is turtles all the way up.

Threads, Events & Message Passing

As software developers, we are all intimately familiar with the shared memory model and the good news is: it is not going anywhere. However, as the core counts continue to increase, it is also very likely that we will quickly hit diminishing returns with the existing shared memory model. So, while we may disagree on whether threads are a correct application level API (see process calculi variants), they are also not going anywhere - either the VM, the language designer, or you yourself will have to deal with them.

With that in mind, the more interesting question to explore is not which abstraction is "correct" or "more performant" (one can always craft an optimized workload), but rather how do we make all of these paradigms work together, in a context of a simple programming model? We need threads, we need events, and we need message passing - it is not a question of which is better.

Getting started with Railtie

The documentation for Railtie is a great place to get started, but the interesting observation is that each of the major Rails components (Action Mailer/Controller/View/Record) is itself a Railtie, and Rails as you know it is simply pieced together by requiring all of the independent components. Even better, if your plugin or gem needs to hook into the Rails initialization process, all you need to do is inherit from Railtie and you are ready to go!

So how does Railtie know to call your classes and initializers? Railtie defines a self.inherited method, which is called anytime a subclass of Railtie is created, and stashes a reference to this class in a @subclasses variable. From there, the framework can simply call Railtie.subclasses and perform all the initialization as usual - a clever use of the Ruby object model.

Creating a Rails 3 plugin: 0-60

As a result of the refactor, creating Rails 3 plugins is at least several orders of magnitude simpler. In fact, as a simple exercise, lets create a custom notification plugin to print all of the internal Rails logging by also hooking into the new Rails::Notification API:

> railtie-plugin.rb

module NewPlugin
  # namespace our plugin and inherit from Rails::Railtie
  # to get our plugin into the initialization process
  class Railtie < Rails::Railtie
 
    # configure our plugin on boot. other extension points such
    # as configuration, rake tasks, etc, are also available
    initializer "newplugin.initialize" do |app|
 
      # subscribe to all rails notifications: controllers, AR, etc.
      ActiveSupport::Notifications.subscribe do |*args|
        event = ActiveSupport::Notifications::Event.new(*args)
        puts "Got notification: #{event.inspect}"
      end
 
    end
  end
end
 

Believe it or not, that is a fully functional Rails 3 plugin, in 10 lines of code. Simply drop this file into your load path and require it in your application. In fact, you can bundle this file into a gem, ship it to Gemcutter and with the help of Bundler, all your users have to do is specify "gem 'ournew-plugin'" in their Gemfile, and the rest is taken care of. With a little more imagination, and fun, we can extend our example above into a 'slowgrowl' plugin, which will - you guessed it - growl at the developer anytime it detects a slow code path in your app! Hat tip to Gavin Stark for the idea.

Getting started with Rails 3

Railtie while an important piece of Rails 3, is hardly the only component worth looking into. Also read up on the changes in ActiveRecord, the new router, watch the summary screencasts on the Rails site, and make sure to check out one of the many great Railscasts on the topic by Ryan Bates. It has been worth the wait, Rails 3 is easily head and shoulders above the competition.

The low level data provided by Speed Tracer finally allows us to peek under the covers to understand what the browser is actually doing - think strace, but for your browser. However, what if we could also bridge the gap between client-side and server-side tools? Wouldn't it be nice if we could go beyond the simple latency and response time reporting for network resources to viewing a full server-side log of what happened, all in one tool? Well, that's exactly what the Speed Tracer team launched at Google IO, so lets take a look at how it works!

Server-side tracing with SpeedTracer

The actual mechanics of getting server-side performance data into Speed Tracer are clever and simple. Whenever the browser receives a network response, it looks for the X-TraceUrl header, which specifies the relative URL and the unique ID of the trace for that specific request. From there, if the developer expands the network resource which provided the X-TraceUrl, Speed Tracer makes a request for the server-side trace, parses the JSON and surface it in the UI. This means that the server-side data is brought in on demand and does not affect the actual load times of your resources, and also that you need a mechanism to record, store, and serve these traces later.

Server side tracing with Rack & Ruby

The original announcement at Google IO demoed this new functionality on top of GWT and the Spring TC server. However, this same functionality is also easily extracted into a Rack middleware - which is what I did. In fact, here is a preview of sample trace from a Rails 3 application using rack-speedtracer:

The middleware takes care of providing the headers, persisting the traces, and then serving all the data back to Speed Tracer in the format it expects and understands under the hood. As a developer, you simply need to require the middleware, and then instrument your code where you want performance data to be recorded. Let's take a look at a simple configuration for a Rails 3 application:

> rails-speedtracer.rb

# in your Gemfile
gem 'rack-speedtracer', :require => 'rack/speedtracer'
 
# in development.rb environment
config.middleware.use Rack::SpeedTracer
 
# define a widgets controller
class WidgetsController < ApplicationController
  def index
    env['st.tracer'].run('Widgets#index') do
      env['st.tracer'].run("ActiveRecord: Widgets.all") do
        Widget.all
      end
 
      env['st.tracer'].run('Render') { render :text => 'oh hai' }
    end
  end
end
 

rack-speedtracer (SpeedTracer Rack middleware)

Downloads: 220 File Size: 0.0 KB

At the moment, you need to manually instrument your code to record this data, but there is also no reason why we couldn't automagically instrument Rails to provide detailed breakdowns of what is going on inside - rack/bug is a great example. Of course, this functionality is not limited to Rails, and will work for any Rack application. There are still a few missing pieces in rack-speedtracer (see readme - patches are needed and welcome!) before you can deploy it in production, but it will work great for development purposes.

Towards unified client & server debugging

Having both the server-side and client-side performance data in a unified view can make our lives, as developers, an order of magnitude simpler - tired of correlating server logs with your browser yet? Having a mechanism to view this data on-demand, and in a single place is exciting, and I hope that the Speed Tracer team continues to innovate in this direction. It would be great to see more metadata about code locations, exceptions, object instantiation, and so on from within the same UI. But in the meantime, do install it, play with it, and give rack-speedtracer a try.

Problem is, the CAP theorem proposed by Eric Brewer and later proved by Seth Gilbert and Nancy Lynch, shows that together, these three requirements are impossible to achieve at the same time. In other words, in a distributed system with an unreliable communications channel, it is impossible to achieve consistency and availability at the same time in the case of a network partition. Alas, such is the tradeoff.

"Pick Two" is too simple

The original CAP conjecture presented by Eric Brewer states that as architects, we can only pick two properties (CA, CP, or PA) at the same time, and many attempts have since been made to classify different distributed architectures into these three categories. Problem is, as Daniel Abadi recently pointed out (and Eric Brewer agrees), the relationships between CA, CP and AP are not nearly as clear-cut as they appear on paper. In fact, any attempt to create a hard partitioning into these buckets seems to only increase the confusion since many of the systems can arbitrarily shift their properties with just a few operational tweaks - in the real world, it is rarely an all or nothing deal.

Focus on Consistency

Following some great conversations about CAP at a recent NoSQL Summer meetup and hours of trying to reconcile all the edge cases, it is clear that the CA vs. CP vs. PA model is, in fact, a poor representation of the implications of the CAP theorem - the simplicity of the model is nice, but in reality the actual design space requires more nuance. Specifically, instead of focusing on all three properties at once, it is more productive to first focus along the continuum of “data consistency” options: none, weak, and full.

On one extreme, a system can demand no consistency. For example, a clickstream application which is used for best effort personalization can easily tolerate a few missed clicks. In fact, the data may even be partitioned by data centre, geography, or server, such that depending on where you are, a different “context” is applied - from home, your search returns one set of results, from work, another! The advantage of such a system is that it is inherently highly available (HA) as it is a share nothing, best effort architecture.

On the other extreme, a system can demand full consistency across all participating nodes, which implies some communications protocol to reach a consensus. A canonical example is a “debit / credit” scenario where full agreement across all nodes is required prior to any data write or read. In this scenario, all nodes maintain the exact same version of the data, but compromise HA in the process - if one node is down, or is in disagreement, the system is down.

CAP Implies Weak Consistency

Strong consistency and high availability are both desirable properties, however the CAP theorem shows that we can’t achieve both of these over an unreliable channel at once. Hence, CAP pushes us into a “weak consistency” model where dealing with failures is a fact of life. However, the good news is that we do have a gamut of possible strategies at our disposal.

In case of a failure, your first choice could be to choose consistency over availability. In this scenario, if a quorum can be reached, then one of the network partitions can remain available, while the second goes offline. Once the link between the two networks is restored, a simple data repair can take place - the minority partition is strictly behind, hence there are no possible data conflicts. Hence we sacrifice HA, but do continue to serve some of the clients.

On the other hand, we could lean towards availability over consistency. In this case, both sides can continue to accept reads and/or writes. Both sides of the partition remain available, and mechanisms such as vector clocks can be used to assist with conflict resolution (although, some conflicts will always require application level resolution). Repeatable reads, read-your-own-writes, and quorum updates are just a few of the examples of possible consistency vs. availability strategies in this scenario.

Hence, a simple corollary to the CAP theorem: when choosing availability under the weak consistency model, multiple versions of a data object will be present, will require conflict resolution, and it is up to your application to determine what is an acceptable consistency tradeoff and a resolution strategy for each type of object.

Speed of Light: Too Slow for PNUTS!

Interestingly enough, dealing with network partitions is not the only case for adopting “weak consistency”. The PNUTS system deployed at Yahoo must deal with WAN replication of data between different continents, and unfortunately, the speed of light imposes some strict latency limits on the performance of such a system. In Yahoo’s case, the communications latency is enough of a performance barrier such that their system is configured, by default, to operate under the “choose availability, under weak consistency” model - think of latency as a pseudo-permanent network partition.

Architecting for Weak Consistency

Instead of arguing over CA vs. CP vs. PA, first determine the consistency model for your application: strong, weak, or shared nothing / best effort. Notice that this choice has nothing to do with the underlying technology, and everything with the demands and the types of data processed by your application. From there, if you land in the weak-consistency model (and you most likely will, if you have a distributed architecture), start thinking how you can deal with the inevitable data conflicts: will you lean towards consistency and some partial downtime, or will you optimize for availability and conflict resolution?

Finally, if you are working under weak consistency, it is also worth noting that it is not a matter of picking just a single strategy. Depending on the context, the application layer can choose a different set of requirements for each data object! Systems such as Voldemort, Cassandra, and Dynamo all provide mechanisms to specify a desired level of consistency for each individual read and write. So, an order processing function can fail if it fails to establish a quorum (consistency over availability), while at the same time, a new user comment can be accepted by the same data store (availability over consistency).

The fact that a vanilla Rails application requires a dedicated worker with a 50MB stack to render a login page is nothing short of absurd. There is nothing new about this, nor is this exclusive to Rails or a function of Ruby as a language - whatever language or web framework you are using, chances are, you are stuck with a similar problem. But GIL or no GIL, we ought to do better than that. Node.js is a recent innovator in the space, and as a community, we can either learn from it, or ignore it at our own peril.

Measuring End-to-End Performance

A modern web-service is composed of many moving components, all of which come together to create the final experience. First, you have to model your data layer, pick the database and then ensure that it can get your data in and out in the required amount of time - lots of innovation in this space thanks to the NoSQL movement. Then, we layer our MVC frameworks on top, and fight religious wars as developers on whose DSL is more beautiful - to me, Rails 3 deserves all the hype. On the user side, we are building faster browsers with blazing-fast JavaScript interpreters and CSS engines. However, the driveshaft (the app server) which connects the two pieces (the engine: data & MVC), and the front-end (the browser + DOM & JavaScript), is often just a checkbox in the deployment diagram. The problem is, this checkbox is also the reason why the ‘scalability’ story of our web frameworks is nothing short of terrible.

It doesn't take much to construct a pathological example where a popular framework (Rails), combined with a popular database (MySQL), and a popular app server (Mongrel) produce less than stellar results. Now the finger pointing begins. MySQL is more than capable of serving thousands of concurrent requests, the app server also claims to be threaded, and the framework even allows us to configure a database pool!

Except that, the database driver locks our VM, and both the framework and the app server still have a few mutexes deep in their guts, which impose hard limits on the concurrency (read, serial processing). The problem is, this is the default behaviour! No wonder people complain about 'scalability'. The other popular choices (Passenger / Unicorn) “work around” this problem by requiring dedicated VMs per request - that's not a feature, that's a bug!

The Rails Ecosystem

To be fair, we have come a long way since the days of WEBrick. In many ways, Mongrel made Rails viable, Rack gave us the much needed interface to become app-server independent, and the guys at Phusion gave us Passenger which both simplified the deployment, and made the resource allocation story moderately better. To complete the picture, Unicorn recently rediscovered the *nix IPC worker model, and is currently in use at Twitter. Problem is, none of this is new (at best, we are iterating on the Apache 1.x to 2.x model), nor does it solve our underlying problem.

Turns out, while all the components are separate, and its great to treat them as such, we do need to look at the entire stack as one picture when it comes to performance: the database driver needs to be smarter, the framework should take advantage of the app servers capabilities, and the app server itself can't pretend to work in isolation.

If you are looking for a great working example of this concept in action, look no further than node.js. There is nothing about node that can't be reproduced in Ruby or Python (EventMachine and Twisted), but the fact that the framework forces you to think and use the right components in place (fully async & non-blocking) is exactly why it is currently grabbing the mindshare of the early adopters. Rubyists, Pythonistas, and others can ignore this trend at their own peril. Moving forward, end-to-end performance and scalability of any framework will only become more important.

Fixing the "Scalability" story in Ruby

The good news is, for every outlined problem, there is already a working solution. With a little extra work, the driver story is easily addressed (MySQL driver is just an example, the same story applies to virtually every other SQL/NoSQL driver), and the frameworks are steadily removing the bottlenecks one at a time.

After a few iterations at PostRank, we rewrote some key drivers, grabbed Thin (evented app server), and made heavy use of continuations in Ruby 1.9 to create our own API framework (Goliath) which is perfectly capable of serving hundreds of concurrent requests at a time from within a single Ruby VM. In fact, we even managed to avoid all the callback spaghetti that plagues node.js applications, which also means that the same continuation approach works just as well with a vanilla Rails application. It just baffles me that this is not a solved problem already.

The state of art in the end-to-end Rails stack performance is not good enough. We need to fix that.

Update: Thanks to Gregg Pollack and his awesome team at Envylabs, here is a recording of an updated and somewhat compressed version of the talk on this subject I gave at OSCON:

Of course, there is no single answer to that question - it depends on your application. AMQP is a great power tool for message routing, but there are other systems that can do a better job at specific tasks. One of such tools is Beanstalkd, which is a simple, and a very fast work queue service rolled into a single binary - it is the memcached of work queues. Originally built to power the backend for Causes Facebook app, it is a mature and production ready open source project. It just seems that not too many people talk about it, perhaps exactly because it works so well.

Beanstalkd Features & Recipes

Adam Wiggins recently published a great comparison of Beanstalk to a few other work-queue services, and speed is where it stands out. A single instance of Beanstalk is perfectly capable of handling thousands of jobs a second (or more, depending on your job size) because it is an in-memory, event-driven system. Powered by libevent under the hood, it requires zero setup (launch and forget, ala memcached), optional log based persistence, an easily parsed ASCII protocol, and a rich set of tools for job management that go well beyond a simple FIFO work queue.

Out of the box, Beanstalk supports multiple 'tubes' (work queues), which are created and deleted on demand. In turn, each job is associated with a single tube, and has a number of parameters: priority, time to run, delay, an id, and an opaque job body itself.

Once a job is inserted into the work queue, the server returns an ID, which we can use to inspect the job. From there, the queue itself is actually a priority heap! Need to jump a head in line? Set a higher priority on the job and Beanstalk will do the rest. Or, what if your worker goes down while processing a job? Because you specified a time to run on the job, Beanstalk will monitor the checked out job and put it back on the work queue if the timeout expires - seamless recovery, nice. Does the worker need more time to complete the job? There is a 'touch' command to notify the server to prolong the timeout. Have a bad job that you want to save for later inspection? Just bury it and take care of it later. Need to throttle all of the workers? You can pause the tube for a specified period of time. And there is more, do checkout the protocol specification.

Beanstalk at PostRank: Chronos

At PostRank we have dozens of Beanstalk processes sprinkled throughout which are being used for job management within the same machine and coordination between entire clusters. The larger deployments, which are the front-line coordinators to our crawlers are serving 50+ million jobs on a daily basis (average job is several kb), without breaking a sweat. On average, each job is just several kilobytes, but the numbers add up, meaning that a pure memory system would require 60GB+ of RAM to make it work for our use case. That is where the Beanstalk ASCII protocol, good old MySQL, and a little Ruby come together to create our scheduling system: Chronos.

The idea behind Chronos is based on a simple observation: we have tens of millions of crawler jobs, each of which is repeated on a custom interval, but only a small portion of that entire set needs to be in memory to make the system run! So, out of that observation, two projects were born: em-jack and em-proxy. EM-Jack is a Ruby Eventmachine client, which provides a simple mechanism to define custom command handlers that go beyond the native Beanstalk protocol. On the other hand, em-proxy is a protocol agnostic (layer 3) proxy, which allows us to intercept any TCP data stream and manipulate it at will.

So, instead of talking directly to Beanstalk, all the traffic is routed through our custom em-proxy (~150 LOC) which parses the Beanstalk protocol, intercepts custom commands, or simply inspects the "delay" parameter, and decides where the job should be routed: beanstalk or the MySQL instance. Jobs that are scheduled at least one hour into the future are persisted into the database, which significantly reduces the memory footprint. Finally, in the background, the upcoming jobs are silently loaded into beanstalk as their execution time approaches. Simple, reliable, scales well, and it gives us all the features available in Beanstalk for job management and coordination (plus persistence and replication of MySQL). A quick API demo:

> em-jack.rb

EM.run do
  jack = EMJack::Connection.new
  r = jack.put("my message", :ttr => 300) { |jobid| puts "put successful #{jobid}" }
 
  j = jack.reserve
  j.callback do |job|
    puts job.jobid
    puts job.body
 
    jack.delete(job) { puts "Successfully deleted" }
  end
end
 

em-jack (An evented beanstalk client)

Downloads: 346 File Size: 0.0 KB

Architecture Limitations and Alternatives

There is an abundance of different job and message queue systems, in part because they are so seemingly simple to write. However, if you ever tried to build your own, you will also know that the scaling and the features available in Beanstalk are non-trivial to replicate. However, there are a few limitations as well. Beanstalk does not currently support replication or any other form of high availability. Likewise, there is no native sharding (a few clients support it), or a native GUI. Other then that, it works, it is fast, it is easy to extend, and it is definitely worth a test drive.

Back in 2006, Google published a paper on "Chubby", a distributed lock service which gained wide adoption within their data centers. Zookeeper, not surprisingly, is a close clone of Chubby designed to fulfill many of the same roles for HDFS and other Hadoop infrastructure. The original paper from Google offers a number of interesting insights, but the biggest takeaway is: Chubby and Zookeeper are both much more than a distributed lock service. In fact, it may be better to think of them as implementations of highly available, distributed metadata filesystems.

Zookeeper as a Metadata Filesystem

At its core, Zookeeper is modeled after a straightforward, tree based, file system API. A client is able to create a node, store up to 1MB of data with it, and also associate as many children nodes as they wish. However, there are no renames, soft or hard links, and no append semantics. Instead, by dropping those features, Zookeeper guarantees completely ordered updates, data versioning, conditional updates (CAS), as well as, a few more advanced features such as "ephemeral nodes", "generated names", and an async notification ("watch") API.

Now, this may seem like a grab bag of features, but don't forget that all of this is functionality is exposed as a remote service. Meaning, any network client can create a node, update associated metadata, submit a conditional update, or request an async notification via a "watch" request, which essentially mirrors the inotify functionality exposed in the Linux kernel.

Likewise, to make things easier, a client can request for Zookeeper to generate the node name to avoid collisions (e.g. to allow multiple concurrent clients to create nodes within same namespace without any collisions). And last but not least, what if you wanted to create a node, which only existed for the lifetime of your connection to Zookeper? That's what "ephemeral nodes" are for - essentially, they give you presence. Now, put all of these things together, and you have a powerful toolkit to solve many problems in distributed computing.

Zookeeper: Distributed Architecture

Of course, if we are to deploy Zookeeper in a distributed environment, we have to think about both the availability and scalability of the service. At Google, Chubby runs on a minimum of five machines, to guarantee high availability, and also transparently provides a built-in master election and failover mechanisms. Not surprisingly, Zookeeper is built under the same model. Given a cluster of Zookeeper servers, only one acts as a leader, whose role is to accept and coordinate all writes (via a quorum). All other servers are direct, read-only replicas of the master. This way, if the master goes down, any other server can pick up the slack and immediately continue serving requests. As an interesting sidenote, Zookeeper allows the standby servers to serve reads, whereas Google's Chubby directs all reads and writes to the single master - apparently a single server can handle all the load!

The one limitation of this design is that every node in the cluster is an exact replica - there is no sharding and hence the capacity of the service is limited by the size of an individual machine. The Google paper briefly mentions a possible logical sharding scheme, but in practice, it seems that they had no need for such a feature just yet. Now let's take a look at the applications.

Applications on top of Zookeeper & Chubby

Within the Hadoop/HBase umbrella, Zookeeper is used to manage master election and store other process metadata. Hadoop/HBase competitors love to point to Zookeeper as a SPOF (single point of failure), but in reality, given the backing architecture this seems to be too often over-dramatized. Perhaps even more interestingly, as the Google paper mentions, Chubby has in effect replaced DNS within the Google infrastructure!

By allowing any client to access and store metadata, you can imagine a simple case where a cluster of databases can share their configuration information (sharding functions, system configs, etc), within a single namespace: "/app/database/config". In fact, each database could request a watch on that node, and receive real-time notifications whenever someone updates the data (for example, a data rebalancing due to adding a new database into the cluster). However, that is just the beginning. Since "ephemeral nodes" provide basic presence information, by having each worker machine register with Zookeeper, we can perform real-time group membership queries to see which nodes are online, and perhaps even figure out what they are currently doing.

What about consensus? With a little extra work we can also leverage the data versioning and notifications APIs to build distributed primitives such as worker queues and barriers. From there, once we have locks and consensus, we can tackle virtually any distributed problem: master election, quorum commits, and so on. In other words, it becomes your swiss army knife for coordinating distributed services.

Working with Zookeeper

Getting started with Zookeeper is relatively straightforward: download the source, load the jar file, and you're ready to experiment in a single node mode. From there, you can try interacting with Zookeeper via a simple shell provided with the project:

# create root node for an application, with world read-write permissions
[zk: localhost:2181(CONNECTED) 2] create /myapp description world:anyone:cdrw
Created /myapp

# Create a sequential (-s) and ephemeral (-e) node
[zk: localhost:2181(CONNECTED) 6] create -s -e /myapp/server- appserver world:anyone:cdrw
Created /myapp/server-0000000001

# List current nodes
[zk: localhost:2181(CONNECTED) 5] ls /myapp
=> [server-0000000000]

# Fetch data for one of the nodes
[zk: localhost:2181(CONNECTED) 8] get /myapp/server-0000000001
appserver

zookeeper (Mirror of Apache Hadoop ZooKeeper)

Downloads: 400 File Size: 0.0 KB

Similarly, we can perform all of the same actions directly from Ruby via several libraries. The zookeeper_client gem provides bindings against the C-based API, but unfortunately it doesn't support some of the more advanced features such as watches and asynchronous notifications. Alternatively, the zookeeper gem by Shane Mingins provides a fully featured JRuby version, which seems to cover the entire API.

And if you are feeling adventurous, you could even try the experimental FUSE mount, or the REST server for Zookeeper. For some ideas, check out game_warden (Ruby DSL for dynamic configuration management) built around zookeeper, and a few other articles on the project. Zookeeper is already in use in a number of projects and companies (Digg, LinkedIn, Wall Street), so you are in good company.

Dissecting Ruby MySQL drivers

The native mysql gem many of us use in production was designed to expose a blocking API: you issue a SQL query, and the library blocks until the server returns a response. So far so good, but unfortunately it also introduces a nasty side effect. Because it blocks inside of the native code (inside mysql_real_query() C function), the entire Ruby VM is frozen while we wait for the response. So, if you query happens to have taken several seconds, it means that no other block, fiber, or thread will be executed by the Ruby VM. Ever wondered why your threaded Mongrel server never really flexed its threaded muscle? Well, now you know.

Fortunately, the little known mysqlplus gem addresses the immediate problem. Instead of using a single blocking call, it forwards the query to the server, and then starts polling for the response. For the curious, there are also two implementations, one in pure Ruby with a select loop, and a native (C) one which uses rb_thread_select. The benefit? Well, now you can have multiple threads execute database queries without blocking the entire VM! In fact, with a little extra work, we can even get some concurrency out of ActiveRecord.

However, we could even drop threads in our quest for concurrency! Instead of making every thread poll on a socket, we could pass each of those sockets to a single event loop (EventMachine) library, and let it handle all the IO scheduling for us: gem install em-mysqlplus. Same API, in fact, it uses mysqlplus under the covers, but now every query has a callback for true non-blocking database access. Take a look at a few examples in the slides:

Non-blocking Rails with MySQL

Now we come around full circle. The downside of a true asynchronous library is that it requires callbacks, spaghetti code and a fully asynchronous stack. Thankfully, we already have Thin for our async app server, and with the introduction of Fibers in Ruby 1.9, we can wrap our asynchronous driver to behave just as if it had a blocking API.

So, we install em-mysqlplus, require em-synchrony to emulate the 'blocking api', implement an activerecord adapter, and we finally have a fully non-blocking ActiveRecord driver which we can drop into our Rails app! Well, almost, a few other modifications: Rails provides a threaded ConnectionPool, which we need to replace with a Fiber aware one, and finally, we need to disable the built in Mutex (hap tip to Mike Perham for doing all the dirty work for us). Now let's give it a try:

> widgets_controller.rb

class WidgetsController < ApplicationController
  def index
    Widget.find_by_sql("select sleep(1)")
    render :text => "Oh hai"
  end
end
 

thin -D start
ab -c 5 -n 10 http://127.0.0.1/widgets/

Server Software: thin
Server Hostname: 127.0.0.1
Server Port: 3000

Concurrency Level: 5
Time taken for tests: 2.210 seconds
Complete requests: 10
Requests per second: 4.53 [#/sec] (mean)

Our widget action simulates a blocking one-second query, we start up a single thin server, and run an ab test against it: 10 requests, with a max concurrency of 5. And as you can see, the test finishes in just slightly over 2 seconds!

Rails 3, Ruby 1.9 and Drizzle

By mid summer we will see production releases of Rails 3, Ruby 1.9, and Drizzle, and that convergence is worth paying attention to. Both Rails 3 and Ruby 1.9 offer raw performance improvements across the board. In the meantime, Drizzle already provides a fully async libdrizzle driver (talks to MySQL & Drizzle) which we could adopt to future proof our applications. Combine all three with a fibered ActiveRecord driver, an async application server such as Thin, and we could make some serious steps forward when it comes to performance of Rails: significantly lower memory footprint and much better performance across the board.

Turns out, the reactor pattern performs extremely well under heavy loads (C10K challenge), hence the continuous rise in adoption (Nginx, Beanstalkd, EventMachine, Twisted, Node.js), but it does have its downsides: it requires reactor-aware libraries, still relies on background processing capabilities for long running computations, and last but not least, once you have nested several callbacks, it results in much more complicated code. Functional purists will disagree with the last statement – after all, we all love JavaScript, and node.js is the new hot thing on the block – but what if we could write event driven code without the added complexity imposed by hundreds of nested callbacks?

Accidental Complexity of Event-Driven Code

Anyone who has written a non-trivial event driven application will be familiar with the following pattern: you often start reading your code bottom-up and then navigate your way up the callback chain. In addition, since there is no single execution context, each callback requires its own nested exception handling, which adds to the complexity - debugging is non-trivial, to say the least. Of course, this is usually not too bad in a context of a simple demo, but it also quickly spirals out of control.

> em-callbacks.rb

EventMachine.run {
  page = EventMachine::HttpRequest.new('http://google.ca/').get
  page.errback { p "Google is down! terminate?" }
  page.callback {
    about = EventMachine::HttpRequest.new('http://google.ca/search?q=eventmachine').get
    about.callback { # callback nesting, ad infinitum }
    about.errback  { # error-handling code }
  }
}
 

Call me old-fashioned, but I much prefer the if-then-else control flow, with top-down execution and code I can actually read without callback gymnastics. And as luck would have it, turns out these are not inconsistent requirements in Ruby 1.9. With the introduction of Fibers, our applications can do fully-cooperative scheduling (worth a re-read to make sense of the rest), which with a little extra work also means that we can abstract much of the complexity of event driven programming while maintaining all of its benefits!

Fibers & EventMachine: under the hood

Ruby 1.9 Fibers are a means of creating code blocks which can be paused and resumed by our application (think lightweight threads, minus the thread scheduler and less overhead). Each fiber comes with a small 4KB stack, which makes them cheap to spin up, pause and resume. Best of all, a fiber can yield the control and wait until someone else resumes it. I bet you see where we're going: start an async operation, yield the fiber, and then make the callback resume the fiber once the operation is complete. Let's wrap our async em-http client as an example:

> em-fiber-wrap.rb

def http_get(url)
  f = Fiber.current
  http = EventMachine::HttpRequest.new(url).get
 
  # resume fiber once http call is done
  http.callback { f.resume(http) }
  http.errback  { f.resume(http) }
 
  return Fiber.yield
end
 
EventMachine.run do
  Fiber.new{
    page = http_get('http://www.google.com/')
    puts "Fetched page: #{page.response_header.status}"
 
    if page
      page = http_get('http://www.google.com/search?q=eventmachine')
      puts "Fetched page 2: #{page.response_header.status}"
    end
  }.resume
end
 

First thing to notice is that we are now executing our asynchronous code within a fiber (Fiber.new{}.resume), and our http_get method sets up the call, assigns the callbacks and then immediately yields control as it tries to return from the function. From there, EventMachine takes over, fetches the data in the background, and then calls the callback method, which in turn resumes our fiber, passing it the actual response. A little bit of fiber gymnastics, but it means that our original code with nested callbacks can now be unwound into a regular top-down execution context with if-then-else control flow. Not bad!

EM-Synchrony: Evented Code With Less Pain

Of course, we wouldn't gain much if the net effect of introducing fibers into our event driven code was swapping callbacks for fiber gymnastics. Thankfully, we can do better because much of the underlying implementation can be easily abstracted at the level of the driver. Let's take a look at our new helper library, em-synchrony:

> em-synchrony.rb

EventMachine.synchrony do
  page = EventMachine::HttpRequest.new("http://www.google.com").get
 
  p "No callbacks! Fetched page: #{page}"
  EventMachine.stop
end
 

Instead of invoking the default EM.run block, we call EM.synchrony, which in turn wraps our execution into a Ruby fiber behind the scenes. From there, the library also provides ready-made, fiber aware classes for some of the most common use cases (http: em-http-request, mysql: em-mysqlplus, and memcached: remcached), as well as, a fiber aware connection pool, iterator for concurrency control, and a multi-request interface. Let's take a look at an example which flexes all of the above:

> em-synchrony-ex.rb

EM.synchrony do
 
  # open 4 concurrent MySQL connections
  db = EventMachine::Synchrony::ConnectionPool.new(size: 4) do
    EventMachine::MySQL.new(host: "localhost")
  end
 
  # perform 4 http requests in parallel, and collect responses
  multi = EventMachine::Synchrony::Multi.new
  multi.add :page1, EventMachine::HttpRequest.new("http://service.com/page1").aget
  multi.add :page2, EventMachine::HttpRequest.new("http://service.com/page2").aget
  multi.add :page3, EventMachine::HttpRequest.new("http://service.com/page3").aget
  multi.add :page4, EventMachine::HttpRequest.new("http://service.com/page4").aget
  data = multi.perform.responses[:callback].values
 
  # insert fetched HTTP data into a mysql database, using at most 2 connections at a time
  # - note that we're writing async code within the iterator!
  EM::Synchrony::Iterator.new(data, 2).each do |page, iter|
    db.aquery("INSERT INTO table (data) VALUES(#{page});")
    db.callback { iter.return(http) }
  end
 
  puts "All done! Stopping event loop."
  EventMachine.stop
end
 

em-synchrony (Fiber aware EventMachine)

Downloads: 408 File Size: 0.0 KB

Synchrony implements a common pattern: original asynchronous methods which return a deferrable object are aliased with "a" prefix (.get becomes .aget, .query becomes .aquery) to indicate that they are asynchronous, and the fiber aware methods take their place as the defaults. This way, you can still mix sync and async code all in the same context (within an iterator, for example). For more examples of em-synchrony in action, take a look at the specs provided within the library itself.

Towards Scalable & Manageable Event-Driven Code

Event driven programming does not have to be complicated. With a little help from Ruby 1.9 much of the complexity is easily abstracted, which means that we can have all the benefits of event-driven IO, without any of the overhead of a thread scheduler or complicated code. Node.js, Twisted, and other reactor frameworks have a lot going for them, but the combination of EventMachine and Ruby 1.9, to me, is a clear winner.

We should also mention that fibers have been ported to Ruby 1.8, but due to their implementation on top of green threads they incur much larger overhead - in other words, this is your reason to switch to Ruby 1.9! And last but not least, fibers or not, don't forget that Ruby 1.9 still has a GIL which means that only one CPU core will be used. Until we see MVM (multi-VM) support, the solution is simply to run multiple reactors (one or more for each core).

Setting aside the fact that we are yet to define what “NoSQL” actually is, some of the attributes that we commonly glob under this label are: document based, schema-free, distributed and “scalable”. The fact that being distributed and being scalable are not one and the same is a subject for another post, instead let’s take a closer look at what schema-free and document-based actually means. In fact, let me jump ahead: I am genuinely surprised that we are yet to see a schema-free engine built on top of MySQL. I know, I know, but suspend you disbelief for a second, because it is not as outrageous as it sounds.

Document Based: a Double Edged Sword

The original reason for and the benefit of the relational model is that by constraining the data schema (read, eliminating structural complexity of the data, or decomposing it into relations), you actually gain power and flexibility in the types of queries you can execute against your database. Said another way, normalized data design allows us to have a general-purpose query language, which allows for queries whose parameters we do not even know at design time, whereas denormalized designs do not. What we loose in flexibility of our data structures, we gain in our ability to interact with the data. Hence, in theory, if you have no way to anticipate the types of queries in the future, a relation model is your best bet. Lose some, win some, chose your poison.

At the same time, we all know that “no join is faster than no join”. The inherent disadvantage of decomposing your data is the required assembly. If you are looking for “speed” or “scalability”, then denormalizing your data is usually the first step. The disadvantage? Now you have introduced a number of potential anomalies into your data: updates, inserts, and deletes can cause data inconsistencies unless you keep careful accounting of all duplication. One-to-One, and One-to-Many relations are usually easy to manage, but Many-to-Many in denormalized schemas are nothing but a recipe for disaster. That is, if you care about consistency.

Finally, since you lose the power of a general purpose query language (SQL), you are now at a mercy of the DSL provided by your new database. Mongo, Couch and many others had to introduce their own query language constructs alongside "map-reduce" functionality to address the problem of querying arbitrarily deep records. Now, I am a fan of both, but frankly, none I have worked with so far are as clean, or as easy to understand as SQL (case in point) - with the downside of making me learn yet another query language.

Schema-free != Document Based

Document based and schema-free are often used interchangeably, but there is an important difference: schema-free does not necessarily imply nested data structures. Likewise, just because MySQL is “relational” does not mean that it must be fixed to a predefined schema - at create time, maybe, but not at runtime. Intersect the two statements, and it means that there is absolutely no reason why we cannot have a schema-free engine in MySQL:

> schema-free.sql

mysql> USE noschema;
mysql> CREATE TABLE widgets;  /* look ma, no schema! */
mysql> INSERT INTO widgets (id, name) VALUES("a", "apple");
mysql> INSERT INTO widgets (id, name, type) VALUES("b", "blackberry", "phone");
 
mysql> SELECT * FROM widgets WHERE id = "a";
+---------+---------------+
| id      | name          |
+---------+---------------+
| a       | apple         |
+---------+---------------+
 
mysql> SELECT * FROM widgets;
+---------+---------------+--------+
| id      | name          | type   |
+---------+---------------+--------+
| a       | apple         | NULL   |
| b       | blackberry    | phone  |
+---------+---------------+--------+
 

As long as we avoid nested data structures, then there is no reason why we should be limited by the columns defined in our tables because we can compose and decompose any relation at runtime. Not only would this mean no migrations or need to store null values, but you could also keep all the tools, drivers, and the SQL query language while adding the full flexibility of being schema-free.

Schema-free DB on top of MySQL

Not able to find any project that would give me this behavior, I ended up prototyping it myself over the weekend, and believe it or not, it works just fine. In fact, the output above is from a real console session with MySQL. All it took is an em-proxy server with a little low-level protocol and query rewriting, and all of the sudden, my MySQL forgot that it requires a schema. Take it for a test-drive yourself (you will need Ruby 1.9):

git clone git://github.com/igrigorik/em-proxy.git && cd em-proxy
ruby examples/schemaless-mysql/mysql_interceptor.rb
mysql -h localhost -P 3307 --protocol=tcp

> schema-free-mysql.rb

# snip ... 
# build the select statements, hide the tables behind each attribute
join = "select #{table}.id as id "
tables.each do |column|
  join += " , #{table}_#{column}.value as #{column} "
end
 
# add the joins to stich it all together
join += " FROM #{table} "
tables.each do |column|
  join += " LEFT OUTER JOIN #{table}_#{column} ON #{table}_#{column}.id = #{table}.id "
end
 
join += " WHERE #{table}.id = '#{key}' " if key
 

mysql_interceptor.rb (MySQL Proxy in Ruby)

Downloads: 493 File Size: 0.0 KB

Of course, this is nothing but a cute code example nor does it even cover all the different use cases, but let us look at the feature set: driver support for every language (you can point Rails + ActiveRecord, JDBC, etc. at it out the box, no problem), tool support (GUI and command line), replication that works, basically impossible to corrupt, transactions, and so on. Not bad for half a day of hacking with a simple data model in the background:

Instead of defining columns on a table, each attribute has its own table (new tables are created on the fly), which means that we can add and remove attributes at will. In turn, performing a select simply means joining all of the tables on that individual key. To the client this is completely transparent, and while the proxy server does the actual work, this functionality could be easily extracted into a proper MySQL engine - I’m just surprised that no one has done so already. For a closer look, check out the proxy code itself, there are plenty of comments, which explain how it is all pieced together.

The gray zone of SQL vs NoSQL

So what is the point of all this? Well, I hope someone actually writes such an engine, because I believe there is a market for it. There is a lot to be said for a drop in, SQL compatible, schema-free engine, and unlike what the NoSQL propaganda may say, there is absolutely no reason why we can’t have many of the benefits of “NoSQL” within MySQL itself. There is no one clear winner for a database engine or model, so put some thought into your decision up front. Just because Mongo, TC, or Couch are 'document-oriented' or 'schema-free' does not mean they are necessarily better for your application. In the meantime, don't get me wrong, I am still rooting for all the NoSQL projects, as well as have high expectations for Drizzle - they are all doing fantastic work.

Belorussian translation kindly provided by Patricia Clausnitzer.