Distributed Concurrency Control

Question

I've been working on this for a few days now, and I've found several solutions but none of them incredibly simple or lightweight. The problem is basically this: We have a cluster of 10 machines, each of which is running the same software on a multithreaded ESB platform. I can deal with concurrency issues between threads on the same machine fairly easily, but what about concurrency on the same data on different machines?

Essentially the software receives requests to feed a customer's data from one business to another via web services. However, the customer may or may not exist yet on the other system. If it does not, we create it via a web service method. So it requires a sort of test-and-set, but I need a semaphore of some sort to lock out the other machines from causing race conditions. I've had situations before where a remote customer was created twice for a single local customer, which isn't really desirable.

Solutions I've toyed with conceptually are:

1. Using our fault-tolerant shared file system to create "lock" files which will be checked for by each machine depending on the customer

2. Using a special table in our database, and locking the whole table in order to do a "test-and-set" for a lock record.

3. Using Terracotta, an open source server software which assists in scaling, but uses a hub-and-spoke model.

4. Using EHCache for synchronous replication of my in-memory "locks."

I can't imagine that I'm the only person who's ever had this kind of problem. How did you solve it? Did you cook something up in-house or do you have a favorite 3rd-party product?

Answer 1

Terracotta is closer to a "tiered" model - all client applications talk to a Terracotta Server Array (and more importantly for scale they don't talk to one another). The Terracotta Server Array is capable of being clustered for both scale and availability (mirrored, for availability, and striped, for scale).

In any case as you probably know Terracotta gives you the ability to express concurrency across the cluster the same way you do in a single JVM by using POJO synchronized/wait/notify or by using any of the java.util.concurrent primitives such as ReentrantReadWriteLock, CyclicBarrier, AtomicLong, FutureTask and so on.

There are a lot of simple recipes demonstrating the use of these primitives in the Terracotta Cookbook.

As an example, I will post the ReentrantReadWriteLock example (note there is no "Terracotta" version of the lock - you just use normal Java ReentrantReadWriteLock)

import java.util.concurrent.locks.*;

public class Main
{
    public static final Main instance = new Main();
    private int counter = 0;
    private ReentrantReadWriteLock rwl = new ReentrantReadWriteLock(true);

    public void read()
    {
        while (true) {
            rwl.readLock().lock();
                try {
                System.out.println("Counter is " + counter);
            } finally {
                rwl.readLock().unlock();
            }
            try { Thread.currentThread().sleep(1000); } catch (InterruptedException ie) {  }
        }
    }

    public void write()
    {
        while (true) {
            rwl.writeLock().lock();
            try {
               counter++;
               System.out.println("Incrementing counter.  Counter is " + counter);
            } finally {
                 rwl.writeLock().unlock();
            }
            try { Thread.currentThread().sleep(3000); } catch (InterruptedException ie) {  }
        }
    }

    public static void main(String[] args)
    {
        if (args.length > 0)  {
            // args --> Writer
            instance.write();
        } else {
            // no args --> Reader
            instance.read();
        }
    }
}

Answer 2

Not sure if I understand the entire context but it sounds like you have 1 single database backing this? Why not make use of the database's locking: if creating the customer is a single INSERT then this statement alone can serve as a lock since the database will reject a second INSERT that would violate one of your constraints (e.g. the fact that the customer name is unique for example).

If the "inserting of a customer" operation is not atomic and is a batch of statements then I would introduce (or use) an initial INSERT that creates some simple basic record identifying your customer (with the necessary UNIQUEness constraints) and then do all the other inserts/updates in the same transaction. Again the database will take care of consistency and any concurrent modifications will result in one of them failing.

Answer 3

you might want to consider using Hazelcast distributed locks. Super lite and easy.

java.util.concurrent.locks.Lock lock = Hazelcast.getLock ("mymonitor");
lock.lock ();
try {
// do your stuff
}finally {
   lock.unlock();
}

Hazelcast - Distributed Queue, Map, Set, List, Lock

Answer 4

We use Terracotta, so I would like to vote for that.

I've been following Hazelcast and it looks like another promising technology, but can't vote for it since I've not used it, and knowing that it uses a P2P based system at its heard, I really would not trust it for large scaling needs.

But I have also heard of Zookeeper, which came out of Yahoo, and is moving under the Hadoop umbrella. If you're adventurous trying out some new technology this really has lots of promise since it's very lean and mean, focusing on just coordination. I like the vision and promise, though it might be too green still.

• http://www.terracotta.org
• http://wiki.apache.org/hadoop/ZooKeeper
• http://www.hazelcast.com

Answer 5

You might also consider Cacheonix for distributed locks. Unlike anything else mentioned here Cacheonix support ReadWrite locks with lock escalation from read to write when needed:

ReadWriteLock rwLock = Cacheonix.getInstance().getCluster().getReadWriteLock();
Lock lock = rwLock.getWriteLock();
try {
  ...
} finally {
  lock.unlock();
}

Full disclosure: I am a Cacheonix developer.

Answer 6

If you can set up your load balancing so that requests for a single customer always get mapped to the same server then you can handle this via local synchronization. For example, take your customer ID mod 10 to find which of the 10 nodes to use.

Even if you don't want to do this in the general case your nodes could proxy to each other for this specific type of request.

Assuming your users are uniform enough (i.e. if you have a ton of them) that you don't expect hot spots to pop up where one node gets overloaded, this should still scale pretty well.