0
\$\begingroup\$

I was coding for the following programming question:

Suppose you have two independent blocking queues that will keep getting new data, the new data will always be greater than the last element of the queue it is going to. You can only use getNext() to get the data from these two blocking queues and each data can be fetched only once. Write a program to output all pairs of data from these two blocking queues that have difference smaller than 1.

My implementation uses three additional queues with two holding data from the two given blocking queues, and one holding the result. I used two threads each handling one of the queue, and send the output result to the result queue. Whenever a new data comes into the first queue, I will pop every data in the second queue that has difference larger or equal to 1, then looping through second queue to output all pairs smaller than 1.

Here is my implementation:

The BlockingQueueCalculator class handles a individual queue and performs pair calculation. It will wait for new element from the input Blocking Queue, and will terminate whenever fed a POISON. The lock object will prevent race condition, so that only one queue is allowed to process at a time.

public class BlockingQueueCalculator implements Runnable {
    public static final double POISON = Double.NEGATIVE_INFINITY;
    private final Deque<Double> q1;
    private final Deque<Double> q2;
    private final BlockingDeque<Double> inputQueue;
    private final Deque<Tuple<Double, Double>> pairsQueue;
    private final Object lock;

    static class Tuple<T, V> {
        T t;
        V v;

        public Tuple(T t, V v) {
            this.t = t;
            this.v = v;
        }
    }

    public BlockingQueueCalculator(Deque<Double> q1,
                                   Deque<Double> q2,
                                   Object lock,
                                   BlockingDeque<Double> input,
                                   Deque<Tuple<Double, Double>> pairs) {
        this.q1 = q1;
        this.q2 = q2;
        this.lock = lock;
        this.inputQueue = input;
        this.pairsQueue = pairs;
    }

    public void calculate(double timestamp) throws InterruptedException {
        if (!q2.isEmpty()) {
            while (!q2.isEmpty() && Math.abs(q2.peekFirst() - timestamp) >= 1.0) {
                System.out.println("Discarding from q2: " + q2.pollFirst());
            }

            for (Double oldTimestamp : q2) {
                System.out.println("Checking q2 element: " + oldTimestamp);
                if (oldTimestamp == null || Math.abs(oldTimestamp - timestamp) >= 1.0) {
                    return;
                }

                System.out.println("Adding to Pairs: " + oldTimestamp + ", " + timestamp);
                pairsQueue.add(new Tuple<>(oldTimestamp, timestamp));
            }
        }
    }

    @Override
    public void run() {
        try {
            double timeStamp;
            synchronized (lock) {
                while ((timeStamp = inputQueue.takeFirst()) != POISON) {
                    System.out.println("Adding to q1: " + timeStamp);
                    q1.add(timeStamp);
                    calculate(timeStamp);

                }
            }
        } catch (InterruptedException e) {
            System.out.println("Thread interrupted: " + e);
            Thread.currentThread().interrupt();
        }
    }
}

The BlockingQueuePairFinder is the main thread that setup environment and start other worker threads.

public class BlockingQueuePairFinder {
    private final BlockingDeque<Double> inputQueue1;
    private final BlockingDeque<Double> inputQueue2;

    private final Deque<Double> q1 = new LinkedList<>();
    private final Deque<Double> q2 = new LinkedList<>();
    private final Deque<BlockingQueueCalculator.Tuple<Double, Double>> pairsList =
            new LinkedList<>();
    private final Object lock = new Object();
    private Thread t1;
    private Thread t2;

    public BlockingQueuePairFinder(BlockingDeque<Double> inputQueue1,
                                   BlockingDeque<Double> inputQueue2) {
        this.inputQueue1 = inputQueue1;
        this.inputQueue2 = inputQueue2;
        this.t1 = new Thread(new BlockingQueueCalculator(q1, q2, lock, inputQueue1, pairsList));
        this.t2 = new Thread(new BlockingQueueCalculator(q2, q1, lock, inputQueue2, pairsList));
    }

    public void run() {
        if(t1 == null || !t1.isAlive()) {
            t1 = new Thread(new BlockingQueueCalculator(q1, q2, lock, inputQueue1, pairsList));
            t1.start();
        }
        if(t2 == null || !t2.isAlive()) {
            t2 = new Thread(new BlockingQueueCalculator(q2, q1, lock, inputQueue2, pairsList));
            t2.start();
        }
    }

    public static void main(String...args) {
        BlockingDeque<Double> q1 = new LinkedBlockingDeque<>();
        BlockingDeque<Double> q2 = new LinkedBlockingDeque<>();

        BlockingQueuePairFinder finder = new BlockingQueuePairFinder(q1, q2);
        finder.run();

        q1.offerLast(0.2);
        q2.offerLast(1.0);
        q1.offerLast(1.4);
        q1.offerLast(3.0);
        q2.offerLast(1.1);
        q2.offerLast(3.5);

        q1.offerLast(BlockingQueueCalculator.POISON);
        q2.offerLast(BlockingQueueCalculator.POISON);

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {

        }

        for(BlockingQueueCalculator.Tuple<Double, Double> tuple : finder.pairsList) {
            System.out.println(tuple.t + ", " + tuple.v);
        }
    }
}

Is this a good OO and concurrency design? If not, what design would you propose?

Also, given the fact that I used lock(semaphore?) to allow only one queue to operate at any time, if I generalize this design to accept a List<BlockingDeque>, then my design would simply block all but one queue at any time. Is this an optimal solution and what would you change under such circumstance?

\$\endgroup\$
1
\$\begingroup\$

Hmm, maybe I don't understand something, but my thoughts are the following:

inputBlockingQueue1 | <- they keep receiving new data
inputBlockingQueue2 |

The fact that they say each new data is bigger than the last one just says that we have two lists of integers that are sorted.

[1] <- 1, 3, 5, 7 ,10
[2] <- 11, 12,14,15,16,17

Now find two pairs two queues that have difference < 1, means to find two equal numbers? Then it would look like merge sort algorithm that part when we have two lists. We check which one is bigger.

Say, for:

[1] <- 1,3,5,6,7,10
q1Value = 1;
[2] <- 10,11,12,13,14,15
q2Value = 10

We check that [1] has 1 which is smaller than 10, we keep taking values until we get to:

[1] becomes empty
q1Value = 10
q2Value = 10 -> ok , we found equal elements

If q1 is empty, we're done. If we still have something else, then we keep taking from queue that has smaller value.

\$\endgroup\$

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.