10
\$\begingroup\$

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 StampedLock which performs better than ReadWriteLock. I do plan on switching.
\$\endgroup\$
3
\$\begingroup\$

The first answer I gave was based on the premise that the data could be 'cloned' out of the HashMap. The alternative way for processing the data, as suggested, is a form of serializing the data away to a slow target (disk, network, etc.). That serialization cannot happen while holding a lock on the source.

The current implementation accomplishes this by 'freezing' the underlying datastore, and then dumping that datastore to the output. To keep the system available, it also creates a 'hold and store' mechanism that tracks changes made, and then, when the serialization is complete, it 'replays' the changes to the underlying store.

The problems with that system are numerous:

  1. in the normal course of events, every get, remove, and put requires a 'readlock' to be held for that operation. This significantly reduces the amount of concurrency available for the store. Actually, now that I look at it, the implementation uses a ReadWriteLock, but there is a bug, and even the 'write-lock' is a readLock(), so there is no effective exclusion from the process. If the lock was implemented right though, it would still require a full lock when the 'replay' was performed.

  2. even when the system is not having a snapshot taken, the overhead is required to conditionally manage the data load. Using a Strategy Pattern we can improve that, by having one simple strategy that is used most of the time, and then a more complicated strategy that is only used when performing snapshots.

  3. Using the compute mechanism of ConcurrentHashMap is overly complicated, and has resulted in the degradation of generic typing to raw types, and is a problem.

I believe the overall strategy of managing a 'diff' concept is, in the long run, the right strategy. Additionally, the concept of the Holder is good too.

The problem with the global read-locks, and alternatively, the blocking write-lock while the data is flushed, is hard to overcome without introducing more granular locking.

I have taken the liberty of re-implementing your code with some alternate schemes. Note that there is no global Lock mechanism. There is a single core AtomicReference which contains the current 'strategy'. The simple PassThrough strategy has almost no overhead, and will have no performance impact on the general use case.

The more complicated LoggedThrough class extends the PassThrough strategy, but it logs all operations going though, and does not pass the values back, unless the recording is complete (the snapshot done). Once the snapshot is complete, the LoggedThrough class can fall-back to a strategy of handling each Holder independently as they are called (get, put, etc.), and a background process flushes any inactive values.

The 'magic' in this granular locking is that each Holder is individually synchronized, and knows its own state. This state can be safely dumped to the backing store, and when it does, the Holder becomes a simple pass-through entity.

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;


public class VersionedSnapshotMap<K,V> {

    private static class Holder<T> {
        private boolean live = true;
        private T value = null;
    }

    private class PassThrough {
        public V put(K key, V val) {
            return  store.put(key,val);
        }
        public V remove(K key) {
            return  store.remove(key);
        }
        public V get(K key) {
            return store.get(key);
        }
    }

    private class LoggedThrough extends PassThrough {

        private final ConcurrentHashMap<K, Holder<V>> diff = new ConcurrentHashMap<>();
        private final AtomicBoolean record = new AtomicBoolean(true);

        @Override
        public V get(K key) {
            // no need to worry about recording things....
            Holder<V> holder = diff.get(key);
            if (holder != null) {
                synchronized(holder) {
                    if (holder.live) {
                        return holder.value;
                    }
                }
            }
            // no race condition, the get can safely get the old value even
            // if a new holder was created in the race.
            return store.get(key);
        }

        private V undercover(K key, V val, boolean remove) {
            if (!record.get()) {
                // recording complete for this logger.
                // either the Holder has:
                //   1. never been created
                //   2. created, but not yet written back
                //   3. created, and written back
                //   4. created, written back, and removed.
                final Holder<V> holder = diff.get(key);
                if (holder != null) {
                    // 2, or 3.
                    synchronized (holder) {
                        if (holder.live) {
                            V prev = holder.value;
                            holder.value = null;
                            // push back this Holder, and mark it dead.
                            // subsequent calls will find it gone...
                            store.put(key, val);
                            holder.live = false;
                            return prev;
                        }
                    }
                }
                // 1, 3, or 4.
                if (remove) {
                    return store.remove(key);
                }
                return store.put(key, val);
            }

            // we are still recording...
            // optimistically create a new Holder.
            // we will have to discard this if another thread has already done one.
            Holder<V> nref = new Holder<>();
            nref.value = store.get(key);

            // yes, put it on the queue even if the recording may have stopped.
            Holder<V> race = diff.putIfAbsent(key, nref);

            // holder becomes whatever instance was first registered for this key.
            Holder<V> holder = race == null ? nref : race;

            synchronized(holder) {
                if (holder.live) {
                    V prev = holder.value;
                    holder.value = val;
                    if (!record.get()) {
                        // we thought we were recording, but that
                        // changed in a race condition. We push our value
                        // back through to the source.
                        holder.live = false;
                        holder.value = null;
                        diff.remove(key);
                        if (remove) {
                            store.remove(key);
                        } else {
                            store.put(key, val);
                        }
                    }
                    return prev;
                }
            }
            if (remove) {
                return store.remove(key);
            }
            return store.put(key, val);
        }

        @Override
        public V put(K key, V val) {
            return undercover(key, val, false);
        }

        @Override
        public V remove(K key) {
            return undercover(key, null, true);
        }

        public void flush() {
            // OK, recordings are no longer applied
            record.set(false);
            while (!diff.isEmpty()) {
                Iterator<Map.Entry<K, Holder<V>>> it = diff.entrySet().iterator();
                while (it.hasNext()) {
                    Map.Entry<K, Holder<V>> entry = it.next();
                    Holder<V> holder = entry.getValue();
                    K key = entry.getKey();
                    synchronized (holder) {
                        if (holder.live) {
                            holder.live = false;
                            if (holder.value  != null) {
                                store.put(key, holder.value);
                            } else {
                                store.remove(key);
                            }
                            holder.value = null;
                        }
                    }
                    it.remove();
                }
            }
        }

    }

    private final PassThrough simplepass = new PassThrough();
    private final ConcurrentHashMap<K, V> store = new ConcurrentHashMap<>();
    private final AtomicReference<PassThrough> core = new AtomicReference<>(simplepass);

    public V get(K key) {
        return core.get().get(key);
    }

    public V put(K key, V val)  {
        return core.get().put(key, val);
    }

    public V remove(K key) {
        return core.get().remove(key);
    }    

    public void snapshot(BiConsumer<? super K, ? super V> action) {
        LoggedThrough logged = new LoggedThrough();
        if (core.compareAndSet(simplepass, logged)) {
            try {
                store.forEach(action);
            } finally {
                logged.flush();
                if (!core.compareAndSet(logged, simplepass)) {
                    throw new IllegalStateException("Unable to restore the simple passthrough");
                }
            }
        } else  {
            throw new IllegalStateException("Only one snapshot at a time, please");
        }
    }

}

I tested the above code using the folowing hacks...

  1. take a copy of your system properties using the snapshot
  2. take another copy, each time you dump an item though, use the Exchanger as an interlock, and randomly change something in the Map.
  3. ensure the original copy, and the concurrently modified copy are the same.
  4. ensure that the modifications made during the second snapshot are now shown in the third.

Here's the test code (please don't hold it up to code review standards, it is a hack...):

private static final <P,Q> void printPQ(P p, Q q, HashMap<P,Q> outsb) {
    outsb.put(p, q);
}

private static final void randomAte(VersionedSnapshotMap<String,String> smap, Exchanger<String> ex, final String term) {
    String[] keys = System.getProperties().stringPropertyNames().toArray(new String[0]);
    Random rand = new Random();
    try {
        String token = "Boo";
        while ((token = ex.exchange(token)) != term) {
            switch(rand.nextInt(10)) {
            case 0:
            case 1:
                // remove a key
                smap.remove(keys[rand.nextInt(keys.length)]);
                break;
            case 2:
            case 3:
                // add a new key
                smap.put("" + System.nanoTime(), "Modded");
                break;
            default:
                // modify an existing key.
                smap.put(keys[rand.nextInt(keys.length)], "" + System.currentTimeMillis());
            }
        }
    } catch (InterruptedException ie) {
        ie.printStackTrace();
    }
}

public static void main(String[] args) throws InterruptedException {

    final VersionedSnapshotMap<String, String> smap = new VersionedSnapshotMap<>();
    System.getProperties().forEach((k,v) -> smap.put(String.valueOf(k), String.valueOf(v)));
    HashMap<String,String> base = new HashMap<>();
    smap.snapshot((k,v) -> VersionedSnapshotMap.printPQ(k,v, base));

    final Exchanger<String> exchanger = new Exchanger<>();

    ExecutorService service = Executors.newCachedThreadPool();
    // go through and randomize values.
    final String DONE = "Done";
    service.execute(() -> randomAte(smap, exchanger, DONE));
    HashMap<String,String> after = new HashMap<>();
    smap.snapshot((k,v) -> {
        printPQ(k, v, after);
        try {
            exchanger.exchange("Hoo");
        } catch (InterruptedException ie) {

        }
    });
    exchanger.exchange(DONE);
    service.shutdown();
    service.awaitTermination(10, TimeUnit.SECONDS);

    System.out.println(base);
    System.out.println(after);
    System.out.println(base.equals(after));

    HashMap<String,String> post = new HashMap<>();
    smap.snapshot((k,v) -> VersionedSnapshotMap.printPQ(k,v, post));
    System.out.println(post);
    System.out.println(base.equals(post));

}
\$\endgroup\$
  • \$\begingroup\$ first of all thanks for your suggestion. i cant test this right now, but i think i see 2 issues with your impl - 1. your "flush" loop at the end of snapshot() is not guaranteed to see all values (that "may" from the javadoc in my question). in which case you might reach snaplog.set(null) and "lose" modifications. 2. put() might get null for slog, then context switch to a snapshot() call, and then put() will resume later (during snapshot) and continue with store.put() - which will alter the map during a snapshot \$\endgroup\$ – radai Oct 6 '14 at 20:02
  • \$\begingroup\$ I understand your concerns, but I believe I am safe on both counts. Re the missing flush: I am not convinced the flush is necessary, as per the documentation: "they are guaranteed to traverse elements as they existed upon construction exactly once", so any member (key) in the Map at entrySet() time is guaranteed to be in the result. Youa re right, actually, the flush is unnecessary... I will remove it. Give me a sec. \$\endgroup\$ – rolfl Oct 6 '14 at 20:31
  • \$\begingroup\$ As for your comment about the intra-modication of the actual keys, by using the synchronize on the DataLog member, that is not possible. The subsequent put can only happen after the original method's modification was logged (the original method is the first one to create a DataLog instance for that key). \$\endgroup\$ – rolfl Oct 6 '14 at 20:34
  • \$\begingroup\$ @radai - I have mocked up the process I described above, and it has problems in my implementation. Don't use my code. \$\endgroup\$ – rolfl Oct 6 '14 at 22:28
  • \$\begingroup\$ I have significantly rewritten my answer, it is essentially 'new', and the old version is gone. You were right with some of your criticism, but the above code should be clearer, and the algorithm is not too far removed from yours, with just some changes to the global locking, and better concurrency/granularity of locking. \$\endgroup\$ – rolfl Oct 7 '14 at 4:16
7
\$\begingroup\$

There are a number of things that I think you have completely overengineered.

Holder

This can be replaced with Optional

SingleSnapshotMap

This class exposes 4 methods:

  • get
  • put
  • remove
  • snapshot

Which is a 'light' implementation of a Map. The amount of locking though is overkill. I would go with short, sweet, and simple....

The following code takes a 'defensive' copy of the HashSet (which is a relatively fast operation - in the order of milliseconds). If you only occasionally do this operation I cannot see any reason why it would be a problem. Once you have the snapshot, it is independent of the store, and the process for 'backing it up' can be as slow as necessary.

Unless you have significant evidence to the contrary, I cannot imagine how trying to maintain a read/write list while allowing the concurrent backup, will help you. Do you have evidence that the functionality may be needed?

public class SingleSnapshotMap<K,V> {

    private final HashMap<K,V> store;
    private final Lock lock = new ReentrantLock();

    public SinglSnapshotMap() {
        store = new HashMap<>();
    }

    public V get(K key) {
        lock.lock();
        try {
            return store.get(key);
        } finally {
            lock.unlock();
        }
    }

    public V put(K key, V val) {
        lock.lock();
        try {
            return store.put(key, val);
        } finally {
            lock.unlock();
        }
    }

    public V remove(K key) {
        lock.lock();
        try {
            return store.remove(key, val);
        } finally {
            lock.unlock();
        }
    }

    public Map<K,V> snapshot() {
        lock.lock();
        try {
            return new HashMap<>(store);
        } finally {
            lock.unlock();
        }
    }
}
\$\endgroup\$
  • \$\begingroup\$ i failed to specify this - but the map is expected to hold millions of keys. that copy will kill me ... i'll update the question. as a side note - Optional is immutable. you cannot use it for what i did with Holder \$\endgroup\$ – radai Oct 6 '14 at 16:01
  • \$\begingroup\$ Have you timed how long it will take to do a new HashMap<>(somemap) ? \$\endgroup\$ – rolfl Oct 6 '14 at 16:05
  • \$\begingroup\$ with millions of entries - out of heapspace. \$\endgroup\$ – radai Oct 6 '14 at 16:06
  • \$\begingroup\$ I guess I got very confused with your expected usage: Map<Long, String> copyMap = new HashMap<>(); map.snapshot(copyMap::put); .... \$\endgroup\$ – rolfl Oct 6 '14 at 16:08
  • \$\begingroup\$ that was very contrived for brevity's sake. the real usage would be streaming to disk. i've fixed my usage example \$\endgroup\$ – radai Oct 6 '14 at 16:09

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.