Mikhalovskii_M_M_synchronizer

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

@@ -26,11 +26,23 @@ public void attachMonitor(StreamingMonitor monitor) {
 
     @Override
     public void run() {
-        // Writer threads are intentionally infinite for the task contract.
         while (true) {
-            output.print(message);
-            onTick.run();
+            try {
+                StreamingMonitor m = monitor;
+                if (m == null) {
+                    throw new IllegalStateException();
+                }
+                boolean allowed = m.awaitTurn(id);
+                if (!allowed) {
+                    return;
+                }
+                output.print(message);
+                onTick.run();
+                m.tickDone(id);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                return;
+            }
         }
     }
-
-}
+}

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

@@ -1,5 +1,80 @@
 package hse.java.lectures.lecture6.tasks.synchronizer;
 
+import java.util.*;
+
 public class StreamingMonitor {
-    // impl your sync here
+
+    private final List<Integer> orderedIds;
+    private final int ticksPerWriter;
+    private final Map<Integer, Integer> ticksDone = new HashMap<>();
+    private int currentIndex = 0;
+    private int totalDone = 0;
+    private final int requiredTotal;
+
+    public StreamingMonitor(List<StreamWriter> writers, int ticksPerWriter) {
+        if (writers == null || writers.isEmpty()) {
+            throw new IllegalArgumentException();
+        }
+        if (ticksPerWriter < 0) {
+            throw new IllegalArgumentException();
+        }
+        this.ticksPerWriter = ticksPerWriter;
+
+        TreeSet<Integer> sorted = new TreeSet<>();
+        for (StreamWriter w : writers) {
+            sorted.add(w.getId());
+        }
+        this.orderedIds = new ArrayList<>(sorted);
+        for (Integer id : orderedIds) {
+            ticksDone.put(id, 0);
+        }
+        this.requiredTotal = orderedIds.size() * ticksPerWriter;
+    }
+
+    public synchronized boolean awaitTurn(int id) throws InterruptedException {
+        while (true) {
+            if (totalDone >= requiredTotal) {
+                return false;
+            }
+            int currentId = orderedIds.get(currentIndex);
+            if (id == currentId && ticksDone.get(id) < ticksPerWriter) {
+                return true;
+            }
+            wait();
+        }
+    }
+
+    public synchronized void tickDone(int id) {
+        if (totalDone >= requiredTotal) {
+            notifyAll();
+            return;
+        }
+
+        int done = ticksDone.get(id) + 1;
+        ticksDone.put(id, done);
+        totalDone++;
+
+        if (totalDone >= requiredTotal) {
+            notifyAll();
+            return;
+        }
+
+        int n = orderedIds.size();
+        for (int step = 1; step <= n; step++) {
+            int idx = (currentIndex + step) % n;
+            int nextId = orderedIds.get(idx);
+            if (ticksDone.get(nextId) < ticksPerWriter) {
+                currentIndex = idx;
+                break;
+            }
+        }
+
+        notifyAll();
+    }
+
+    public synchronized void awaitCompletion() throws InterruptedException {
+        while (totalDone < requiredTotal) {
+            wait();
+        }
+    }
 }

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

@@ -17,17 +17,23 @@ 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
+        StreamingMonitor monitor = new StreamingMonitor(tasks, 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();
         }
-    }
 
-}
+        try {
+            monitor.awaitCompletion();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}