1
\$\begingroup\$

This solution has a total of four classes.

  1. Main.java: Class with main method. Execution starts here.
  2. BoundedBufferV2.java: The class representing a thread-safe queue of limited capacity.
  3. ProducerThreadV2.java: A class representing the producer thread. It produces a random string and tries to push it into the queue. Waits for a second before doing the same thing again.
  4. ConsumerThreadV2.java: A class representing the consumer thread. It tries to get an item from the queue. Waits for a second before doing the same thing again.

Main.java

public class Main {
    public static void main(String[] args) {
        BoundedBufferV2 boundedBufferV2 = new BoundedBufferV2(10);
        for (int i = 0; i < 5; i++) {
            new ProducerThreadV2(boundedBufferV2).start();
        }
        for (int i = 0; i < 5; i++) {
            new ConsumerThreadV2(boundedBufferV2).start();
        }
    }
}

BoundedBufferV2.java - This is the main class where the thread-safety is implemented.

public class BoundedBufferV2 {
    private final Queue<String> queue;
    private final Lock lock;
    private final Semaphore full;
    private final Semaphore empty;

    public BoundedBufferV2(int capacity) {
        this.queue = new ArrayDeque<>(capacity);
        this.lock = new ReentrantLock();
        this.full = new Semaphore(0, true);
        this.empty = new Semaphore(capacity, true);
    }

    public void enqueue(String item) {
        boolean hasSpace = empty.tryAcquire();
        if (!hasSpace) return;
        boolean lockAcquired = lock.tryLock();
        if (!lockAcquired) {
            empty.release();
            return;
        }
        try {
            queue.add(item);
            System.out.println("ENQ-> " + queue);
        } finally {
            lock.unlock();
            full.release();
        }
    }

    public String dequeue() {
        boolean hasItems = full.tryAcquire();
        if (!hasItems) return null;
        boolean lockAcquired = lock.tryLock();
        if (!lockAcquired) {
            full.release();
            return null;
        }
        try {
            String item = queue.poll();
            System.out.println("DEQ-> " + queue);
            return item;
        } finally {
            lock.unlock();
            empty.release();
        }
    }
}

ProducerThreadV2.java

import java.util.Random;

public class ProducerThreadV2 extends Thread {
    private final Random random;
    private final BoundedBufferV2 boundedBufferV2;

    public ProducerThreadV2(BoundedBufferV2 boundedBufferV2) {
        this.random = new Random();
        this.boundedBufferV2 = boundedBufferV2;
    }

    @Override
    public void run() {
        while (true) {
            String randomStr = String.valueOf(random.nextInt());
            boundedBufferV2.enqueue(randomStr);
            try {
                Thread.sleep(1_000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
}

ConsumerThreadV2.java

public class ConsumerThreadV2 extends Thread {
    private final BoundedBufferV2 boundedBufferV2;

    public ConsumerThreadV2(BoundedBufferV2 boundedBufferV2) {
        this.boundedBufferV2 = boundedBufferV2;
    }

    @Override
    public void run() {
        while (true) {
            boundedBufferV2.dequeue();
            try {
                Thread.sleep(1_000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
}

A couple of problems I already realize are busy-waiting if Semaphore acquisition or Lock acquisition does not happen and returning null from the method. I am looking for inputs on Deadlock, race-condition or data race. Thank you.

\$\endgroup\$
1
  • 1
    \$\begingroup\$ A couple of problems I already realize are busy-waiting if Semaphore acquisition or Lock acquisition does not happen and returning null from the method. This is a synonym of your code not working \$\endgroup\$ Mar 7 at 19:31

1 Answer 1

1
\$\begingroup\$

versionitis

This submission did not include a maven pom.xml file. Consider adding one, and using that to deal with SemVer details. Names like BoundedBufferV2 seem less convenient than e.g. BoundedBuffer.

documentation

This submission contains no /** javadoc */ remarks.

Dealing with concurrent processing can be a bit tricky, and typically the key to correct code is carefully documenting the class invariants, the things that are always true across all interleavings of thread execution. That didn't happen here.

Given such a spec, we could then reason about each method in isolation, asking if it preserves our invariants.

automated test suite

There isn't one.

More than one library would offer a convenient framework for structuring your tests so others could verify that invariants hold, even when run against future JVMs and future libraries that may be released in the coming years.

invariants

It appears the key predicate you wish to enforce is bounded memory consumption, so the deque will never have more than e.g. 10 entries.

Yet we have two semaphores plus a mutex, in addition to the deque. Which seems like more coordination datastructures than necessary. The names empty and full are less helpful than a numRecords (or numRemainingSlots) coordination variable would be.

We do not see the synchronized keyword in this source code, though it could have simplified the code. If that was a deliberate design decision, it would be worth writing that down explicitly, as comments in the source.

records per second

Throughput is calibrated to be ~ 1 record produced per second, and ~ 1 record consumed per second. I was expecting that one or the other would be slightly higher, so we settle into steady state of either mostly empty or full.

Or perhaps you'd care to randomly perturb the rate every so often.

Also, bursts of producing, and bursts of consuming, would be of interest. Where a burst might fill the deque, or completely drain it.

latency

I do not understand why we're polling with .tryAcquire(), rather than blocking indefinitely or blocking with a timeout. Polling means that consumer sees records that are more stale than necessary. If we block, then we can obtain a fresh record immediately after it is produced. Consider producing timestamp records, rather than random numbers. Then the consumer could report on how long records lingered in the queue.

Given the while (true) loops, it seems reasonable for at least one side to patiently wait.


There are some hidden assumptions behind this code. Please make them explicit -- write them down within the source. Then future maintenance engineers will be in a frame of mind similar to how you're currently viewing the problem space.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.