Adding Functionality to ArrayBlockingQueue

Question

I'm trying to add functionality to ArrayBlockingQueue, specifically I want the queue to only keep unique elements, i.e not enqueuing an entry if it is already contained in the queue. Since the functionality I desired is the same as the extension of Vector in item 4.4 of JCIP, I tried implementing it using the approaches there.

Implementation by extending doesn't work, because ArrayBlockingQueue implements its mutual exclusion using a package private ReentrantLock, so as an extending class I could not get a reference to it. This would be a fragile approach even if it did work.
Implementation by client side locking doesn't work, since there is no client side locking support.

Implementation by composition seemed like the way to go at first, producing code such as

public class DistinctBlockingQueue<E> implements BlockingQueue<E> {
    private final BlockingQueue<E> backingQueue;

    public DistinctBlockingQueue(BlockingQueue<E> backingQueue) {
        this.backingQueue = backingQueue;
    }

    @Override
    public synchronized boolean offer(E e) {
        if (backingQueue.contains(e)) {
            return false;
        }

        return backingQueue.offer(e);
    }

    @Override
    public synchronized E take() throws InterruptedException {
        return backingQueue.take();
    }

    // Other methods...
}

Unfortunately, when composing an ArrayBlockingQueue, this approach yields a deadlock in the following simple scenario:

Thread A calls take() and acquires the synchronized lock and the ArrayBlockingQueue's inner lock.
Thread A blocks upon seeing that the queue is empty and releases ArrayBlockingQueue's inner lock.
Thread B calls offer() with an element but can't acquire the synchronized lock, blocking forever.

My question is, how is it possible to achieve this functionality without rewriting ArrayBlockingQueue?

FaNaJ · Accepted Answer · 2016-05-23 12:26:46Z

4

Perhaps one simple and fast solution would be to use a java.util.concurrent.ConcurrentMap :

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class DistinctBlockingQueue<E> implements BlockingQueue<E> {

    private final BlockingQueue<E> backingQueue;
    private final ConcurrentMap<E, Boolean> elements = new ConcurrentHashMap<>();

    public DistinctBlockingQueue(BlockingQueue<E> backingQueue) {
        this.backingQueue = backingQueue;
    }

    @Override
    public boolean offer(E e) {
        boolean[] add = {false};
        elements.computeIfAbsent(e, k -> add[0] = true);
        return add[0] && backingQueue.offer(e);
    }

    @Override
    public E take() throws InterruptedException {
        E e = backingQueue.take();
        elements.remove(e);
        return e;
    }

    // Other methods

}

Note that there's no need for synchronization.

EDIT :

The documentation at java.util.concurrent.ConcurrentHashMap says :

/**
 * If the specified key is not already associated with a value,
 * attempts to compute its value using the given mapping function
 * and enters it into this map unless {@code null}.  The entire
 * method invocation is performed atomically, so the function is
 * applied at most once per key.  Some attempted update operations
 * on this map by other threads may be blocked while computation
 * is in progress, so the computation should be short and simple,
 * and must not attempt to update any other mappings of this map.
 *
 * @param key key with which the specified value is to be associated
 * @param mappingFunction the function to compute a value
 * @return the current (existing or computed) value associated with
 *         the specified key, or null if the computed value is null
 * @throws NullPointerException if the specified key or mappingFunction
 *         is null
 * @throws IllegalStateException if the computation detectably
 *         attempts a recursive update to this map that would
 *         otherwise never complete
 * @throws RuntimeException or Error if the mappingFunction does so,
 *         in which case the mapping is left unestablished
 */
public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
    ...
}

I've added some additional checks :

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class DistinctBlockingQueue<E> implements BlockingQueue<E> {

    private final BlockingQueue<E> backingQueue;
    private final ConcurrentMap<E, Boolean> elements = new ConcurrentHashMap<>();

    public DistinctBlockingQueue(BlockingQueue<E> backingQueue) {
        this.backingQueue = backingQueue;
    }

    @Override
    public boolean offer(E e) {
        boolean[] add = {false};
        elements.computeIfAbsent(e, k -> add[0] = true);
        if (add[0]) {
            // make sure that the element was added to the queue,
            // otherwise we must remove it from the map
            if (backingQueue.offer(e)) {
                return true;
            }
            elements.remove(e);
        }
        return false;
    }

    @Override
    public E take() throws InterruptedException {
        E e = backingQueue.take();
        elements.remove(e);
        return e;
    }

    @Override
    public String toString() {
        return backingQueue.toString();
    }

    // Other methods

}

and... let's do some concurrency tests :

BlockingQueue<String> queue = new DistinctBlockingQueue<>(new ArrayBlockingQueue<>(100));

int n = 1000;
ExecutorService producerService = Executors.newFixedThreadPool(n);

Callable<Void> producer = () -> {
    queue.offer("a");
    return null;
};

producerService.invokeAll(IntStream.range(0, n).mapToObj(i -> producer).collect(Collectors.toList()));
producerService.shutdown();

System.out.println(queue); // prints [a]

edited May 23, 2016 at 12:26

answered May 23, 2016 at 6:55

FaNaJ

1,3591 gold badge16 silver badges39 bronze badges

Sign up to request clarification or add additional context in comments.

8 Comments

D Levant Over a year ago

Thanks for your answer, but your solution isn't good. The problem arises when two threads try to offer the same item. Suppose the first thread reaches its return statement and gets context switched, then the second thread starts to run and when it reaches its return statement e will be offered, and then when the first thread gets rescheduled e will be offered once more. We must make sure that the map and queue do not get modified in between the absent check and the offer, which is why we must use a lock, or perhaps a more clever means of synchronization.

FaNaJ Over a year ago

@DLevant This will never happen. computeIfAbsent does the trick... See updates.

D Levant Over a year ago

There's nothing that prevents another thread from being scheduled right after add[0] was set to true on the first thread, in which case the same element can be added twice. Tests like these are hardly a proof when it comes to concurrency issues. Try adding something like notify(); wait(1000); before if(add[0]) in your code, and I believe that you should see what I'm talking about.

bowmore Over a year ago

@DLevant "There's nothing that prevents another thread from being scheduled right after add[0] was set to true on the first thread, in which case the same element can be added twice." This is false, the second thread would not be able to add the same element again, and its add[0] would remain false. Subsequently it would not add the element to the backingQueue.

FaNaJ Over a year ago

Dear @DLevant , the truth is that the entire invocation of offer doesn't need to be performed atomically. we just need to be sure that an element is added to the elements only once. computeIfAbsent is supposed to do this in a fast and atomic manner. it doesn't necessarily mean that it's slow if it's complex and vice versa. so that add[0] = true will be executed only once per element and as @bowmore said, the second thread would not be able to add the same element again.

|

D Levant · Accepted Answer · 2016-05-24 13:36:42Z

I found a partial answer to my question. The offer operation is not atomic as I wanted, however the queue is distinct.

public class DistinctBlockingQueue<E> implements BlockingQueue<E> {
    private final BlockingQueue<E> backingQueue;
    private final Set<E> entriesSet = ConcurrentHashMap.newKeySet();

    public DistinctBlockingQueue(BlockingQueue<E> backingQueue) {
        this.backingQueue = backingQueue;
        entriesSet.addAll(backingQueue);
    }

    @Override
    public boolean offer(E e) {
        if (!entriesSet.add(e))
            return false;

        boolean added = backingQueue.offer(e);
        if (!added) {
            entriesSet.remove(e);
        }

        return added;
    }

    @Override
    public E take() throws InterruptedException {
        E e = backingQueue.take();
        entriesSet.remove(e);

        return e;
    }

    // Other methods...
}

The additional Set is not an issue, since I would want to use one anyway in order to have reasonable performance.

However, I can think of one issue with this implementation, if it's used in conjunction with a bounded queue implementation (such as ArrayBlockingQueue), the set will not be bounded, thus the set can grow very large when there are many offers blocked.

This solution divides an operation that clearly should be atomic, so I highly suspect that there should be other issues that I'm overlooking.

Collectives™ on Stack Overflow

Adding Functionality to ArrayBlockingQueue

2 Answers 2

8 Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

8 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related