Preserving thread safety using lambda

Question

I'm trying to update some code to use lambda expressions but I'm having a bit of trouble preserving thread safety.

I have multiple threads running that eventually call the following callback, which has a synchronized method that adds some results to a LinkedList.

final List<Document> mappedDocs = new LinkedList<>();
final MapCallback<Integer, Document> mapCallback = new MapCallback<Integer, Document>() {
    @Override
    public synchronized void done(int file, List<Document> results) {
        mappedDocs.addAll(results);
    }
};

However when I convert it to a lambda expression I lose the synchronized keyword and I'm not entirely sure how to get it back. I'm now getting a NullPointerException whenever I run my code.

final MapCallback<Integer, Document> mapCallback = (int file, List<Document> results) -> mappedDocs.addAll(results);

How can I make this thread safe again?

Answer 1

You could synchronise it on a different monitor, e.g.:

final MapCallback<Integer, Document> mapCallback = (int file, List<Document> results) -> {
  synchronized(mappedDocs) {
    mappedDocs.addAll(results);
  }
};

Or alternatively use a thread safe structure, for example a CopyOnWriteArrayList or a BlockingQueue.

Answer 2

I'd strongly recommend making mappedDocs be a thread-safe data structure (such as one from java.util.concurrent) or perhaps a synchronized wrapper created using Collections.synchronizedList.

I think you're lucky that synchronization works using the anonymous inner class. This works because there is exactly one instance of it, and there is no other code that mutates mappedDocs.

(Actually you might have a memory visibility problem, even as things stand. If other threads call MapCallback to add elements, something else needs to synchronize on mappedDocs after its construction and prior to reading the added elements.)

The root of the issue is that an anonymous inner class used this way is kind-of like a function, but since the creation of a new object is manifest, it's tempting to do things like synchronizing on it. But this is quite fragile. If this were refactored so that multiple instances of the AIC were created (e.g., to process documents from multiple sources), or if different AICs are created (e.g., to delete documents from the list if they need reprocessing), synchronizing on individual AIC instances would be completely broken.

Converting mappedDocs to a thread-safe data structure or wrapper takes care of the memory visibility issue and the concurrent access issue. This lets you use the simple lambda form, and it lets you introduce new operations on mappedDocs without regard to what threads are operating on it.