"Atomically" update an entire array

Question

I have a single writer thread and single reader thread to update and process a pool of arrays(references stored in map). The ratio of writes to read is almost 5:1(latency of writes is a concern).

The writer thread needs to update few elements of an array in the pool based on some events. The entire write operation(all elements) needs to be atomic.

I want to ensure that reader thread reads the previous updated array if writer thread is updating it(something like volatile but on entire array rather than individual fields). Basically, I can afford to read stale values but not block.

Also, since the writes are so frequent, it would be really expensive to create new objects or lock the entire array while read/write.

Is there a more efficient data structure that could be used or use cheaper locks ?

Answer 1

How about this idea: The writer thread does not mutate the array. It simply queues the updates.

The reader thread, whenever it enters a read session that requires a stable snapshot of the array, applies the queued updates to the array, then reads the array.

class Update
{
    int position;
    Object value;
}

ArrayBlockingQueue<Update> updates = new ArrayBlockingQueue<>(Integer.MAX_VALUE);

void write()
{
    updates.put(new Update(...));
}

Object[] read()
{
    Update update;
    while((update=updates.poll())!=null)
        array[update.position] = update.value;

    return array;
}

Answer 2

Is there a more efficient data structure?

Yes, absolutely! They're called persistent data structures. They are able to represent a new version of a vector/map/etc merely by storing the differences with respect to a previous version. All versions are immutable, which makes them appropiate for concurrency (writers don't interfere/block readers, and vice versa).

In order to express change, one stores references to a persistent data structure in a reference type such as AtomicReference, and changes what those references point to - not the structures themselves.

Clojure provides a top-notch implementation of persistent data structures. They're written in pure, efficient Java.

The following program exposes how one would approach your described problem using persistent data structures.

import clojure.lang.IPersistentVector;
import clojure.lang.PersistentVector;

public class AtomicArrayUpdates {

    public static Map<Integer, AtomicReference<IPersistentVector>> pool
        = new HashMap<>();
    public static Random rnd = new Random();
    public static final int SIZE = 60000;
    // For simulating the reads/writes ratio
    public static final int SLEEP_TIMÉ = 5;

    static {        
        for (int i = 0; i < SIZE; i++) {
            pool.put(i, new AtomicReference(PersistentVector.EMPTY));
        }
    }

    public static class Writer implements Runnable {   
        @Override public void run() {
            while (true) {
                try {
                    Thread.sleep(SLEEP_TIMÉ);
                } catch (InterruptedException e) {}

                int index = rnd.nextInt(SIZE);
                IPersistentVector vec = pool.get(index).get();

                // note how we repeatedly assign vec to a new value
                // cons() means "append a value".
                vec = vec.cons(rnd.nextInt(SIZE + 1)); 
                // assocN(): "update" at index 0
                vec = vec.assocN(0, 42); 
                // appended values are nonsense, just an example!
                vec = vec.cons(rnd.nextInt(SIZE + 1)); 

                pool.get(index).set(vec);

            }
        }
    }

    public static class Reader implements Runnable {
        @Override public void run() {
            while (true) {
                try {
                    Thread.sleep(SLEEP_TIMÉ * 5);
                } catch (InterruptedException e) {}

                IPersistentVector vec = pool.get(rnd.nextInt(SIZE)).get();
                // Now you can do whatever you want with vec.
                // nothing can mutate it, and reading it doesn't block writers!
            }
        } 
    }

    public static void main(String[] args) {
        new Thread(new Writer()).start();
        new Thread(new Reader()).start();
    }
}