11
\$\begingroup\$

I want to code the following scenario:

  1. One Producer: produces n (n=100 here) objects
  2. m Consumers (m = 5 here): consumes k (k = 10 here) objects at a time in a round robin fashion; Consumer 1 first consumes the first 10 objects followed by Consumer 2 consuming another 10 objects and so on. This means that consumer 1 has to wait until consumer 5 is done consuming 10 objects.

I have written the code and it is working, but I feel like it is sub-optimal, might have bugs and has room for improvement.

public static void main(String[] args) {
        ConsumerProducer cp = new ConsumerProducer();
        cp.StartconsumerProducer();
}

// Creates Threads and waits for them to finish till timeout
public class ConsumerProducer {

    public void StartconsumerProducer() {
        ConsumerProducerMonitor mon = new ConsumerProducerMonitor();
        List threads = new ArrayList();
        // Create a producer
        Thread p1 = new Thread(new Producer(mon), "P1");
        p1.start();
        // Create consumer 1
        Thread c1 = new Thread(new Consumer(mon), "C1");
        c1.start();
        // Create consumer 2
        Thread c2 = new Thread(new Consumer(mon), "C2");
        c2.start();
        // Create consumer 3
        Thread c3 = new Thread(new Consumer(mon), "C3");
        c3.start();
        threads.add(p1);
        threads.add(c1);
        threads.add(c2);
        threads.add(c3);

        for (int i = 0; i < threads.size(); ++i) {
            try {
                ((Thread)threads.get(i)).join(20000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
}

// Producer class
public class Producer implements Runnable {
    ConsumerProducerMonitor mon;

    Producer(ConsumerProducerMonitor mon) {
        this.mon = mon;
    }

    @Override
    public void run() {
        // while (true) {
        for (int i = 0; i < 100; ++i) {
            mon.put(i, Thread.currentThread().getName());
        }
    }
}

// Consumer Class
public class Consumer implements Runnable {

    Consumer(ConsumerProducerMonitor mon) {
        this.mon = mon;
    }

    @Override
    public void run() {
        int ret = 0;
        while (ret >= 0) {
            ret = mon.get(Thread.currentThread().getName());
        }
    }

    private final ConsumerProducerMonitor mon;
}

// Produces and consumes items in round robin fashion. Uses Semaphores for the same.
public class ConsumerProducerMonitor {

    // produces items
    public synchronized void put(int item, String threadName) {
        if (isProduced) {
            return;
        }

        this.item = item;
        System.out.println("Producer " + threadName + " put Item: " + this.item);
        if (this.item == 99) {
            isProduced = true;
            this.notifyAll();
        }
    }

    private void consumeItems(String threadName) {
        for (int i = 0; i < 10; ++i) {
            if (this.item < 0) {
                return;
            }
            this.item--;
        }
        System.out.println("Consumer " + threadName + " consumed Items from " + (this.item + 10) + ", to " + this.item);
        if (!sem.tryAcquire()) {
            System.out.println("Failed to aquire semaphore for consumer: " + threadName);
        }
    }

    // consumes item
    public synchronized int get(String threadName) {
        if (!isProduced) {
            try {
                this.wait();
            } catch (InterruptedException e) {
                System.out.println("Caught Interrupted Exceptino while waiting to consume item: " + e.getMessage());
            }
        }

        if (this.item < 0) {
            sem.release(NUM_SEMAPHORES);
            return this.item;
        }

        if (isConsuming) {
            try {
                this.wait();
                isConsuming = true;
            } catch (InterruptedException e) {
                System.out.println("Caught Interrupted Exceptino while waiting to consume item: " + e.getMessage());
            }
        }

        switch (sem.availablePermits()) {
            case 1:
                if (threadName.equals("C3")) {
                    consumeItems(threadName);
                    if (threadName.equals("C3")) {
                        sem.release(NUM_SEMAPHORES);
                    }
                }
                break;
            case 2:
                if (threadName.equals("C2")) {
                    consumeItems(threadName);
                }
                break;
            case 3:
                if (threadName.equals("C1")) {
                    consumeItems(threadName);
                }
                break;
            default:
                break;
        }
        isConsuming = false;
        this.notifyAll();
        return this.item;
    }

    private static int NUM_SEMAPHORES = 3;
    private final Semaphore sem = new Semaphore(NUM_SEMAPHORES);
    private boolean isProduced = false;
    private boolean isConsuming = false;
    int item;
}

Output:

Producer P1 put Item: 0

Producer P1 put Item: 1

Producer P1 put Item: 2

Producer P1 put Item: 3

Producer P1 put Item: 4

Producer P1 put Item: 5 ...

Producer P1 put Item: 99

Consumer C1 consumed Items from 99, to 89

Consumer C2 consumed Items from 89, to 79

Consumer C3 consumed Items from 79, to 69

Consumer C1 consumed Items from 69, to 59

Consumer C2 consumed Items from 59, to 49

Consumer C3 consumed Items from 49, to 39

Consumer C1 consumed Items from 39, to 29

Consumer C2 consumed Items from 29, to 19

Consumer C3 consumed Items from 19, to 9

Consumer C1 consumed Items from 9, to -1
\$\endgroup\$
10
\$\begingroup\$

First of all it would be good to use the standard Java Conventions, i.e. method names start with a lower case letter. i.e. StartconsumerProducer shold either be startconsumerProducer or simply run.

Also handle InterruptedException correctly. For details have a look at the book Java Concurrency in Practice

In your case it does not make sense to catch the InterruptedException you can simply declare it as thrown.

The ArrayList for the threads is not necessary, you have a fix number of Threads, you can replace it by an Array.

public static void main(String[] args) throws InterruptedException {
    ConsumerProducer cp = new ConsumerProducer();
    cp.run();
}

You are not really waiting for all threads to fininsh. If one of your threads run in an timeout you simply ignore it. You can fix this by stetting a flag that signals your thread to terminate.

I've done this by adding a class Terminatable which Consumer and Producer extend and which have a flag that signals them to terminate as soon as possible.

Your ConsumerProducerMonitor contains only one value which is overwritten by the Producer Loop via the put method, nothing prevents this. The notifyAll calls are useless because nobody actually waits for the ConsumerProducerMonitor

There are a lot of issues in your code so I've decided to rewrite it in a way that actually works.

It uses a Queue to transport the elements Elem from the Producer to the Consumer

The Producer notifies any waiting Consumer via notifyAll about new Elements.

The Consumer uses an Index to determine whether he should pull elements from the queue and increments the shouldRun after he has taken the 10 elements from the queue.

In the Consumer I give you an example how the wait() in Java can be used, and a possible way how to deal with InterruptedExcpetions.

import java.util.Arrays;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class ConsumerProducer {

    public static void main(String[] args) throws InterruptedException {
        ConsumerProducer cp = new ConsumerProducer();
        cp.run();
    }

       // Creates Threads and waits for them to finish till timeout
        private LinkedBlockingQueue<Elem> queue = new LinkedBlockingQueue<>();

        public void run() throws InterruptedException {
            Terminatable[] terminateables = new Terminatable[] {
                    new Producer(queue), new Consumer(queue),
                    new Consumer(queue), new Consumer(queue), };

            startThreads(terminateables);
            join(terminateables);
        }

        private static void join(Terminatable[] terminatables)
                throws InterruptedException {
            List<Terminatable> maybeNotTerminated = Arrays
                    .asList(terminatables);

            while (!maybeNotTerminated.isEmpty()) {
                for (int idx = maybeNotTerminated.size() - 1; idx >= 0; --idx) {
                    Terminatable t = maybeNotTerminated.get(idx);
                    if (t.isAlive()) {
                        t.join(20_000);
                        // force termination
                        t.terminate = true;
                    } else {
                        maybeNotTerminated.remove(idx);
                    }
                }
            }
        }

        private static void startThreads(Runnable[] runnables) {
            for (int i = 0; i < runnables.length; i++) {
                new Thread(runnables[i]).start();
            }
        }

    static abstract private class Terminatable implements Runnable {
        volatile boolean terminate = false;

        public boolean isAlive() {
            return Thread.currentThread().isAlive();
        }

        public void join(long millis) throws InterruptedException {
            Thread.currentThread().join(millis);
        }
    }

    static private class Elem {
        public final int idx;
        public final String name;

        public Elem(int i, String name) {
            this.idx = i;
            this.name = name;
        }
    }

    // Producer class
    static private class Producer extends Terminatable implements Runnable {

        private Queue<Elem> queue;

        Producer(Queue<Elem> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            for (int i = 0; i <= 100; ++i) {
                Elem e = new Elem(i, Thread.currentThread().getName());                 
                queue.offer(e);

                synchronized (queue) {
                    queue.notifyAll();
                }

            }
        }
    }

    // Consumer Class
    static private class Consumer extends Terminatable implements Runnable {

        private Queue<Elem> queue;
        private final int myIdx;
        private static int consumerCounter = 0;
        private static AtomicInteger shouldRun = new AtomicInteger(0);

        public Consumer(Queue<Elem> queue) {
            this.queue = queue;
            myIdx = consumerCounter;
            consumerCounter++;
        }


        @Override
        // very long with two nested loops
        // should be refactored.
        public void run() {
            while (!terminate) {
                for (int idx = 0; idx < 10; ++idx) {                
                    try {
                        // wait until elements are produced
                        do {
                            if (queue.isEmpty()) {
                                synchronized (queue) {
                                    queue.wait(100);
                                }
                            }
                        } while (shouldRun.get() != myIdx);
                        // continue only if this consumer should execute
                        Elem elem = queue.poll();
                        if (elem != null) {
                            System.out.println("Thread "
                                    + Thread.currentThread().getName()
                                    + " fetch Element " + elem.idx);
                        }
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                    }

                }

                // allow the next consumer to run
                shouldRun.set((myIdx + 1) % consumerCounter);               
            }

        }

    }
}
\$\endgroup\$
  • \$\begingroup\$ Thanks so much for reviewing the code. I am going to spend a bit of time to understand this and try it. I got Java 1.6, need to upgrade to 1.7 since you are using the new version. \$\endgroup\$ – animageofmine Jun 16 '15 at 21:52
  • \$\begingroup\$ I believe I understand the code you suggested. Looks like the do/while loop in consumer class is keeping the thread busy and using processor time. I was wondering if there is a better way to deal with it. May be put the thread to suspended / sleep state and wait for the other consumer to notify when it is done with processing its own chunk of elements. For example. if consumer 1 was consuming elements 1- 10, consumer 2 is not spinning continuously, but rather sleeping and waiting for consumer 1 to wake it up. \$\endgroup\$ – animageofmine Jun 16 '15 at 22:53
  • \$\begingroup\$ You found one of my desing flaws. You are absolutely right that that's a waste of cpu time / resources. A very simple way to fix it is is to use sleep another one is to use wait notify using the shouldRun as Lock object. \$\endgroup\$ – andih Jun 17 '15 at 3:23
  • \$\begingroup\$ I get your point. I am marking this as an answer and spend a bit of time later on to implement the changes in while loop. The only lurking thought I still have is that we are using so many synchronization mechanisms here. (Atomic integer, synchronized, wait/notify) in addition to the blocking queue. I might be completely wrong here, but it looks like a very high cost for simply using so many of them and may be there is an optimization that can be done here. Anyway, thanks to you for verbose explanation and code. Can't appreciate enough. Thanks a lot again! \$\endgroup\$ – animageofmine Jun 17 '15 at 16:45
  • \$\begingroup\$ Instead of creating the threads and checking for termination can the executor not be used? \$\endgroup\$ – Eashwer ISS ANAND Apr 3 '16 at 10:12

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.