There’s a number of tricks people use when profiling and debugging UNIX daemons.

The problem with a daemon is that it’s not generally connected to a user interface. It sits there in the background accepting requests through some communications medium, and sending responses back. By its very design, it’s meant to get out of the way and be unobtrusive.

So when you are trying to find out what a daemon is doing, in order to test it or to diagnose a problem, tricks are required.

Read the rest of this entry »

Most replicated databases provide ‘eventual consistency’.

The reason for this is simple: replication, by definition, means there are multiple copies of the data lying about. And whenever something is changed, that change has to be applied to all the copies. And since doing that takes time, there will inevitably be situations where some of the copies have performed a given update, and some haven’t. This is known as ‘inconsistency’ – the replicas not being all the same. Applications have to be written to expect this; if you issue an update to the database, you can’t expect to get the new data back if you select it immediately. If the user submits a form that changes how the next page is displayed, that page can’t just read the database to generate itself (as it might do when being visited *not* from the form); the page has to manually take account of the change to reflect it, as the database might not. And users get frustrated if they change something, but it doesn’t seem to ‘work’. They submit the change again and again, putting more load on your system.

Eventual consistency means that there may be inconsistency while updates are ‘in flight’, but once this (usually short) period is over, unless more updates arrive, the replicas will become consistent again; and any given change will ‘eventually’ happen everywhere.

It’s possible to avoid this; using transactions and a two-phase commit protocol or more advanced systems like Paxos, one can arrange that all transactions are globally serialized (or an illusion thereof), so no node can ever see an inconsistent database state. However, doing so puts load on the system due to the extra messages interchanged and the extra temporary storage required for in-progress transactions while they are being agreed, and harms application latency due to the extra synchronisation required before doing anything, and can harm throughput due to making things wait on locks, or retrying transactions due to optimistic concurrency control detecting a conflict – and it cannot be enforced at the same time as the system is running with a network partition; if a network link fails you have to either abandon consistency or bring all or part of the system to a halt until communication is restored. There can be no globally consistent state without a way of communicating globally!

So strictly consistent replication is only used for systems that need absolute consistency enough to justify the costs.

However, here at GenieDB, we’ve discovered a way to get strict consistency cheaply for most operations, while the network is fully connected; and when network links fail, we fall back to eventual consistency in a graceful manner. By “most operations”, I mean:

• Any SELECT of a record by its primary key will consistently return the most recent state of that record, even if that is the deletion of the record
• Any other SELECTs – by non-primary key columns, by range of primary keys, or whole table scans – will consistently return the most recent state of the records it finds (even if they are deleted), but may not see newly created records

The way it works is quite simple: whenever we update a record (including deletions), we hash the primary key to choose a ‘primary server’ for that record; and we synchronously send the new state of the record to that server. We start an asynchronous ‘eventual’ replication of the change to the nodes in the background, and return control to MySQL.

When we fetch a record by primary key, we ask the primary server for that record first to see if there’s a recent version; if so, we use that rather than reading from our local disk. If it’s down or unreachable, we just use our local copy after all – which might be inconsistent, but is better than nothing.

So once a write has returned as complete, a read will always return the new data, no matter if replication is still in progress.

When the database is read without knowing a specific primary key – those table scans and secondary-index scans – we consult our local disk to find eligible primary keys, then look them up on their primary servers. If the records have been deleted, or the secondary index field changed so they are no longer eligible for the query, they’re skipped. Otherwise, we return their consistent current values from the primary servers.

Distributing records by hashing is ordinarily used as a cache, such as memcache, to allow an application to retrieve information quickly without needing to send out relatively expensive SQL queries. As our core engine is a lot faster than a MySQL database, all this talking to primary servers actually harms our performance due to the network latency; so we allow the application to turn the consistency algorithm off on a per-operation basis, for situations where eventual consistency is good enough.

What we provide isn’t quite strict consistency; it’s strictly consistent for all but a few specific ‘tricky’ kinds of query, and even then, it’s almost strictly consistent; and even in the face of multiple network failures it can never be worse than eventual consistency. But that covers a lot of cases. Enough that, for many applications, they can be written to assume strict consistency, and the odds of a network outage being kept low enough that the cost of what amounts to a few outdated selects per year is negligible compare to the cost of engineering the application to cope with eventual consistency.

This kind of thing is the core of what we’re doing here at GenieDB – database technologies such as replication offer seductive advantages, but usually at some unpleasant cost. We’re finding ways to reduce those costs, to make the advantages of these technologies more widely available and easier to use. And where we can’t remove them all, we make them optional on as fine-grained a level as we can, so developers can pick and choose throughout their application!

When somebody reports a bug in software you’re responsible for, the worst thing that can possibly happen is to find that it’s somebody else’s fault.

If it’s your fault, you see, then you can fix it! And since it’s a problem in code you wrote, it’s usually not too hard for you to figure out what’s wrong in the first place.

But if it’s not immediately obvious what could be going wrong, and the trail of diagnosis seems to be leading into a third-party component you use, then a dull, throbbing, headache sets in. Bugs in third-party components are hard often hard to trace (as you’re not familiar with their code, even if you have access to the source), and hard to fix.

So we were delighted when, upon approaching Oracle with the suspicion that our software was deadlocking might be Berkeley DB’s fault, they promptly replied with some suggestions as to what to try – and when we subsequently produced an isolated test case that demonstrated the problem, they came back to us the next working day with a diagnosis of the problem and a patch!

The issue was simple: we’re using lock timeouts as part of our deadlock-handling strategy. We want locks to timeout in the core server, but never in clients, as they may not be able to restart their transactions. So we set a lock timeout of a few seconds in the server (which never performs operations lasting more than a second or so anyway), and set a timeout of zero (meaning “never time out”) in the clients.

But we were seeing deadlocked systems, with the server lock failing to time out. The fact that db_stat showed the lock sitting there with an expired timeout pointed the finger of blame therein.

So we produced a small standalone application that used Berkely DB in exactly the same way we did at the point where it got stuck, and lo, managed to reproduce it.

Sending this test app to Oracle, we were amazed to have an explanation of the problem, plus a patch, within about three hours!

It seems almost fashionable to write about concurrency bugs these days. Books like Coders at Work and Beautiful Code contain harrowing accounts of Real Programmers finding subtle concurrency bugs in vast code bases.

Concurrency bugs are difficult to find, because they tend to involve the interaction between potentially widely-separated bits of code in unpredictable ways – and often only when several independent evens happen to coincide within narrow timing windows.

As we’re writing a distributed database here at GenieDB, we’re the kinds of people who are wise in the art of concurrency bugs, so we generally avoid them in the first place by only introducing concurrency where it is beneficial to do so. Obviously, the nodes in a distributed system execute concurrently, so we’ve kept the protocols for interacting between them simple and easy to analyse, and done our best to minimise state dependencies between them. This means that the ordering and timing of events should be unable to reveal unpleasant surprises. Within our software on each node, we’ve introduced concurrency only where we have to, in order to avoid blocking or to maximise performance. The interfaces between threads is very carefully marked, with the minimum of shared state; and we carefully reason about, and document, the expectations for managing that shared state, to avoid slip-ups in future.

However, this does not make us smugly immune to concurrency bugs; we fixed one yesterday that had been puzzling us for several days.

One of our daemons, when run within the test cluster harness, was sometimes dying while starting up. No core dump, no error message, nothing – it just disappeared midway through initialisation. We’d never had this problem before; it was only happening in the new Solaris port. Annoyingly, it was very timing-sensitive; running it under tools like truss or gdb made it always start up successfully, so we couldn’t see what was going wrong. Some of the logging we put in for tracing had to be taken out, as the problem never surfaced with it in place. It would always start up flawlessly from the command prompt; the problem only happened when it was run through our cluster testing harness, which ssh-ed into a group of servers and started the daemons on each.

But through a slow (and infuriating) process of deduction, we realised it was dying shortly after daemonising. In fact, the call to our daemonise function didn’t seem to return. Some careful logging of the return value of fork() led us to the horrifying conclusion that fork() would return in the parent – with a PID for the child – but it would never return in the child.

At which point it clicked. The process of daemonisation under UNIX is well documented in various places. The general procedure is to fork off a child process that becomes the actual daemon, then to terminate the parent. This means that the command that starts the daemon terminates, causing the shell or other process that started it to continue, while the daemon child process continues in the background. The child process then takes steps to isolate itself from the context of the parent (setsid(), a second fork, changing directory to the root, etc).

However, this standard process has a slight bug; for whatever reason, as it was timing sensitive, we noticed it on Solaris – but it was, presumably, just as present on Linux, the BSDs, and elsewhere.

Our daemon parent process was being invoked via ssh; so as soon as it terminated, the ssh connection terminated. If you are running a process in an ssh session and you close that session from your end – by killing the ssh process, losing the network connection, etc – then the ssh daemon on the server will send a “HUP” signal to the “process group” of the shell session. HUP is short for “hangup”, from the days of modem-driven dial in sessions; it is intended to tell a process that its connection to the user using it has gone away. By default, processes spawned from your ssh shell are all in the same process group; and by default, the HUP signal kills the processes; so if your ssh connection dies, then your processes on the server are killed off. Which is all well and good.

One of the steps performed in the child process of a daemon, immediately after it has been created by fork(), is to split away from the parent process group. This is so the daemon will not be killed off if the shell that spawned it happens to be terminated by a connection being closed; as a daemon, it does not interact with the user any further, and it should be a part of the system as a whole rather than the starting user’s session. However, it cannot form a new process group until after it’s forked, as the child process needs to exist before it can be assigned to a new group. So setsid() occurs immediately after fork() has returned in the child process.

What we were seeing is that the parent process forked and then terminated, and so the ssh daemon closed the session down, and thereby sent a HUP signal to the entire process group; but this all happened before the child process had ever had a chance to run. So it didn’t get a chance to break away from the process group. It did return from the fork() call – straight into the pending HUP signal, which kills it, despite the very next line of code being a setsid() that would have saved it.

Once we realised this might be the case, we made the parent process defined a handler for HUP signals that just prints a message, before it fork()s (so the handler is inherited by the child). Sure enough, the child process often received a HUP signal before getting to setsid(), immediately upon returning from fork(). So merely ignoring the signal, rather than letting it proceed to kill the process, made our problem go away.

So it seems the standard approach to daemonisation has a minor bug; there’s a time window where the parent process has terminated, but the child is not yet entirely isolated from the parent’s process group. Therfore, the child will be killed if the session terminates during this time interval. We only saw it because we were starting a daemon from ssh, which is a rare but not unheard of thing (particularly in automatically managed clusters). We suspect that the problem may have gone unnoticed in daemons that interpret the HUP signal as a request to re-read configuration files. If such a daemon happens to have installed their HUP signal handler before forking, then a signal arriving during daemonisation would just the “reload the configuration” flag to be set, causing a spurious but harmless re-reading of the configuration as soon as the daemon starts.

So here we are, developing a distributed highly-parallel database; and we find a concurrency bug in a standard practice that has existed within Unix for ages. Concurrency bugs can arise in the least expected of places…

Cloud environments offer enticing cost savings for Web businesses, but they present some interesting challenges. One of these challenges is that clouds tend to provide large numbers of relatively lightweight virtual servers, which may potentially fail; high availability is meant to happen in your application, by designing it so that it can tolerate virtual servers coming and going.

This is great for Web servers, and with a little cleverness, it even works well for very simple key-value data stores. However, it’s not a great fit for the database layer; traditional database products rely on a small number of powerful servers, where one or more may be single points of failure, so would usually be implemented on highly redundant hardware.

Here at GenieDB, we’ve been spending the last few years busily implementing a high-level database designed to transparently operate on a cluster of unreliable small servers… So, we’ve been looking at how we can help in cloud environments.

Our existing core technology is already a perfect match; we already support zero-downtime dynamic reconfiguration of clusters (planned and unplanned), with access to the data via independent MySQL instances on each server. We provide the software to be run on each server, and provide instructions on how to seamlessly add and remove servers from a cluster without downtime.

But many users won’t have an existing cluster management infrastructure in place, and configuring each node manually is no way to run a cluster. So as well as documenting the low-level operations on nodes, we’ve written something we call Cluster Tool, which handles bulk operations on clusters. Given some software installation tarballs and a configuration file listing the servers in your cluster and how you’d like them configured, Cluster Tool will ssh into those servers, install our software, configure it, and correctly follow the processes required. It files away a copy of your configuration, so when it’s presented with a new cluster configuration (which may have added, changed, and removed servers), it can decide what servers need uninstalling, which need upgrading, which need reconfiguring and which need installing, to correctly migrate the cluster to its new state. We ship Cluster Tool with its source code, so you can use it as-is, read the source as a reference implementation alongside the low-level management documentation, modify it to fit into your existing cluster management practices, or just wrap it in your own software that feeds it new cluster configurations.

With these foundations in place, automating cloud deployments was easy. We produced Cloud Tool, which uses a pluggable backend to manage virtual servers from various cloud providers. To grow a cluster, it requests more servers, then adds them to the Cluster Tool configuration, and asks Cluster Tool to add them to the cluster. When they are no longer needed, it asks Cluster Tool to remove them from the cluster, then it requests that the servers be de-provisioned.

It’s as simple as that! And like Cluster Tool, it’s provided as source code, and only uses documented interfaces in the software its built on top of (in this case, Cluster Tool itself), so you can build upon it in your own ways, such as running hybrid clusters based on privately-owned hardware that expand out onto cloud servers during peak load.

The key idea in a replicated database is that you store multiple copies of your data, spread over several servers. The advantage is that when you want to read some data, you can pick a server (ideally, the very server your code is running on has its own local replica, so you can bypass the network), and do the reads there. This means that you can handle more reads per second by just adding more servers, and you can have copies of the data that are near to all the places you want to read data; this reduces latency, and can reduce bandwidth usage over your backbone if you have distributed sites compared to all your application servers talking to a centralised database server.

However, it has a cost: when you do a write, you need to update all the replicas. The naive approach is to require a writer to write to every replica independently; even worse is to have a “master” server that does it for them, as many popular replication systems do, since it then becomes a bottleneck. More advanced implementations implement a multicast system of some kind, ranging from actual IP multicast to diffusion over spanning trees.

But however you do it, replicating changes to a number of servers isn’t an atomic operation. When you issue that write, there will be a time delay before every server has it; and if there’s a network partition in progress, or a server is offline, that time delay could extend to days or weeks. So “synchronous replication” – where the write request does not return to the issuer as “completed” until positive acknowledgement has been received from every replica – is rare in practice; it makes writes very slow. On the other hand, “asynchronous replication” involves returning writes to the issuer as “completed” as soon as they’ve been received by a master server, or the multicast process initiated; this makes writes return quickly, improving the throughput of bulk writes tremendously. But it has a cost: programmers have to be careful.

A normal web page to allow a user to update their details will issue SQL queries to select the user’s details, then populate a form with them; if the user submits the form, we then issue an SQL query to update the user’s details, then continue as normal – fetching the data back from the database to redisplay it. This pattern avoids duplicating code, as the same logic is used to fetch the user’s details in each case; and it ensures that we will always show the user the actual data as it is in the database – so any trimming to length limits or other SQL-layer data transformations are accurately reflected.

But in a system with replication lag, this leads to problems. The user changes their details, then when they get the page back, it still shows their old details – as the change to their details had not yet replicated to the server the details were pulled back from to regenerate the page! This may not happen in testing, where the system is underloaded so replication is fast; but when the system becomes busy, replication lags can stretch to hours. The fun part is that the user may then think they did something wrong, and keep resubmitting their change until it “sticks”; this means that as the load increases, the rate at which updates come in increases by a factor proportional to how overloaded the system already is – creating a disastrous feedback loop.

Generally, applications need to be written to hide replication delays; when displaying data to a user, the application must manually account for recent changes the user has made, on top of what actually comes back from the database. This complicates development, hampering progress, and introduces plenty of scope for intermittent bugs that upset customers.

Which is why we developed proprietary technology at GenieDB to hide replication lag within the database itself, meaning that in normal usage, as soon as a write operation has returned to the application, subsequent reads (even from other servers) will return the latest information. Of course, in situations with network and server failures, this isn’t always possible, so you can still end up with some replication lag in extreme circumstances (alas, we can’t change the laws of physics), but except in emergencies, GenieDB hides replication lag from the application. We want to make it easy for applications to scale to replicated storage!

So it was with some amusement that, as we started setting up company profiles on popular professional and social network sites, we saw signs of replication lag. As staff members updated their personal profiles to state that they were part of the company, we’d go and refresh the company’s page to see if they had “done it right”, and lo, they would not appear yet.

So have we done something wrong in adding ourselves to the company? Or is it just replication lag? Should we assume the former and try again (risking breaking a working setup that just hasn’t replicated yet, and/or further loading the site’s database with unnecessary updates)? Or leave it for half an hour and see if the company page catches up?

As users of a site, we shouldn’t be having to think like this. And, of course, few users are replicated database professionals, so most of them won’t think like this; they’ll just think the site is broken if they encounter the artefacts of replication lag.