An excellent written test question to assess your Java multi-threading knowledge. Please practice it by doing it yourself.
Q. Review the code shown below and then answer the following questions.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | import java.util.ArrayList; import java.util.ArrayList; import java.util.List; public class WhatIsWrong { private static final int PRODUER_WAIT = 3; private static final int CONSUMER_WAIT = 2; private static boolean isDone = false; public static void main(String[] args) throws InterruptedException { List<Integer> list = new ArrayList<>(); Producer prod = new Producer(list); Consumer cons = new Consumer(list); prod.start(); cons.start(); Thread.sleep(30); isDone = true; } static class Producer extends Thread { private List<Integer> list; public Producer(List<Integer> list) { this.list = list; } public void run() { for (int i = 0; i < 10000; i++) { list.add(i); try { Thread.sleep(PRODUER_WAIT); } catch (InterruptedException e) { e.printStackTrace(); } } } } static class Consumer extends Thread { private List<Integer> list; public Consumer(List<Integer> list) { this.list = list; } public void run() { while (!isDone) { if (list.size() < 10) { try { Thread.sleep(CONSUMER_WAIT); } catch (InterruptedException e) { e.printStackTrace(); } } else { List<Integer> batch = new ArrayList<>(); for (Integer i : list) { batch.add(i); list.remove(i); } consume(batch); } } } private void consume(List<Integer> batch) { System.out.println("Printing numbers....."); System.out.println(batch); } } } |
Q2. What are the issues with the above code?
Q3. How will you go about fixing it?
Q4. How will you convert the polling based solution to event based solution?
A1. The above code tries to create 10,000 integers from 0 to 9,999, and then batches them into sizes of 10 or more.
- The main thread spawns 2 new threads “Producer” and “Consumer“
- Both threads share the same “list“
- The “Producer” add numbers from 0 to 9,999 to the list, and the “Consumer” remove those numbers for consumption (i.e. printing) in batches.
A2. The issues with the above code are
#1. java.util.ConcurrentModificationException is thrown.
1 2 3 4 5 | Exception in thread "Thread-2" java.util.ConcurrentModificationException at java.util.AbstractList$Itr.checkForComodification(AbstractList.java:372) at java.util.AbstractList$Itr.next(AbstractList.java:343) at WhatIsWrong$Consumer.run(WhatIsWrong.java:58) |
Learn more: https://www.java-success.com/dealing-concurrent-modifications-java/
#2. After 30ms the isDone value will be set to “true”, hence the consume thread comes out of the while(!isDone) loop, and completes even before processing all the numbers in the list. The “Producer” being a user or non daemon thread continues to add all the numbers even the “Consumer” thread exit. Once the “Producer” thread is done, the JVM exits.
A3. Here is the code that fixes the 2 issues mentioned above. This code will print a desired output, but still will not be correct, which you will see later that it may hang without completing.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | import java.util.ArrayList; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; public class WhatIsWrong { private static final int PRODUER_WAIT = 3; private static final int CONSUMER_WAIT = 2; private static volatile boolean isDone = false; // volatile for the isDone to be visible to // the consumer thread. public static void main(String[] args) throws InterruptedException { List<Integer> list = new CopyOnWriteArrayList<>(); // to fix java.util.ConcurrentModificationException Producer prod = new Producer(list); Consumer cons = new Consumer(list); prod.start(); cons.start(); prod.join(); // main thread waits for the Producer thread to finish if (list.size() == 0) { // condition for the consumer thread to exit loop if it has processed all isDone = true; } } static class Producer extends Thread { private List<Integer> list; public Producer(List<Integer> list) { this.list = list; } public void run() { for (int i = 0; i < 10000; i++) { list.add(i); try { Thread.sleep(PRODUER_WAIT); } catch (InterruptedException e) { e.printStackTrace(); } } } } static class Consumer extends Thread { private List<Integer> list; public Consumer(List<Integer> list) { this.list = list; } public void run() { while (!isDone) { if (list.size() < 10) { try { Thread.sleep(CONSUMER_WAIT); } catch (InterruptedException e) { e.printStackTrace(); } } else { List<Integer> batch = new ArrayList<>(); for (Integer i : list) { batch.add(i); list.remove(i); } consume(batch); } } } private void consume(List<Integer> batch) { System.out.println("Printing numbers....."); System.out.println(batch); } } } |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 | Printing numbers..... [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] Printing numbers..... [10, 11, 12, 13, 14, 15, 16, 17, 18, 19] Printing numbers..... [20, 21, 22, 23, 24, 25, 26, 27, 28, 29] Printing numbers..... [30, 31, 32, 33, 34, 35, 36, 37, 38, 39] Printing numbers..... [40, 41, 42, 43, 44, 45, 46, 47, 48, 49] Printing numbers..... [50, 51, 52, 53, 54, 55, 56, 57, 58, 59] Printing numbers..... [60, 61, 62, 63, 64, 65, 66, 67, 68, 69] .......... |
1 2 3 | private static final int PRODUER_WAIT = 0; private static final int CONSUMER_WAIT = 0; |
- one or more threads sits waiting for a signal;
- another thread comes along and notifies the waiting threads (i.e. “wakes it/them up” with the signal).
Event driven solution with wait/notify
The wait/notify needs to be within a lock. The block level lock is placed on the “list“. So, the “list” is synchronized, and wait/notify also invoked on the “list”.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | import java.util.ArrayList; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; public class WhatIsWrong { // PRODUCER/CONSUMER times are removed private static volatile boolean isDone = false; public static void main(String[] args) throws InterruptedException { long t1 = System.currentTimeMillis(); List<Integer> list = new CopyOnWriteArrayList<>(); //to fix java.util.ConcurrentModificationException Producer prod = new Producer(list); Consumer cons = new Consumer(list); prod.start(); cons.start(); prod.join(); // wait for the Producer to finish synchronized (list) { if (list.size() == 0) { isDone = true; } } long t2 = System.currentTimeMillis(); System.out.println("Elapsed time in ms = " + (t2 - t1)); } static class Producer extends Thread { private List<Integer> list; public Producer(List<Integer> list) { this.list = list; } public void run() { for (int i = 0; i < 1000000; i++) { // System.out.println("adding " + list.size()); synchronized (list) { //acquire lock list.add(i); //System.out.println("added " + i); try { list.wait(); // give the lock to the consumer and wait } catch (InterruptedException e) { e.printStackTrace(); } } } } } static class Consumer extends Thread { private List<Integer> list; public Consumer(List<Integer> list) { this.list = list; } public void run() { while (!isDone) { synchronized (list) { //acquire lock if (list.size() >= 10) { //System.out.println("10 or more" + list.size()); List<Integer> batch = new ArrayList<>(); for (Integer i : list) { batch.add(i); list.remove(i); // CopyOnWriteArrayList won't throw // ConcurrentModificationException } consume(batch); } list.notifyAll(); // notify the producer & relinquish the lock } } } private void consume(List<Integer> batch) { System.out.println("Printing numbers....."); System.out.println(batch); } } } |
output:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 | ..... Printing numbers..... [999920, 999921, 999922, 999923, 999924, 999925, 999926, 999927, 999928, 999929] Printing numbers..... [999930, 999931, 999932, 999933, 999934, 999935, 999936, 999937, 999938, 999939] Printing numbers..... [999940, 999941, 999942, 999943, 999944, 999945, 999946, 999947, 999948, 999949] Printing numbers..... [999950, 999951, 999952, 999953, 999954, 999955, 999956, 999957, 999958, 999959] Printing numbers..... [999960, 999961, 999962, 999963, 999964, 999965, 999966, 999967, 999968, 999969] Printing numbers..... [999970, 999971, 999972, 999973, 999974, 999975, 999976, 999977, 999978, 999979] Printing numbers..... [999980, 999981, 999982, 999983, 999984, 999985, 999986, 999987, 999988, 999989] Printing numbers..... [999990, 999991, 999992, 999993, 999994, 999995, 999996, 999997, 999998, 999999] Elapsed time in ms = 22062 |
Above code works, but inefficient. Why?
As “CopyOnWriteArrayList” is an enhanced version of ArrayList in which all modifications (add, set, remove, etc) are implemented by making a fresh copy, which is inefficient when used in loops. So, let’s use iterator.remove() as the underlying “list” is already synchronized.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | import java.util.ArrayList; import java.util.Iterator; import java.util.List; public class WhatIsWrong { // PRODUCER/CONSUMER times are removed private static volatile boolean isDone = false; public static void main(String[] args) throws InterruptedException { long t1 = System.currentTimeMillis(); List<Integer> list = new ArrayList<>(); //as "CopyOnWriteArrayList" is an enhanced version of //ArrayList in which all modifications (add, set, remove, etc) //are implemented <strong>by making a fresh copy</strong>, //which is inefficient when used in loops. Producer prod = new Producer(list); Consumer cons = new Consumer(list); prod.start(); cons.start(); prod.join(); // wait for the Producer to finish synchronized (list) { if (list.size() == 0) { isDone = true; } } long t2 = System.currentTimeMillis(); System.out.println("Elapsed time in ms = " + (t2 - t1)); } static class Producer extends Thread { private List<Integer> list; public Producer(List<Integer> list) { this.list = list; } public void run() { for (int i = 0; i < 1000000; i++) { // System.out.println("adding " + list.size()); synchronized (list) { //acquire lock list.add(i); //System.out.println("added " + i); try { list.wait(); // give the lock to the consumer and wait } catch (InterruptedException e) { e.printStackTrace(); } } } } } static class Consumer extends Thread { private List<Integer> list; public Consumer(List<Integer> list) { this.list = list; } public void run() { while (!isDone) { synchronized (list) { //acquire lock if (list.size() >= 10) { List<Integer> batch = new ArrayList<>(); // Iterator won't throw ConcurrentModificationException Iterator<Integer> it = list.iterator(); while(it.hasNext()) { batch.add(it.next()); it.remove(); } consume(batch); } list.notifyAll(); // notify the producer & relinquish the lock } } } private void consume(List<Integer> batch) { System.out.println("Printing numbers....."); System.out.println(batch); } } } |
output:
1 2 3 4 5 6 7 8 9 10 11 12 | ..... [999950, 999951, 999952, 999953, 999954, 999955, 999956, 999957, 999958, 999959] Printing numbers..... [999960, 999961, 999962, 999963, 999964, 999965, 999966, 999967, 999968, 999969] Printing numbers..... [999970, 999971, 999972, 999973, 999974, 999975, 999976, 999977, 999978, 999979] Printing numbers..... [999980, 999981, 999982, 999983, 999984, 999985, 999986, 999987, 999988, 999989] Printing numbers..... [999990, 999991, 999992, 999993, 999994, 999995, 999996, 999997, 999998, 999999] Elapsed time in ms = 21020 |
