GSS/Pithos update: iPhone, Android, Mongo & getting to 2.0

It has been a while since my last post on GSS/Pithos, and there have been quite a few important announcements that I have not blogged about, only tweeted:

• a shiny native iPhone client
• a brand new Android client
• a back-office application for statistics and admin tasks
• public folders for hosting static web sites
• a slew of performance, usability and bug fixes for the web & desktop clients
• new user registration and quota upgrade workflows

And since the code base has been adopted by other European NRNs, we have also improved our release engineering process and documentation. This however was just the tip of the iceberg. What we are now close to completing, is nothing short of an almost complete rewrite of the core GSS server, aiming at even greater performance and scalability. We are about to release GSS 2.0.

The initial plan was far more modest. We had identified the PostgreSQL installation that stores all of the the system’s data besides the actual files, as a future bottleneck when user adoption would start to accelerate. In such an event there were a few low-to-medium-cost solutions that we could pursue, like replication and manual sharding, but since the problem seemed best suited for a NoSQL solution, we decided to bite the bullet and attempt a transition.

Our field search included lots of products, like Mongo, Couch, Cassandra, Riak, Redis, HBase, MemcacheDB and others. The basic requirements were high write throughput, sharding, replication and easy migration. Most of the products boasted high performance for writes and lots supported replication. In the end we discarded the ones that didn’t support sharding, which limited the options a lot, and we investigated the various APIs, trying to figure out what it would mean to rewrite GSS for one of these. In the end we chose Mongo for its rich query support, since it was the best fit for our problem at hand, making the transition of the existing code base easier. Although we’d love to have had the opportunity to try different ports for each of those datastores, it is most likely that we would have had to rewrite the entire server from scratch, which was not an option.

The initial plan was to try to isolate the changes to the data access layer, essentially building a new adapter for the middle tier that would appear almost identical to the existing code. This effort was fruitless however, since we could not find satisfactory solutions for various issues, like transactions, data mapping, entity lifecycle, etc. We came to the conclusion that even if we managed to build a not-too-leaky abstraction for Mongo, it would have probably killed the performance, making the switch from PostgreSQL pointless.

So we ditched JPA and container-managed transactions and decided to make a simpler POJO mapping to Mongo using Morphia, a Java data mapper for Mongo that takes care of the nitty-gritty details and repetitive coding, for only a minor performance hit. Since Mongo only supports atomicity for a single query, we reverted to manual transaction handling and decided to ditch the EJBs that formed the backbone of GSS altogether, since they didn’t buy us anything at that point. The new GSS server would be a set of servlets, service and domain objects, wired together via Google Guice. Since we could host the new server in a plain servlet container, we went from JBoss to Jetty, which vastly reduced the server startup time from 60 seconds down to 2. As a consequence, development round-trip times went down, and coding was fast again. Lots of object copying and conversions from entities to DTOs and back again were now unnecessary, since there was no more a transaction boundary at the session bean layer. This kept memory usage low, increased cache locality and boosted performance. There are lots of interesting details about our approach, but they would be better discussed in separate blog posts. If you are interested, you can already check out the code in the gss2 branch of the GSS repository.

The next generation GSS server is not production ready yet, but we'll be getting there soon, and so far it has been a great ride!

Retrying transactions with exponential backoff

The method I described in my previous post about retrying transactions could use some improvement. A deficiency that will only become relevant in highly congested servers is the constant retry interval. When two or more transactions try to commit at the same time and fail, with the code from the last post they will retry probably simultaneously again. Random runtime events (like process/thread scheduler decisions, JIT compiler invocations, etc.) might help avoid collisions, but in general the collided transactions may well collide again. And again. And then again, until the specified maximum number of retries is reached.

The general method to alleviate such problems is to randomize the retry intervals. The most well-known algorithm in this category is called exponential backoff. This is one way to implement it for the utility method in my previous post (full code in gss):

public T tryExecute(final Callable<T> command) throws Exception {
  T returnValue = null;
  // Schedule a Future task to call the command after delay milliseconds.
  int delay = 0;
  ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
  for (int i = 0; i < TRANSACTION_RETRIES; i++) {
    final int retry = i;
    ScheduledFuture<T> future = executor.schedule(new Callable<T>() {

      @Override
      public T call() throws Exception {
        return command.call();
      }
    }, delay, TimeUnit.MILLISECONDS);

    try {
      returnValue = future.get();
      break;
    } catch (ExecutionException e) {
      Throwable cause = e.getCause();
      if (!(cause instanceof EJBTransactionRolledbackException) ||
          retry == TRANSACTION_RETRIES - 1) {
        executor.shutdownNow();
        throw new Exception(cause);
      }
      delay = MIN_TIMEOUT + (int) (MIN_TIMEOUT * Math.random() * (i + 1));
      String origCause = cause.getCause() == null ?
          cause.getClass().getName() :
          cause.getCause().getClass().getName();
      logger.info("Transaction retry #" + (i+1) + " scheduled in " + delay +
          " msec due to " + origCause);
    }

  }
  executor.shutdownNow();
  return returnValue;
}

The notable changes in this version are the delayed invocations of the Callable and the retry interval calculation. The former is accomplished using a ScheduledFuture from the excellent utilities in java.util.concurrent, which gets executed after a random time period. The interval calculation is something that can be implemented in a number of ways, either with monotonically increasing intervals or not. I opted for the formula above since it provided the fastest collision elimination in my tests, much faster than the monotonically increasing interval formulas I tried. The MIN_TIMEOUT constant is more of a black art. It should be tuned to the particular hardware and network setup in order to attain maximum efficiency: the minimum number of retries with the minimum interval between retries.

Another issue raised in a comment in my previous post was that the exception I am using as guard, EJBTransactionRolledbackException, may be too generic for this purpose. This is definitely true, EJBTransactionRolledbackException just wraps an OptimisticLockException as one would expect for this case, which in turn wraps the Hibernate-specific StaleObjectStateException that is thrown initially. However, contention in a database table, index or row might not necessarily result in optimistic locking violations, but in deadlocks as well, when one transaction holds exclusive locks on resources that are needed by the other and vice versa. Deadlock detection is performed by the DBMS, which forces a rollback of both transactions. This time however the initial exception (at least in my testing, might be DBMS or JDBC driver-specific) is GenericJDBCException which gets wrapped in a PersistenceException, which in turn is put inside a EJBTransactionRolledbackException before being received by our call site. Therefore, the generic nature of the guard is not a problem in this case. On the contrary it covers more relevant cases and one might argue even that there is no better service that can be offered to the caller, than retrying in all such occasions.

Retrying transactions in Java

When building a web-based system, the most often encountered persistence solution is the use of a relational DBMS. This provides many benefits, but probably the most important one is that its behavior and usage are widely understood after so many decades of use. The simplicity of the ACID transactional model, coupled with the nice fit of the HTTP request/response protocol, make for a killer combination.

The simplicity starts to break down however, when the load on the DBMS server increases, causing transactions to start failing. The DBMS can guarantee that no data corruption will occur, but handling such cases is not that simple. The scalable Optimistic Locking model requires transactions to be resubmitted when failure occurs. Since this is supposed to be rather rare (otherwise optimistic locking should probably not be used), developers often cheat by not dealing with such errors, letting them bubble up to the user, who is then responsible for resubmitting the transaction. This leads to simple, elegant codebases and unhappy users.

Adding ORM to the mix only makes things worse. In JavaEE, the JPA API hides much of the complexity of dealing with relational databases, but corner cases become harder. Add the container-managed transaction boundaries provided by session EJBs and one might be tempted to raise his hands up in despair.

However a solution is not as hard as it sounds. Provided that session EJB methods are sufficiently free from side effects, one can replay a transaction for a number of times from the call site (external to the session bean) with a small amount of code and a small added complexity to the call sites. Since I seem to come up against this requirement quite often, I thought I should describe one way to do it. The following tryExecute() helper method, retries any action supplied in a standard Callable interface for a number of times, in the face of optimistic locking violations:

public T tryExecute(Callable<T> command) throws Exception {
  T returnValue = null;
  for (int i = 0; i < TRANSACTION_RETRIES; i++) {
    try {
      returnValue = command.call();
      break;
    } catch(EJBTransactionRolledbackException trbe) {
      if (i == TRANSACTION_RETRIES - 1)
        throw trbe;
      } catch (Exception e) {
        throw e;
      }
      logger.debug("Transaction retry: " + (i+1));
    }
  return returnValue;
}

By abstracting away the session bean method code in a parameterized class we can reuse this method in every place that needs to retry transactions. Using a Callable as a parameter lets us return any values received from the session bean. For supplying bean methods that do not return values, we could create Runnable instances and wrap them to Callable using Executors.callable(). You can see the original code here.

If you use the above code in a JavaSE environment without EJB, you should probably need to catch OptimisticLockException for JPA, or even StaleObjectStateException, if you don't mind depending on a Hibernate API. I use the above helper method like this in my code:

file = new TransactionHelper<FileHeaderDTO>().tryExecute(new Callable<FileHeaderDTO>() {
  @Override
  public FileHeaderDTO call() throws Exception {
    return sessionBean.createFile(user.getId(), folder.getId(), name, mimeType, uploadedFile);
  }
});

Java's verbosity makes this not that easy to understand at first sight, but what happens is very simple indeed: I wrap a one-liner call to the session bean method createFile(), in a Callable and supply it to tryExecute(). The result from the bean method is received in a local variable as usual. One thing to keep in mind though is that since Java doesn't have closures, the above anonymous function can only receive final variables as parameters. In my example user, folder, name, mimeType and uploadedFile are all final. You might have to resort to using temporary local variables for storing final values on some occasions.

Heresy! Web apps without SQL databases

Apple's Jens Alfke posted a thought-provoking piece the other day about data storage in the web world. In it he questions the conventional wisdom that SQL databases are the one and only solution for that problem space. While discussing the scalability problems faced by twitter.com (built on Ruby on Rails), he ponders over using plain files as a storage format and playing some clever tricks to get more performance in a simple way. The comments on his post are also interesting and the debate is still going.

Although I've been working for seven years in web applications with SQL databases, that's not the first time I've heard of this heretic opinion. As one of his readers accurately points out, Paul Graham's Viaweb (now Yahoo! Store) was (still is?) also using flat files to store its data in a FreeBSD UFS file system. Many large-scale server applications (say, Directory Servers) use storage managers like Berkeley DB to keep their data, without having the SQL query engine or the relational model to depend upon. Also, very highly transactional applications in the financial banking sector frequently use distributed in-memory object caches like Tangosol's (now Oracle's) Coherence as an ultra fast data store, with an RDBMS as a backup.

There is a lesson to be learned here, and it is finely articulated by Poul-Henning Kamp's presentation on Varnish, a web accelerator. Poul-Henning, a veteran FreeBSD kernel developer, created a Squid-killer by, to use his own words, not fighting with the operating system. The architecture of his web proxy is closely aligned with the way a contemporary UNIX-like operating system works, thus avoiding pointless layers of abstraction and architectural mismatch. The same logic accounts for Coherence's performance gains, a good mapping of the data storage architecture (in-memory object maps) to the web application architecture (Java EE or a lightweight alternative). The same way Varnish avoids traditional file I/O by using the OS's virtual memory (directly mapping files into pages via mmap), Coherence avoids the database overhead (SQL query optimizers, table indexes, disk access, etc.) by always dealing with objects stored in memory.

If for a particular web-based application problem we can devise a non-SQL storage solution that maps well to its use case requirements, then why suffer the (financial, support, technical) overhead of a SQL database? If there are no requirements of OLAP functionality or data warehousing, why bother? Remember the KISS principle: Keep It Simple Stupid! Who would knowingly subject himself to the evils of the object-relational mismatch (the Vietnam of Computer Science), if there was another way to do it?

Admittedly, the great thing about SQL databases is that they let you query your data in ways that you may have not contemplated as necessary. In a system with constantly changing requirements for data view (or with requirements for ad-hoc, dynamic viewing) you may well not have any other choice, but to use a SQL DBMS. Perhaps Object-oriented DBMSs can be a solution too, but I believe they are just starting to obtain OLAP and data warehousing capabilities and the performance may not be quite there yet.

Can it be an accident that Google uses Bigtable, a distributed storage system of their own devising for internal use? When distributed/federated databases have major drawbacks (namely cost and vendor lock-in), where do you go when you have to scale? Horizontal partitioning is a solution, but you'd better plan ahead.

In such circumstances, it may be appropriate to consider even heretic solutions.


Update: For those who won't take the time to watch the whole Varnish presentation (hey, it's fun, really!) Kostas Kalevras reminds me that you can find the meat of the presentation in the architect's notes. Apart from Poul-Henning's witty comments, they contain pretty much all the important points made there.