I’ve been doing a lot of work with chef lately and the default workflow really sucks. Typically what I will do is make a change, commit it, then push it up to github. Then I will login to my chef server and run rake install in /var/chef-repo to pull from github and upload all the new cookbooks and roles into the running chef instance. Then I will login to the target server and run chef-client to see if my changes actually worked. This is more or less how it’s described in the wiki, which is why I call it the default workflow.

There are quite a few unnecessary steps there. It would be pretty rad if I could just type git push production on my laptop and then have the chef server pick up all the changes and then automatically have all the hosts in my cluster run chef-client. I don’t quite have it automated to that level yet. However, I was able to make it so that a git push production will trigger an upload of all my changes into the chef server. Here’s how it works.

This is all assuming that you are using Ubuntu Karmic server, by the way. If you are on another platform you may or may not be shit out of luck.

Setup the appropriate ACL’s for your chef repo. This is necessary if you are working in a team environment and wish to allow other people to push chef cookbooks to production.

sudo apt-get install acl
sudo mount -o remount,acl /dev/sda1

Change the fstab entry for the disk containing your chef repo to include the acl option.

/dev/sda1       /           ext3    acl,defaults,errors=remount-ro,noatime    0 1

Create a group for the users who will be using this shared repo.

sudo groupadd chefadmin
sudo usermod cliff -G chefadmin

Then fix the permissions on your chef repo.

sudo chown -R root:chefadmin /var/chef-repo
sudo chmod -R g+w /var/chef-repo
sudo setfacl -R -m g:chefadmin:rwX /var/chef-repo
find /var/chef-repo -type d | xargs sudo setfacl -R -m d:g:chefadmin:rwX

Now you need to add a post-receive hook to your chef repo. Put the following in /var/chef-repo/.git/post-receive

#!/bin/sh

cd ..
env -i git reset --hard
rake roles upload_cookbooks

This hook will get run after a successful push to the repo. It will update the working tree by doing a hard reset and then run the rake tasks to upload everything to chef server.

And the last step is to simply add the git repo on your chef server as a remote.

git remote add production git+ssh://chef.mydomain.com/var/chef-repo/.git

Now git push production will automatically upload to your chef server and it will be usable by everyone in the chefadmin group as well.

I was at the NoSQL meetup in SF two weeks ago to present about Dynomite and I had a great time. There has been some discussion afterwards about the various performance numbers bandied about and I wanted to speak for Dynomite about the exact meaning of the performance numbers I present in my slides. And generally, I wanted to talk about what I think is a superior approach to performance testing that meets the needs of the potential users of a distributed database.

The Current State of Performance Testing

Mapping a Multidimensional Space

Any reasonably mature database project is going to have quite a few knobs to turn in order to have different effects on latency, scalability, and absolute throughput. For instance, in Dynomite you can change cache sizes, replication and quorum constants, and, of course, the actual storage engine used to persist the data. Factors like the OS used, the physical raid configuration, filesystem, number of CPU’s, RAM and network topology all figure in to the performance results that can be achieved by a distributed database. Naturally, the workload used to test a database has the greatest influence over the results.

When you put all of these things together you see that any values (latency, throughput, etc) obtained from a configuration are actually points in a multidimensional space, where each configuration parameter is a dimension. Trying to map the entire space results in a combinatorial explosion and does not even shed any light on database to database comparisons.

Rosy Results

The second issue with most performance tests is that the publicly available numbers (especially those appearing in slides) are produced by project maintainers. This creates a moral hazard: project maintainers are motivated to cast their software in the best light, even if the benchmarks are based on systems and configurations that skew far from common use. I know that if I configure Dynomite to not hit disk and use the native binary term protocol then I can get the throughput up to 20k reqs/s with random reads and writes. Of course, that means nothing since the vast majority of users want their data to be persistent.

Naysayers

Results that look a little too rosy will inevitably lead to a backlash of naysayers, which we are starting to see the seeds of today. Some of them have very real concerns, but others will actively seek to poison the conversation with misleading statistics and FUD about projects. Criticism is always welcome, but uninformed criticism is not helpful. One should also be wary of criticism leveled by purveyors of commercial database offerings, including hosted databases.

Possibly a Better Way

Equalizing the Datacenter

If we are going to advance, or at least be able to compare apples and apples, then we will need to settle on a basic procedure that can be followed for performance testing distributed databases. The first issue is machine equivalence. Most of the companies that sponsor open source distributed database development operate out of their own datacenter and thus will have different machine specs.

One option to start comparing fairly is to use EC2 for public performance testing. It would be easy enough to assemble a set of test scripts configurable for different cluster sizes and different configurations and spin up fresh EC2 nodes to run the benchmarks. The repeatability of such a scheme would allow for many benchmarks to be run to try and see how stable performance is in a shared network environment such as EC2. Even better, these scripts can be used by anyone to easily verify the publicly posted performance numbers of a project.

Of course EC2 does not offer what one would call blazing performance. 99.9th percentile performance for I/O is especially poor in some of the smaller EC2 instances. So whatever numbers a project posts from EC2, they are almost guaranteed to be worse than what can be had from dedicated hardware in a dedicated datacenter. Therefore the point is not to see how fast a database can perform on an absolute scale. The point is to equalize hardware variance between the tests of various databases.

Dataset Size

The underlying data structures used to persist data to disk can often be sensitive to the size of a dataset. The size of the dataset in absolute number of items, the average size of values in bytes, and the byte size of the dataset as a whole can all factor in to the performance of the database. For smaller dataset sizes competently written caching layers will make sure that reads almost never touch disk, for instance.

Therefore any real test of a data store will involve a dataset size several times that of available memory. The test will also have realistic sizes for the values. 10 bytes per value is in no way a realistic use case for a database. Something like 100 bytes per value on average or more is a much more realistic use case. Even better, tests can be made to load up real life datasets, such as the Netflix prize data or the AOL query logs. These will introduce some variance into the data. Sets of larger values might be used as well, such as Wikipedia data, a Freebase dump, or some other freely available set of larger (1k and above) values.

Workload Profile

This brings us to the workload profile under test. There are as many workloads as there are applications in existence, but many can be grouped with similar workloads, and a good performance test will hit the most common ones:

• Bulk load.

  A common database use pattern is to load a dataset for read-only serving to clients. For these applications quickly loading data from something like Hadoop is an important use case. Important metrics that can be gleaned from this workload are throughput in transactions/second, throughput in megabytes/second, and number of concurrent clients.

• Latency sensitive traffic. Read-only, write-only, and read / write mixture.

  This category of workload encompasses latency sensitive work. In other words, operations which are on the critical path for live traffic. There are a couple of important questions to be asked for this sort of workload.

  • What is the steady state latency?

    Latency should be measured with varying numbers of concurrent clients. Speaking for Dynomite, measuring at 12, 50, 100, and 400 concurrent clients per host seems, anecdotally, to give a good spread of latency results. Latency should also be aggregated into percentiles. At the very least 99.9%, 99%, and 50% as averages mask much relevant information.

  • What is the maximum throughput at saturation?

    Saturation is the number of concurrent clients at which the cluster’s throughput peaks. Typically the cluster will reach a point of diminishing returns where throughput will either go down or hold steady as more concurrent clients are added. The important metrics here are, of course, the maximum throughput and the number of concurrent clients at which that throughput is reached. Throughput should be measured in transactions / sec and megabytes / sec.

  • And at what number of concurrent clients does the cluster become overwhelmed?

    After reaching saturation the cluster throughput will either hold steady as the number of clients are increased (at the cost of client latency), or it will dramatically drop as the cluster becomes overwhelmed and starts experiencing failures, timeouts, and or memory overruns. On a graph of concurrent clients vs. throughput the point at which the cluster becomes overwhelmed should be pretty evident.

• Scanning.

  This category of workload is not very well supported (if at all) by the distributed key value stores. The bigtable clones, however, are mostly designed around exactly this kind of workload. Therefore if a distributed database supports scanning, this should be tested. A scanning test would ideally iterate over an entire table or datastore in whatever order the database imposes. The important metric for this test is the scanning throughput in transactions / sec and megabytes / sec.

An Effort Already Underway

Jon Travis has already started a project for testing out Project Voldemort. It is called vpork. This is a step in the right direction. As a community we should come together and start talking about how we test performance because it is important. We owe it to our users and we owe it to ourselves to be honest about the capabilities of our projects. Only through our commitment to quality, even at a cost to our own egos, will we build great systems.

“Murder your darlings.” - Sir Arthur Quiller-Couch

One of my goals with Dynomite is to make it a high throughput system. It should be able to scale horizontally as hardware is added. But it should also be able to scale vertically and fully utilize hardware with varying degrees of capability. I wanted to get a baseline throughput reading for what could be called a single large instance.

So I setup a test to see what a single node could handle in terms of throughput. The test environment contains 2 identical nodes. Each node is a full host (not virtualized) with 8 cores, 16GB ram and a number of disks put together in RAID5. One node runs an instance of Dynomite, the other runs the benchmark.escript test harness.

The first thing I found is that the thrift bindings for Erlang are absolutely terrible as far as performance goes. When loading the system to saturation with thrift it’s entirely CPU bound. Total throughput for a node regardless of configuration turned out to be right around 200 req/s. One bottleneck down.

After identifying thrift as a likely cause I changed the benchmarking script to generate load using Erlang’s native messaging. The results were much better. Dstat showed that Dynomite was no longer CPU bound. Fortunately there was more throughput than with thrift. Writes averaged about 366 req/s and reads were 1238 req/s. This was with the default configuration having 64 partitions. After playing with the configuration a bit I noticed that throughput went up when the number of partitions went down. In particular, writes went to 753 req/s and reads went to 1456 req/s.

So it bears looking into what I mean by partition. The Dynamo paper describes a system where each key is consistently hashed to a partition and then that partition is consistently hashed to a node. For the purposes of expedience I decided to make every partition on a node refer to a unique instance of the underlying storage engine. So for instance for the dets engine this means that in a single node cluster there are 64 dets files that are written to and read from when there are 64 partitions. This allows things like partition transfer and synchronization very easy to implement and I thought that it would improve throughput since there would be more processes to handle requests. Unfortunately, until I saw these throughput numbers the full implications of this design decision were not apparent. Having data scattered across 64 different files means that seek time is being something of a bottleneck.

So I tried using the ets engine, which is entirely contained in memory. Even then reads and writes were only topping out to about 2000 req/s. That’s pitiful for something which does not even hit disk. Indeed, when benchmarked on its own independent of Dynomite ets tops out at about 23,000 req/s. So I basically stripped Dynomite down until I could point to whatever was being slow. I wrote a custom profiling server and broke down everything on the critical path to doing an operation. What I found was that everything was pretty fast. On the order of microseconds. Everything except for the call to the storage server. That was taking 20 milliseconds on average. But when I looked at the benchmarks from inside the storage server, it was still taking on the order of microseconds to do its work. So somehow the context switch to the storage server was taking much longer than the actually work getting done. This points to an issue with the VM.

Luckily, Erlang R13A just came out. And while it’s not yet production ready, it is complete enough to test against. And the results with R13A were very impressive. For a single node the ets throughput went to 18,000 req/s. Dets throughput was 7000 req/s for writes and 6000 req/s for reads with the single partition configuration. Not bad at all.

So I spent a lot of time working through issues which were essentially problems in the VM. But I would not call it time wasted. I learned a good deal about the performance profile of Dynomite and gained some new insight into how certain patterns in Erlang will translate into runtime performance. I also gained some insight into what design changes need to happen in the future for Dynomite. I will have future write ups about the design changes I have in mind, but for the time being I would say that the current version is ready for release.

I’ve been doing a lot of throughput optimizations on Dynomite lately. I got the unsaturated latency down to a point where I was happy with it, so the next step was to optimize the throughput of a fully saturated node. Generally when I find something interesting or at least strike something off the list as being a culprit for throughput issues I’ll fire off a cryptic message into the aether of twitter. This morning I posted this:

And it piqued a lot more interest than I thought it would.

Deferring the response for a gen_server call is a popular performance optimization in Erlang applications. For some folks, it’s the default behavior they use handle_call callbacks that do not modify the state of the underlying server. The logic of doing so seems sound. After all, spinning up new processes is cheap and a gen_server call only handle so much throughput, so spawning off a process to handle each call should be a good strategy to eliminate bottlenecks. Generally, this strategy looks something like this:

handle_call(do_something, From, State) ->
  spawn_link(fun() ->
      Result = do_a_bunch_of_shit(State),
      gen_server:reply(From, Result)
    end),
  {noreply, State};

So in trying to reduce the bottlenecks in the dynomite code base, I tried doing this to the membership server. The membership server is in the critical path for all operations: get, put, has, and remove. The mediator will call membership:servers_for_key which hashes the key to its partition and will do a lookup in its routing table to figure out which nodes are currently serving that partition. I wrote a minibenchmark test to see what would happen.

{ok, _} = membership:start_link(a, [a,b,c,d,e,f]),
{Keys, _} = lib_misc:fast_acc(fun({List, Str}) -> 
    Mod = lib_misc:succ(Str),
    {[Mod|List], Mod}
  end, {[], "aaaaaaaa"}, 10000),
Start = lib_misc:now_float(),
lists:foreach(fun(Str) ->
    membership:servers_for_key(Str)
  end, Keys),
End = lib_misc:now_float(),
?debugFmt("membership can do ~p reqs/s", [10000/(End-Start)])

The original version of the servers_for_key callback:

handle_call({servers_for_key, Key}, From, State) ->
  Config = configuration:get_config(),
  Part = int_partition_for_key(Key, State, Config),
  Nodes = int_nodes_for_partition(Part, State),
  MapFun = fun(Node) -> {list_to_atom(lists:concat([storage_, Part])), Node} end,
  {reply, lists:map(MapFun, Nodes), State};

And the deferred response version:

handle_call({servers_for_key, Key}, From, State) ->
  Config = configuration:get_config(),
  spawn_link(fun() ->
  Nodes = int_nodes_for_key(Key, State, Config),
  Part = int_partition_for_key(Key, State, Config),
  MapFun = fun(Node) -> {list_to_atom(lists:concat([storage_, Part])), Node} end,
  gen_server:reply(From, lists:map(MapFun, Nodes))
  end),
  {noreply, State};

The difference in throughput between the two is pretty astonishing. The first version can do somewhere in the neighborhood of 85000 reqs/s. The second version can only manage about 12000 reqs/s. And in this case I think the difference actually makes sense because it’s essentially doing a CPU bound operation. So by spawning off we are essentially asking the CPU to do a lot more work to arrive at the same answer. There is the setup overhead for a new process. Then the overhead of copying the closure from the fun into that new process. Not to mention that once we’re dealing with multiple copies of that same membership data you can pretty much count out getting many cache hits at the CPU.

Figuring this out has challenged some of my assumptions about the Erlang VM. It’s also given me some ideas for other areas to look at in optimizing Dynomite performance. In particular, I am going to revisit the code around the pmap being used to distribute operations to the different nodes.

And the whole “closures and scope” thing was a reference to this hilarious exchange.

AppEngine is not google competing with AWS. It is an incubator for acquisition targets. Computer time is cheap, but developer time is not. So why not get a bunch of unpaid developers to write apps that are tailored to your platform. Then you can buy the good ones at a bargain basement price. The cost of migrating off the google platform will devalue the company to any other suitors.

Shipping around small files sucks. SCP takes forever and a day. Tossing them into a tarball and sending them over would take just as much time in total. Courtesy of Brian McCallister:

On the destination machine: nc -vl 9000 | cpio -id

And on the source machine: find ./my/data/lives/here | cpio -o | nc -v destination.blah.com 9000

Significantly faster. I clocked scp at 628k/sec and cpio/nc at 2183k/sec. Machines are on the same subnet. Screwing with the block size in cpio might result in better performance.