September 13, 2017
When we update rows in our database we use transactions. By using transactions we can safely modify our data even when multiple clients are accessing the database concurrently. In java, we have many techniques for modifying data concurrently, volatile fields, synchronized blocks, classes from java.util.concurrent.
But we are missing a high-level abstraction like transactions in a database to concurrently modify data in a safe way. I think atomic methods are a good fit for such a high-level abstraction.
A method is atomic if it is "all or nothing". If another thread reads the data the other thread can only see the state before or after the execution of the atomic block, no intermediate state. After the atomic method was successful the changes are visible to all other threads. The atomic method only modifies data of its own object without side effects.
Here is an example of a class with atomic methods implemented by the synchronized statement:
vmlens
public class AtomicPositiveValue {
private int value;
public AtomicPositiveValue(int newValue) throws Exception
{
if( newValue < 0 )
{
throw new Exception("value is negative");
}
this.value = newValue;
}
public synchronized int get()
{
return value;
}
public synchronized void set(int newValue) throws Exception
{
if( newValue < 0 )
{
throw new Exception("value is negative");
}
value = newValue;
}
}
Let us modify an instance of our class from multiple threads. The main advantage of atomic methods is that we can simulate this by a single threaded method as the following.
public void testSetAndGetParallel() throws Exception
{
AtomicPositiveValue atomicPositiveValue= new AtomicPositiveValue(0);
int threadA = atomicPositiveValue.get();
int threadB = atomicPositiveValue.get();
atomicPositiveValue.set(threadA + 5);
atomicPositiveValue.set(threadB + 5);
assertEquals( atomicPositiveValue.get() , 5 );
}
public void testSetAndGetSequentiell() throws Exception
{
AtomicPositiveValue atomicPositiveValue= new AtomicPositiveValue(0);
int threadA = atomicPositiveValue.get();
atomicPositiveValue.set(threadA + 5);
int threadB = atomicPositiveValue.get();
atomicPositiveValue.set(threadB + 5);
assertEquals( atomicPositiveValue.get() , 10 );
}
This is not the result we wanted. The chaining of the get and set method leads to a non atomic update and the result depends on the order of the set and get calls from the different threads. We need an atomic update method:
public synchronized int update(int delta) throws Exception
{
int temp = value + delta;
if( temp < 0 )
{
throw new Exception("value is negative");
}
value = temp;
return value;
}
Now we always achieve the same result:
public void testUpdate() throws Exception
{
AtomicPositiveValue atomicPositiveValue= new AtomicPositiveValue(0);
atomicPositiveValue.update(5); // Thread A
atomicPositiveValue.update(5); // Thread B
assertEquals( atomicPositiveValue.get() , 10 );
}
As we have seen chaining atomic methods from the same object typically leads to nonatomic methods. Therefore we need to provide atomic methods for all use cases.
Now let us see what happens when we compose atomic methods. For example let us create a method which transfers an amount from one instance to another.
public synchronized void transfer(AtomicPositiveValue other, int amount) throws Exception
{
other.update( -1 * amount );
update(amount);
}
To test this method we need a multi-threaded unit test. The ConcurrentTestRunner runs each test method parallel by 4 threads.
@RunWith(ConcurrentTestRunner.class)
public class TestDeadlockAtomicValue {
private final AtomicPositiveValue first;
private final AtomicPositiveValue second;
public TestDeadlockAtomicValue() throws Exception
{
first = new AtomicPositiveValue(1000);
second = new AtomicPositiveValue(1000);
}
@Test
public void testTransferFirstToSecond() throws Exception
{
second.transfer( first , 1);
}
@Test
public void testTransferSecondToFirst() throws Exception
{
first.transfer( second , 1);
}
}
If we run the test we see deadlocks:
- deadlock: Monitor@com.anarsoft.vmlens.concurrent.example.AtomicPositiveValue.update()<->Monitor@com.anarsoft.vmlens.concurrent.example.AtomicPositiveValue.update()
parent2Child:
thread: Thread-4
stack:
- com.anarsoft.vmlens.concurrent.example.AtomicPositiveValue.update <<Monitor@com.anarsoft.vmlens.concurrent.example.AtomicPositiveValue.update()>>
- com.anarsoft.vmlens.concurrent.example.AtomicPositiveValue.transfer <<Monitor@com.anarsoft.vmlens.concurrent.example.AtomicPositiveValue.update()>>
- com.anarsoft.vmlens.concurrent.example.TestDeadlockAtomicValue.testTransferSecondToFirst
- sun.reflect.NativeMethodAccessorImpl.invoke
- sun.reflect.DelegatingMethodAccessorImpl.invoke
- java.lang.reflect.Method.invoke
- org.junit.runners.model.FrameworkMethod$1.runReflectiveCall
- org.junit.internal.runners.model.ReflectiveCallable.run
- org.junit.runners.model.FrameworkMethod.invokeExplosively
- org.junit.internal.runners.statements.InvokeMethod.evaluate
- com.anarsoft.vmlens.concurrent.junit.internal.ConcurrentStatement.evaluateStatement
- com.anarsoft.vmlens.concurrent.junit.internal.ConcurrentStatement.evaluate
- com.anarsoft.vmlens.concurrent.junit.internal.ParallelExecutorThread.run
child2Parent:
thread: Thread-1
stack:
- com.anarsoft.vmlens.concurrent.example.AtomicPositiveValue.update <<Monitor@com.anarsoft.vmlens.concurrent.example.AtomicPositiveValue.update()>>
- com.anarsoft.vmlens.concurrent.example.AtomicPositiveValue.transfer <<Monitor@com.anarsoft.vmlens.concurrent.example.AtomicPositiveValue.update()>>
- com.anarsoft.vmlens.concurrent.example.TestDeadlockAtomicValue.testTransferFirstToSecond
- sun.reflect.NativeMethodAccessorImpl.invoke
- sun.reflect.DelegatingMethodAccessorImpl.invoke
- java.lang.reflect.Method.invoke
- org.junit.runners.model.FrameworkMethod$1.runReflectiveCall
- org.junit.internal.runners.model.ReflectiveCallable.run
- org.junit.runners.model.FrameworkMethod.invokeExplosively
- org.junit.internal.runners.statements.InvokeMethod.evaluate
- com.anarsoft.vmlens.concurrent.junit.internal.ConcurrentStatement.evaluateStatement
- com.anarsoft.vmlens.concurrent.junit.internal.ConcurrentStatement.evaluate
- com.anarsoft.vmlens.concurrent.junit.internal.ParallelExecutorThread.run
The trace was generated by vmlens.
Chaining of atomic methods leads to deadlock. At least when they are implemented by synchronized statements. And since how the atomic method is implemented should by hidden, we need to avoid chaining of atomic methods.
To see that we can easily change the implementation of our AtomicPositiveValue, let us see how it can be implemented with compareAndSet. Suppose in our performance test we see a bottleneck at the get method. And we decide to use the following faster implementation using AtomicInteger with compareAndSet:
public class AtomicPositiveValueUsingAtomicInteger {
private final AtomicInteger value;
public AtomicPositiveValueUsingAtomicInteger(int newValue) throws Exception {
if (newValue < 0) {
throw new Exception("value is negative");
}
value = new AtomicInteger(newValue);
}
public int get() {
return value.get();
}
public int update(int delta) throws Exception {
int current = value.get();
int update = current + delta;
if (update < 0) {
throw new Exception("value is negative");
}
while (!value.compareAndSet(current, update)) {
update = current + delta;
if (update < 0) {
throw new Exception("value negative");
}
}
return update;
}
}
The behaviour of our class is the same as the synchronized implementation. It is only faster.
Atomic methods let us use classes in a thread safe way without knowing the implementation details, similar to database transactions. If we want to test if our usage is correct we can simply chain the method calls. To test a specific thread interleaving we can simply order the calls accordingly. In contrast to database transactions, which have automatic deadlock detection, we can not chain atomic methods.
© 2020 vmlens Legal Notice Privacy Policy