Max De Marzi

A while ago, I showed you a way to scale Neo4j writes using RabbitMQ. Which was kinda cool, but some of you asked me for a different solution that didn’t involve adding yet another software component to the stack.

Turns out we can do this in just Neo4j using a little help from the Guava library. The solution involved a background service running that holds the writes in a queue, and every once in a while (like say every second) commits those writes in one transaction.

There are many different ways to get data into Neo4j, this solution is really meant for when you don’t know what the data will be ahead of time. For example if you are building a consumer facing application and you have 10k users logged in concurrently all performing actions. In this example we are going to build something very generic. Let’s say you are keeping track of web sites users have visited, so you can do further analysis on that data.

The service will take a userId, a website url and create a unique relationship between them. The users and the urls will be unique nodes as well, so the “writes” are pretty heavy (we’re not just creating an empty node). We’ll make two versions of our writes. A synchronous version that executes the write immediately in a transaction and an asynchronous version which will delegate the writes to a background service. An AbstractScheduledService to be precise. The service will hold a queue for our writes:

public class BatchWriterService extends AbstractScheduledService{
    public LinkedBlockingQueue<HashMap<String, Object>> queue = new LinkedBlockingQueue<>();

A scheduler tells it when to execute, in our case, we’ll have it run every second:

@Override
    protected Scheduler scheduler() {
        return Scheduler.newFixedRateSchedule(0, 1, TimeUnit.SECONDS);
    }

The runOneIteration method will drain the queue, and perform the writes.

    @Override
    protected void runOneIteration() throws Exception {
    ...

I won’t go over how the graph writer service works in this blog post, instead I want to show you what it does. The full source code is your reward for reading the whole thing (or cheating and scrolling to the bottom). I always start with the hello world end point to make sure it’s wired up correctly:

I always include a warm-up routine in my unmanaged extensions so when I’m testing I can quickly get things into memory or in production get a server ready before adding it to the load balancer.

We’ve made an end point for initialization that basically creates two unique index constraints. One for users and one for sites. We’ll run this just once.

Now let’s try adding some data. In this case I will tell it that user “max” visited the site http://www.neo4j.org :

Let’s take a look at what our database has now:

We can see it created two nodes, and a relationship between them. Next we’ll try the async version going to a different site http://www.maxdemarzi.com :

Let’s take a look at our data again:

It did not create another “max” node, instead it used the one we had. In both cases Neo4j created our data. Let’s take a look at console.log to see what our async process did.

Performed a set of transactions with 1 writes in 50 [msec] @ 2014-06-30T16:07:25.460-05:00

Here we see that our batch writer service received a write and performed it for us in the background. Awesome! We have the simple case working… but let’s try this with some load and compare it’s performance. First we’ll try running the synchronous version which will create a transaction for every single write. We’ll use Gatling for this performance test, the source code for the test is on github.

The test ran for 30 seconds and it managed to create 2815 writes in a pretty crappy 93 req/s with a mean of 170ms and a max of 510ms. What’s worse is that we even got an error. Taking a look at the logs we see that two transactions were trying to update the same node at the same time.

16:19:56.813 [qtp6401337-74] WARN o.e.jetty.servlet.ServletHandler – /v1/service/4234497/visited
org.neo4j.kernel.DeadlockDetectedException: Don’t panic.

A deadlock scenario has been detected and avoided. This means that two or more transactions, which were holding locks, were wanting to await locks held by one another, which would have resulted in a deadlock between these transactions. This exception was thrown instead of ending up in that deadlock.

See the deadlock section in the Neo4j manual for how to avoid this: http://docs.neo4j.org/chunked/stable/transactions-deadlocks.html

Details: ‘Transaction(2149, owner:”qtp6401337-74″)[STATUS_ACTIVE,Resources=2] can’t wait on resource RWLock[Node[3272]] since => Transaction(2149, owner:”qtp6401337-74″)[STATUS_ACTIVE,Resources=2] <-[:HELD_BY]- RWLock[IndexLock[Site:url]] <-[:WAITING_FOR]- Transaction(2148, owner:"qtp6401337-79")[STATUS_ACTIVE,Resources=2] <-[:HELD_BY]- RWLock[Node[3272]]'.

Let’s try the async version:

Wow look at that! No errors since it only has a single writer, so it cannot deadlock. Now we’re going at just under 6k req/s that seems pretty amazing compared to under 100 r/s! 1ms mean and a max of 290ms, but let’s see what is going on under the covers before we claim victory:

Performed a set of transactions with 566 writes in 3366 [msec] @ 2014-06-30T19:46:29.661-05:00
Performed a transaction of 1,000 writes in 9678 [msec] @ 2014-06-30T19:46:39.340-05:00
Performed a transaction of 1,000 writes in 5957 [msec] @ 2014-06-30T19:46:45.298-05:00
Performed a transaction of 1,000 writes in 4270 [msec] @ 2014-06-30T19:46:49.569-05:00
Performed a transaction of 1,000 writes in 2845 [msec] @ 2014-06-30T19:46:52.415-05:00
Performed a transaction of 1,000 writes in 2599 [msec] @ 2014-06-30T19:46:55.015-05:00
Performed a set of transactions with 5000 writes in 25352 [msec] @ 2014-06-30T19:46:55.015-05:00
Performed a transaction of 1,000 writes in 1218 [msec] @ 2014-06-30T19:46:56.234-05:00
Performed a transaction of 1,000 writes in 622 [msec] @ 2014-06-30T19:46:56.857-05:00
Performed a transaction of 1,000 writes in 962 [msec] @ 2014-06-30T19:46:57.820-05:00
Performed a transaction of 1,000 writes in 682 [msec] @ 2014-06-30T19:46:58.503-05:00
Performed a transaction of 1,000 writes in 654 [msec] @ 2014-06-30T19:46:59.157-05:00
Performed a set of transactions with 5000 writes in 4142 [msec] @ 2014-06-30T19:46:59.158-05:00
Performed a transaction of 1,000 writes in 486 [msec] @ 2014-06-30T19:46:59.644-05:00
Performed a transaction of 1,000 writes in 421 [msec] @ 2014-06-30T19:47:00.066-05:00
Performed a transaction of 1,000 writes in 374 [msec] @ 2014-06-30T19:47:00.441-05:00
Performed a transaction of 1,000 writes in 416 [msec] @ 2014-06-30T19:47:00.858-05:00
Performed a transaction of 1,000 writes in 461 [msec] @ 2014-06-30T19:47:01.320-05:00
Performed a set of transactions with 5000 writes in 2162 [msec] @ 2014-06-30T19:47:01.320-05:00
Performed a transaction of 1,000 writes in 419 [msec] @ 2014-06-30T19:47:01.740-05:00
Performed a transaction of 1,000 writes in 481 [msec] @ 2014-06-30T19:47:02.222-05:00
Performed a transaction of 1,000 writes in 365 [msec] @ 2014-06-30T19:47:02.588-05:00
Performed a transaction of 1,000 writes in 286 [msec] @ 2014-06-30T19:47:02.875-05:00
Performed a transaction of 1,000 writes in 320 [msec] @ 2014-06-30T19:47:03.195-05:00
…
Performed a set of transactions with 5000 writes in 1693 [msec] @ 2014-06-30T19:47:57.424-05:00
Performed a transaction of 1,000 writes in 365 [msec] @ 2014-06-30T19:47:57.789-05:00
Performed a transaction of 1,000 writes in 261 [msec] @ 2014-06-30T19:47:58.051-05:00
Performed a transaction of 1,000 writes in 278 [msec] @ 2014-06-30T19:47:58.330-05:00
Performed a transaction of 1,000 writes in 293 [msec] @ 2014-06-30T19:47:58.623-05:00
Performed a set of transactions with 4335 writes in 1281 [msec] @ 2014-06-30T19:47:58.706-05:00

So what’s happening here? Looks like it takes a while to warm up, but then settles in at 1000 writes every 300ms or so. We can also see the writes finished at 19:47:58, but our tests finished at 19:46:55. That’s a whole minute behind. So what’s happening is we built a system that will keep on accepting requests until it eventually crashes… and when it does we’ll lose a minute, an hour, a day, who knows how much data!

We don’t want that to happen! So let’s write a little back pressure in to the system. We’ll force our queue to have a capacity of 25000 writes. Any requests that come in while the queue is full will block until the queue is drained.

That brings us down to around 1900 r/s, but our test ended at 20:23:40 and our writes finished 8 seconds later at 20:23:48. Our maximum loss is the size of the queue, and we can be at most “however long it takes one queue size set of writes” behind. I didn’t spend much time tuning it for the optimal numbers because that’s going to be hardware dependent. The numbers are much better than single transactions, but will get even better once the system has data as it would be in production. Instead of creating 2 new unique nodes (and their corresponding index entries) as well as a unique relationship between them, we’ll be just updating a relationship. Here is what those numbers look like if you are curious:

You can generate similar numbers by running the test multiple times. Take a look at the log, you’ll see you’ll be at most a second or two behind. The source code for the batch writer service is on github. I’ll be going over this in more detail on my Neo4j High Performance Video Course.