Сопильняк Ольга, 496, Threads

projects/pom.xml

@@ -30,6 +30,7 @@
         <module>dkhurtin</module>
         <module>ale3otik</module>
         <module>Pitovsky</module>
+        <module>sopilnyak</module>
     </modules>
 
     <dependencyManagement>

projects/sopilnyak/pom.xml

@@ -0,0 +1,68 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>ru.mipt.diht.students</groupId>
+        <artifactId>parent</artifactId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+
+    <groupId>ru.mipt.diht.students</groupId>
+    <artifactId>sopilnyak</artifactId>
+    <version>1.0-SNAPSHOT</version>
+    <name>sopilnyak</name>
+    <url>http://maven.apache.org</url>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <build>
+        <plugins>
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <descriptorRefs>
+                        <descriptorRef>jar-with-dependencies</descriptorRef>
+                    </descriptorRefs>
+                    <archive>
+                        <manifest>
+                            <mainClass>ru.mipt.diht.students.sopilnyak.moduletests.Counter</mainClass>
+                        </manifest>
+                    </archive>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>3.8.1</version>
+            <scope>test</scope>
+        </dependency>
+
+        <!-- import Twitter library -->
+        <dependency>
+            <groupId>org.twitter4j</groupId>
+            <artifactId>twitter4j-core</artifactId>
+            <version>[4.0,)</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.12</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

projects/sopilnyak/src/main/java/ru/mipt/diht/students/sopilnyak/threads/BlockingQueue.java

@@ -0,0 +1,156 @@
+package ru.mipt.diht.students.sopilnyak.threads;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class BlockingQueue<T> {
+
+    private int maxSize;
+    private Queue<T> queue;
+    private Lock queueLock = new ReentrantLock();
+    private Lock stateChanged = new ReentrantLock();
+    private Condition popWait = stateChanged.newCondition();
+    private Condition pushWait = stateChanged.newCondition();
+
+    BlockingQueue(int inputMaxSize) {
+        this.maxSize = inputMaxSize;
+        queue = new ArrayDeque<>();
+    }
+
+    // push_back to the queue
+    public final void offer(List<T> e) {
+        try {
+            offer(e, 0, false);
+        } catch (InterruptedException ex) {
+            ex.printStackTrace();
+        }
+    }
+
+    // pop_front from the queue
+    public final List<T> take(int n) {
+        try {
+            return take(n, 0, false);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+            return null;
+        }
+    }
+
+    public final void offer(List<T> e, long timeout) throws InterruptedException {
+        offer(e, timeout, true);
+    }
+
+    public final List<T> take(int n, long timeout) throws InterruptedException {
+        return take(n, timeout, true);
+    }
+
+    private long getTimeout(long timeLimit) throws InterruptedException {
+        long currTime = System.currentTimeMillis();
+        if (currTime > timeLimit) {
+            throw new InterruptedException("Timeout");
+        }
+        return timeLimit - currTime;
+    }
+
+    public final void offer(List<T> e, long timeout, boolean needTimeout) throws InterruptedException {
+        long timeLimit = System.currentTimeMillis() + timeout;
+
+        if (needTimeout) {
+            if (!stateChanged.tryLock(getTimeout(timeLimit), TimeUnit.MILLISECONDS)) {
+                throw new InterruptedException("Timeout");
+            }
+        } else {
+            stateChanged.lock();
+        }
+
+        try {
+            boolean added = false;
+            while (!added) {
+                try {
+                    if (needTimeout) {
+                        if (!queueLock.tryLock(getTimeout(timeLimit), TimeUnit.MILLISECONDS)) {
+                            throw new InterruptedException("Timeout");
+                        }
+                    } else {
+                        queueLock.lock();
+                    }
+
+                    if (queue.size() + e.size() <= maxSize) {
+                        queue.addAll(e);
+                        added = true;
+                    }
+                } finally {
+                    queueLock.unlock();
+                }
+
+                if (!added) {
+                    if (needTimeout) {
+                        if (!popWait.await(getTimeout(timeLimit), TimeUnit.NANOSECONDS)) {
+                            throw new InterruptedException("Timeout");
+                        }
+                    } else {
+                        popWait.await();
+                    }
+                }
+            }
+        } finally {
+            pushWait.signalAll();
+            stateChanged.unlock();
+        }
+    }
+
+    public final List<T> take(int n, long timeout, boolean needTimeout) throws InterruptedException {
+        long timeLimit = System.currentTimeMillis() + timeout;
+        if (needTimeout) {
+            if (!stateChanged.tryLock(getTimeout(timeLimit), TimeUnit.MILLISECONDS)) {
+                throw new InterruptedException("Timeout");
+            }
+        } else {
+            stateChanged.lock();
+        }
+
+        try {
+            List answer = new ArrayList<>();
+            while (answer.size() < n) {
+                try {
+                    if (needTimeout) {
+                        if (!queueLock.tryLock(getTimeout(timeLimit), TimeUnit.MILLISECONDS)) {
+                            throw new InterruptedException("Timeout");
+                        }
+                    } else {
+                        queueLock.lock();
+                    }
+                    if (queue.size() >= n) {
+                        for (int i = 0; i < n; i++) {
+                            answer.add(queue.poll());
+                        }
+                    }
+                } finally {
+                    queueLock.unlock();
+                    if (answer.size() == n) {
+                        return answer;
+                    }
+                }
+                if (answer.size() < n) {
+                    if (needTimeout) {
+                        if (!pushWait.await(getTimeout(timeLimit), TimeUnit.NANOSECONDS)) {
+                            throw new InterruptedException("Timeout");
+                        }
+                    } else {
+                        pushWait.await();
+                    }
+                }
+            }
+            return answer;
+        } finally {
+            popWait.signalAll();
+            stateChanged.unlock();
+        }
+    }
+}

projects/sopilnyak/src/main/java/ru/mipt/diht/students/sopilnyak/threads/Counter.java

@@ -0,0 +1,54 @@
+package ru.mipt.diht.students.sopilnyak.threads;
+
+public class Counter {
+
+    private static final Object MONITOR = new Object();
+
+    private static int currentId;
+
+    private static class ThreadCount extends Thread {
+
+        private int id, nextId;
+
+        ThreadCount(int inputId, int inputNextId) {
+            this.id = inputId;
+            this.nextId = inputNextId;
+        }
+
+        @Override
+        public void run() {
+            while (true) {
+                synchronized (MONITOR) {
+                    while (id != currentId) {
+                        try {
+                            MONITOR.wait();
+                        } catch (InterruptedException e) {
+                            e.printStackTrace();
+                        }
+                    }
+                    System.out.println("Thread-" + String.valueOf(id + 1));
+                    currentId = nextId;
+                    MONITOR.notifyAll();
+                }
+            }
+        }
+    }
+
+    public static void main(String[] args) {
+        int n;
+        try {
+            n = Integer.valueOf(args[0]);
+            if (n <= 0) {
+                throw new NumberFormatException();
+            }
+        } catch (Exception e) {
+            System.err.println("Wrong number of threads");
+            return;
+        }
+        currentId = 0;
+        for (int i = 0; i < n; i++) {
+            ThreadCount thread = new ThreadCount(i, (i + 1) % n);
+            thread.start();
+        }
+    }
+}