The problem
ConcurrentHashMap provides very weak consistency guarantees w.r.t iteration:
guaranteed to traverse elements as they existed upon construction exactly once, and may (but are not guaranteed to) reflect any modifications subsequent to construction
(note the "may" part).
In my case, I have a concurrent map that I need to periodically back-up. It's very important that what I back-up be a consistent point-in-time representation of the map. Back-ups are few and far between, triggered on a schedule and never run at the same time. I ended up implementing my own "map" (only the methods I use, not a full-blown map impl) based on 2 underlying concurrent maps and a RW lock:
notes
- the map is expected to be big. VERY big. so probably no copying it. also - there's absolutely no guarantee that a copy-constructor call will land you a consistent copy - it uses iteration under the hood.
Holder.java (used to return values out of closures/inner classes)
public class Holder<T> {
private T value;
private boolean isEmpty = true;
public T getValue() {
return value;
}
public void setValue(T value) {
this.value = value;
isEmpty = false;
}
public boolean isEmpty() {
return isEmpty;
}
}
AutoCloseableLock.java (abuses AutoCloseable for locking, which I think makes for cleaner code)
import java.util.concurrent.locks.Lock;
public class AutoCloseableLock implements AutoCloseable{
private final Lock delegate;
public AutoCloseableLock(Lock delegate) {
this.delegate = delegate;
}
public AutoCloseableLock lock() {
delegate.lock();
return this;
}
@Override
public void close() {
delegate.unlock();
}
}
SingleSnapshotMap.java - a (partial) map implementation to allows a single consistent snapshot
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.BiConsumer;
public class SingleSnapshotMap<K,V>{
private ConcurrentHashMap<K,V> baseMap = new ConcurrentHashMap<>();
private ConcurrentHashMap<K,V> diffMap;
private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
private AutoCloseableLock readLock = new AutoCloseableLock(readWriteLock.readLock());
private AutoCloseableLock writeLock = new AutoCloseableLock(readWriteLock.readLock());
private final Object DELETION_MARKER = new Object();
public V put(K key, V value) {
try (AutoCloseableLock ignored = readLock.lock()){
if (diffMap != null) {
final Holder<V> prevValueHolder = new Holder<>();
diffMap.compute(key, (k,v) -> {
if (v == null) { //no previous mapping in diff. check in base
prevValueHolder.setValue(baseMap.get(key));
} else if (v == DELETION_MARKER) { //was marked as deleted. means prev==null
prevValueHolder.setValue(null);
} else {
prevValueHolder.setValue(v);
}
return value; //new value is arg either way
});
return prevValueHolder.getValue();
} else {
return baseMap.put(key, value);
}
}
}
public V get(K key) {
try (AutoCloseableLock ignored = readLock.lock()){
if (diffMap != null) {
final Holder<V> valueHolder = new Holder<>();
diffMap.compute(key, (k,v) -> {
if (v == null) { //no value in diff. check base
valueHolder.setValue(baseMap.get(key));
} else if (v == DELETION_MARKER) { //was marked as deleted. return null
valueHolder.setValue(null);
} else { //got a value
valueHolder.setValue(v);
}
return v; //do not change the current mapping
});
return valueHolder.getValue();
} else {
return baseMap.get(key);
}
}
}
public V remove(K key) {
try (AutoCloseableLock ignored = readLock.lock()){
if (diffMap != null) {
final Holder<V> prevValueHolder = new Holder<>();
((ConcurrentHashMap)diffMap).compute(key, (k,v) -> {
if (v == null) {
prevValueHolder.setValue(baseMap.get(key));
} else if (v == DELETION_MARKER) {
prevValueHolder.setValue(null);
} else {
prevValueHolder.setValue((V) v);
}
return DELETION_MARKER;
});
return prevValueHolder.getValue();
}
return baseMap.remove(key);
}
}
private void startSnapshot() {
try (AutoCloseableLock ignored = writeLock.lock()){
if (diffMap != null) {
throw new IllegalStateException("only a single snapshot at a time");
}
diffMap = new ConcurrentHashMap<>();
}
}
private void endSnapshot() {
try (AutoCloseableLock ignored = writeLock.lock()){
if (diffMap == null) {
throw new IllegalStateException("no snapshot active");
}
//nothing else active. "flush" diff back into base
for (Map.Entry<K,V> e : diffMap.entrySet()) {
if (e.getValue() == DELETION_MARKER) {
baseMap.remove(e.getKey());
} else {
baseMap.put(e.getKey(), e.getValue());
}
}
diffMap = null;
}
}
public void snapshot(BiConsumer<? super K, ? super V> action) {
startSnapshot();
try {
baseMap.forEach(action);
} finally {
endSnapshot();
}
}
}
Expected usage
SingleSnapshotMap<Long, String> map = new SingleSnapshotMap<>();
//place stuff into map
Map<Long, String> copyMap = new HashMap<>();
map.snapshot((aLong, s) -> {
//stream key-value pair to disk somewhere
}); //meanwhile activity goes on in the background
//copyMap now holds a consistent point-in-time copy of map
Things I'm concerned about
- Correctness - above all. I have some tests around this class (horrible nightmare code full of locks threads sleeps and yields), but MT code is tricky.
- Elegance - if there's a library that does this that I've missed, or a simpler solution.
- Performance and concurrency - the process running inside the snapshot() method could be long. I want to continue map operations while its running in the background.
Things I'm already aware of
- Java 8 has
StampedLockwhich performs better thanReadWriteLock. I do plan on switching.
