Yakovleva_M_A_synchronizer

src/main/java/hse/java/lectures/lecture6/tasks/queue/BoundedBlockingQueue.java

@@ -1,25 +1,54 @@
 package hse.java.lectures.lecture6.tasks.queue;
 
+import java.util.LinkedList;
+import java.util.Queue;
+
 public class BoundedBlockingQueue<T> {
 
+    private final Queue<T> q;
+    private final int capacity;
+    private int size;
 
     public BoundedBlockingQueue(int capacity) {
-
+        if (capacity <= 0) {
+            throw new IllegalArgumentException();
+        }
+        this.capacity = capacity;
+        q = new LinkedList<T>();
+        size = 0;
     }
 
     public void put(T item) throws InterruptedException {
-
+        if (item == null) {
+            throw new NullPointerException();
+        }
+        synchronized (this) {
+            while (size == capacity) {
+                wait();
+            }
+            q.add(item);
+            size++;
+            notifyAll();
+        }
     }
 
     public T take() throws InterruptedException {
-        return null;
+        synchronized (this) {
+            while (size == 0) {
+                wait();
+            }
+            T item = q.poll();
+            size--;
+            notifyAll();
+            return item;
+        }
     }
 
-    public int size() {
-        return 0;
+    public synchronized int size() {
+        return size;
     }
 
     public int capacity() {
-        return 0;
+        return capacity;
     }
 }

src/main/java/hse/java/lectures/lecture6/tasks/synchronizer/StreamWriter.java

@@ -28,8 +28,12 @@ public void attachMonitor(StreamingMonitor monitor) {
     public void run() {
         // Writer threads are intentionally infinite for the task contract.
         while (true) {
-            output.print(message);
-            onTick.run();
+            try {
+                if (!monitor.work_tick(id, message, output, onTick)) return;
+            } catch (Exception e) {
+                Thread.currentThread().interrupt();
+                break;
+            }
         }
     }
 

src/main/java/hse/java/lectures/lecture6/tasks/synchronizer/StreamingMonitor.java

@@ -1,5 +1,78 @@
 package hse.java.lectures.lecture6.tasks.synchronizer;
 
+import java.io.PrintStream;
+import java.util.*;
 public class StreamingMonitor {
-    // impl your sync here
+    private final List<StreamWriter> writers;
+    private final int ticks_per_writer;
+
+    private final Map<Integer, Integer> done_ticks = new HashMap<>();
+
+    private int cur_ind;
+    private int cnt_ticks;
+    private final int max_cnt_ticks;
+
+    public StreamingMonitor(List<StreamWriter> writers, int ticks_per_writer) {
+        this.writers = writers;
+        this.ticks_per_writer = ticks_per_writer;
+        this.max_cnt_ticks = writers.size() *  ticks_per_writer;
+
+        for (StreamWriter writer : writers) {
+            done_ticks.put(writer.getId(), 0);
+        }
+        cur_ind = 0;
+        cnt_ticks = 0;
+    }
+
+    private boolean can_run(int id) {
+        if (cnt_ticks >= max_cnt_ticks) {
+            return true;
+        }
+        return writers.get(cur_ind).getId() == id && done_ticks.get(id) < ticks_per_writer;
+    }
+
+    public synchronized void wait_id(int id) throws InterruptedException {
+        while (!can_run(id)) {
+            wait();
+        }
+    }
+
+    private void next() {
+        int n = writers.size();
+
+        for (int i = 1; i <= n; i++) {
+            int next = (cur_ind + i) % n;
+            int nextId = writers.get(next).getId();
+
+            if (done_ticks.get(nextId) < ticks_per_writer) {
+                cur_ind = next;
+                return;
+            }
+        }
+    }
+
+    public synchronized boolean work_tick(int id, String message, PrintStream output, Runnable onTick) throws InterruptedException {
+        while (!can_run(id)) {
+            wait();
+        }
+
+        if (cnt_ticks >= max_cnt_ticks) {
+            notifyAll();
+            return false;
+        }
+        output.print(message);
+        onTick.run();
+        done_ticks.put(id, done_ticks.get(id) + 1);
+        cnt_ticks++;
+        next();
+
+        notifyAll();
+        return cnt_ticks < max_cnt_ticks;
+    }
+
+    public synchronized void wait_all() throws InterruptedException {
+        while (cnt_ticks < max_cnt_ticks) {
+            wait();
+        }
+    }
 }

src/main/java/hse/java/lectures/lecture6/tasks/synchronizer/Synchronizer.java

@@ -1,5 +1,7 @@
 package hse.java.lectures.lecture6.tasks.synchronizer;
 
+import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.List;
 
 public class Synchronizer {
@@ -22,12 +24,26 @@ public Synchronizer(List<StreamWriter> tasks, int ticksPerWriter) {
      * in strict ascending id order.
      */
     public void execute() {
-        // add monitor and sync
-        for (StreamWriter writer : tasks) {
+        List<StreamWriter> sorted = new ArrayList<>(tasks);
+        sorted.sort(Comparator.comparingInt(StreamWriter::getId));
+
+        StreamingMonitor monitor = new StreamingMonitor(sorted, ticksPerWriter);
+
+        for (StreamWriter writer : sorted) {
+            writer.attachMonitor(monitor);
+        }
+
+        for (StreamWriter writer : sorted) {
             Thread worker = new Thread(writer, "stream-writer-" + writer.getId());
             worker.setDaemon(true);
             worker.start();
         }
+
+        try {
+            monitor.wait_all();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
     }
 
 }