Hi, could you please help me out with this programming assignment. It needs to b

Question

Hi, could you please help me out with this programming assignment. It needs to be done in Java and it would be very helpful if you could add comments.

Description
To handle a large amount of data it has been decided to filter the data into separate files.
The data is larger than the local storage capacity so all the data cannot be handled at once.
In this assignment we are only interested in a concurrent implementation of this filtering.
So a number of simplifications have been made to the above scenario.
• The data is assumed to be integers.
• The size of the local storage capacity is m (below m is chosen to be 4).
• The size of the data is n (below n is chosen to be 100m = 400).
• Any storage needed to handle concurrency correctly is assumed to be negligible
compared to n and m, so this is disregarded.
• The filtering is into k files (below k is chosen to be 2).
[Clearly, in a real application m and n would be several magnitudes larger.]

Task
Develop a multithreaded program with three concurrent threads, A, B and C. Thread A
generates n random integer numbers and writes them into a shared memory with a capacity
of m integers. Due to the limited capacity of the shared memory, thread A cannot write
all numbers at once. Threads B and C read the integer numbers from the shared memory.
Thread B writes the even numbers to th file even-numbers and thread C writes the odd
numbers to the file odd-numbers. Every number generated by thread A should be written
in one of the two files. No number should be written twice. Choose any method, or
combination of methods, for synchronization.
To achieve full marks:
• Encapsulate the synchronized methods and the shared memory into one structure.
• Avoid busy waiting.
• Avoid static variables.

Extension task
Write code that could easily be modified to cater for different values of m, n, and k above.
For k = 3 you may consider the three classes of odd numbers, even numbers, and multiples
of 3. Note that these are non-disjoint: numbers belonging to more than one class can
be filtered into any of these but should still only appear in one of the output files.

Explanation / Answer

Shared storage is nothing but a wrapper over an array or linked list. Which has all synchronized (thread safe) methods so that threads that are producing and consuming do not interfere each other. Now this Wrapper class is passed to each producer and consumer so that they are only one resource that is shared between them. And they will consume or produce on same storage. So many instances of threads will be calling insertNumber() & removeNumber() methods from this SharedStorage instance. And SharedStorage class will take care of synchronization, wait and notify. Why Consumers don't terminate in this example? Generally threads are used from independent operations or for countinous service. Produer and Consumer problems are contineous services. Generally the producer will be a countenous producer producing some events and consumers will be consuming them and then waiting for more events. In this case we don't know when we have finished all the numbers. Because you can't tell by the size of queue. If size = 0 -> it means there is no item to consume yet. It does not mean that all the numbers are consumed. Below code will do the required things. Plese note that consumers will be running forever as there is no terminating condition for consumers. Terminate the program to see the output files. import java.io.*; import java.util.LinkedList; import java.util.Queue; import java.util.Random; public class MyFilter { public static void main(String args[]) { SharedStorage storage = new SharedStorage(100); int n = 100; // Create and start producer and consumer threads Thread producer = new Thread(new ThreadA(storage, n)); Thread consumer1 = new Thread(new ThreadK(storage, 1)); Thread consumer2 = new Thread(new ThreadK(storage, 2)); Thread consumer3 = new Thread(new ThreadK(storage, 3)); producer.start(); consumer1.start(); consumer2.start(); consumer3.start(); try { producer.join(); consumer1.join(); consumer2.join(); consumer3.join(); } catch (InterruptedException ie) { } } } /** * Producer thread generates numbers up to n */ class ThreadA implements Runnable { private SharedStorage sharedStorage; private Random r = new Random(); private int n; public ThreadA(SharedStorage storage, int n){ sharedStorage = storage; this.n = n; } public void run() { for (int i=0;i= 3) { // Print to multiples of k System.out.println("Consumer3 number"); n = sharedStorage.removeMultipleOfK(k); File yourFile = new File("multipleofk.txt"); yourFile.createNewFile(); // if file already exists will do nothing BufferedWriter bw = new BufferedWriter(new FileWriter(yourFile, true)); bw.write(n + " "); bw.close(); } else if (k == 2) { //Print to even file System.out.println("Consumer2 number"); n = sharedStorage.removeEvenNumber(); File yourFile = new File("even.txt"); yourFile.createNewFile(); // if file already exists will do nothing BufferedWriter bw = new BufferedWriter(new FileWriter(yourFile, true)); bw.write(n + " "); bw.close(); } else if (k ==1 ) { // print to odd file System.out.println("Consumer1 number"); n = sharedStorage.removeOddNumber(); File yourFile = new File("odd.txt"); yourFile.createNewFile(); // if file already exists will do nothing BufferedWriter bw = new BufferedWriter(new FileWriter(yourFile, true)); bw.write(n + " "); bw.close(); } } catch (InterruptedException ie) { } catch (IOException ioe) { } } } } class SharedStorage { private int m; private Queue queue; private int next = 0; public SharedStorage(int capacity) { this.m = capacity; this.queue = new LinkedList(); } public synchronized void insertNumber(int n) { try { while (queue.size() == m) wait(); queue.add(n); notifyAll(); } catch (InterruptedException ie) { } } public synchronized int removeOddNumber() throws InterruptedException { try { while (queue.size() == 0 || (queue.peek()== null || queue.peek() % 2 == 0)) wait(); int n = queue.poll(); notifyAll(); return n; } catch (InterruptedException ie) { throw ie; } } public synchronized int removeEvenNumber() throws InterruptedException { try { while (queue.size() == 0 || (queue.peek()!= null && queue.peek() % 2 == 1)) wait(); int n = queue.poll(); notifyAll(); return n; } catch (InterruptedException ie) { throw ie; } } public synchronized int removeMultipleOfK(int k) throws InterruptedException { try { while (queue.size() == 0 || (queue.peek()!= null && queue.peek() % k != 0)) wait(); int n = queue.poll(); notifyAll(); return n; } catch (InterruptedException ie) { throw ie; } } }

Related Questions

Navigate

• Browse (All)
• Browse H
• Subjects

---

---

Integrity-first tutoring: explanations and feedback only — we do not complete graded work. Learn more.