Muraveva_A_B_Synchronizer

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

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

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

@@ -26,11 +26,13 @@ public void attachMonitor(StreamingMonitor monitor) {
 
     @Override
     public void run() {
-        // Writer threads are intentionally infinite for the task contract.
         while (true) {
+            if (!monitor.waitTurn(id)) {
+                break;
+            }
             output.print(message);
             onTick.run();
+            monitor.tickDone();
         }
     }
-
-}
+}

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

@@ -1,5 +1,45 @@
 package hse.java.lectures.lecture6.tasks.synchronizer;
 
 public class StreamingMonitor {
-    // impl your sync here
-}
+
+    private final int[] sortedIds;
+    private final int targetTicks;
+    private int currentIdx = 0;
+    private int completedTicks = 0;
+
+    public StreamingMonitor(int[] sortedIds, int targetTicks) {
+        this.sortedIds = sortedIds;
+        this.targetTicks = targetTicks;
+    }
+
+    public synchronized boolean waitTurn(int id) {
+        try {
+            while (completedTicks < targetTicks && sortedIds[currentIdx] != id) {
+                wait();
+            }
+            return completedTicks < targetTicks;
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return false;
+        }
+    }
+
+    public synchronized void tickDone() {
+        currentIdx++;
+        if (currentIdx == sortedIds.length) {
+            currentIdx = 0;
+            completedTicks++;
+        }
+        notifyAll();
+    }
+
+    public synchronized void awaitDone() {
+        try {
+            while (completedTicks < targetTicks) {
+                wait();
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}

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

@@ -17,17 +17,24 @@ public Synchronizer(List<StreamWriter> tasks, int ticksPerWriter) {
         this.ticksPerWriter = ticksPerWriter;
     }
 
-    /**
-     * Starts infinite writer threads and waits until each writer prints exactly ticksPerWriter ticks
-     * in strict ascending id order.
-     */
     public void execute() {
-        // add monitor and sync
+        int[] sortedIds = tasks.stream()
+                .mapToInt(StreamWriter::getId)
+                .sorted()
+                .toArray();
+
+        StreamingMonitor monitor = new StreamingMonitor(sortedIds, ticksPerWriter);
+
+        for (StreamWriter writer : tasks) {
+            writer.attachMonitor(monitor);
+        }
+
         for (StreamWriter writer : tasks) {
             Thread worker = new Thread(writer, "stream-writer-" + writer.getId());
             worker.setDaemon(true);
             worker.start();
         }
-    }
 
-}
+        monitor.awaitDone();
+    }
+}