Thread safety of this queue backed by ConcurrentHashMap

Question

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class DedupingQueue<E> implements QueueWrapper<E> {
    private static final Logger LOGGER = LoggerFactory.getLogger(DedupingQueue.class);

private final Map<E, Future<E>> itemsBeingWorkedOn = new ConcurrentHashMap<>();
private final AsyncWorker<E> asyncWorker;

public DedupingQueue(AsyncWorker<E> asyncWorker) {
    this.asyncWorker = asyncWorker;
}

@Override
public Future<E> submit(E e) {
    if (!itemsBeingWorkedOn.containsKey(e)) {
        itemsBeingWorkedOn.put(e, asyncWorker.executeWorkAsync(e, this));
    } else {
        LOGGER.debug("Rejected [{}] as it's already being worked on", e);
    }
    return itemsBeingWorkedOn.get(e);
}

@Override
public void complete(E e) {
    LOGGER.debug("Completed [{}]", e);
    itemsBeingWorkedOn.remove(e);
}

@Override
public void rejectAndRetry(E e) {
    itemsBeingWorkedOn.putIfAbsent(e, asyncWorker.executeWorkAsync(e, this));
}

}

I am having some difficulty reasoning the thread safeness of the above code.

I reckon complete and rejectAndretry are completely thread-safe as the map is thread safe. But what about submit? Also, how can I make it thread safe in the most efficient way?

Answer 1

What if you have 2 threads that submit the same E. The first one runs untill the switch here in the code snippet below. Then the second thread runs up to that point. Both pass the check because it's not in the map yet, but then both will put in a new Future in the map.

@Override
public Future<E> submit(E e) {
    if (!itemsBeingWorkedOn.containsKey(e)) {
        //***switch here***
        itemsBeingWorkedOn.put(e, asyncWorker.executeWorkAsync(e, this));
    } else {
        LOGGER.debug("Rejected [{}] as it's already being worked on", e);
    }
    return itemsBeingWorkedOn.get(e);
}

Easiest way to make this thread safe would be to just make the entire method synchronised. If you're expecting to skip a lot that are already in you might consider this idea based on double checked locking:

@Override
public Future<E> submit(E e) {
    if (!itemsBeingWorkedOn.containsKey(e)) {
        synchronised(this){
            if (!itemsBeingWorkedOn.containsKey(e)) {
                itemsBeingWorkedOn.put(e, asyncWorker.executeWorkAsync(e, this));
           }
       }
    } else {
        LOGGER.debug("Rejected [{}] as it's already being worked on", e);
    }
    return itemsBeingWorkedOn.get(e);
}

This might have some really sublte unexpected results when one finishes (and is taken out of the map) just when another is submitted (this can return null). Depending on how you intend to use the class this may not be a problem though.