diff --git a/Task_2_1_2/.gitignore b/Task_2_1_2/.gitignore new file mode 100644 index 0000000..b63da45 --- /dev/null +++ b/Task_2_1_2/.gitignore @@ -0,0 +1,42 @@ +.gradle +build/ +!gradle/wrapper/gradle-wrapper.jar +!**/src/main/**/build/ +!**/src/test/**/build/ + +### IntelliJ IDEA ### +.idea/modules.xml +.idea/jarRepositories.xml +.idea/compiler.xml +.idea/libraries/ +*.iws +*.iml +*.ipr +out/ +!**/src/main/**/out/ +!**/src/test/**/out/ + +### Eclipse ### +.apt_generated +.classpath +.factorypath +.project +.settings +.springBeans +.sts4-cache +bin/ +!**/src/main/**/bin/ +!**/src/test/**/bin/ + +### NetBeans ### +/nbproject/private/ +/nbbuild/ +/dist/ +/nbdist/ +/.nb-gradle/ + +### VS Code ### +.vscode/ + +### Mac OS ### +.DS_Store \ No newline at end of file diff --git a/Task_2_1_2/.idea/.gitignore b/Task_2_1_2/.idea/.gitignore new file mode 100644 index 0000000..26d3352 --- /dev/null +++ b/Task_2_1_2/.idea/.gitignore @@ -0,0 +1,3 @@ +# Default ignored files +/shelf/ +/workspace.xml diff --git a/Task_2_1_2/.idea/gradle.xml b/Task_2_1_2/.idea/gradle.xml new file mode 100644 index 0000000..2a65317 --- /dev/null +++ b/Task_2_1_2/.idea/gradle.xml @@ -0,0 +1,17 @@ + + + + + + + \ No newline at end of file diff --git a/Task_2_1_2/.idea/misc.xml b/Task_2_1_2/.idea/misc.xml new file mode 100644 index 0000000..d5d5cd9 --- /dev/null +++ b/Task_2_1_2/.idea/misc.xml @@ -0,0 +1,7 @@ + + + + + + + \ No newline at end of file diff --git a/Task_2_1_2/.idea/vcs.xml b/Task_2_1_2/.idea/vcs.xml new file mode 100644 index 0000000..6c0b863 --- /dev/null +++ b/Task_2_1_2/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/Task_2_1_2/build.gradle b/Task_2_1_2/build.gradle new file mode 100644 index 0000000..3b5b6ce --- /dev/null +++ b/Task_2_1_2/build.gradle @@ -0,0 +1,27 @@ +plugins { + id 'java' + id 'jacoco' +} + +group = 'ru.nsu.nmashkin.task212' +version = '1.0-SNAPSHOT' + +repositories { + mavenCentral() +} + +dependencies { + testImplementation platform('org.junit:junit-bom:5.10.0') + testImplementation 'org.junit.jupiter:junit-jupiter' + testRuntimeOnly 'org.junit.platform:junit-platform-launcher' +} + +test { + useJUnitPlatform() +} + +jacocoTestReport { + reports { + xml.required = true + } +} \ No newline at end of file diff --git a/Task_2_1_2/gradle/wrapper/gradle-wrapper.jar b/Task_2_1_2/gradle/wrapper/gradle-wrapper.jar new file mode 100644 index 0000000..249e583 Binary files /dev/null and b/Task_2_1_2/gradle/wrapper/gradle-wrapper.jar differ diff --git a/Task_2_1_2/gradle/wrapper/gradle-wrapper.properties b/Task_2_1_2/gradle/wrapper/gradle-wrapper.properties new file mode 100644 index 0000000..cf77dd8 --- /dev/null +++ b/Task_2_1_2/gradle/wrapper/gradle-wrapper.properties @@ -0,0 +1,6 @@ +#Sat May 16 12:16:09 NOVT 2026 +distributionBase=GRADLE_USER_HOME +distributionPath=wrapper/dists +distributionUrl=https\://services.gradle.org/distributions/gradle-8.14-bin.zip +zipStoreBase=GRADLE_USER_HOME +zipStorePath=wrapper/dists diff --git a/Task_2_1_2/gradlew b/Task_2_1_2/gradlew new file mode 100644 index 0000000..1b6c787 --- /dev/null +++ b/Task_2_1_2/gradlew @@ -0,0 +1,234 @@ +#!/bin/sh + +# +# Copyright © 2015-2021 the original authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +############################################################################## +# +# Gradle start up script for POSIX generated by Gradle. +# +# Important for running: +# +# (1) You need a POSIX-compliant shell to run this script. If your /bin/sh is +# noncompliant, but you have some other compliant shell such as ksh or +# bash, then to run this script, type that shell name before the whole +# command line, like: +# +# ksh Gradle +# +# Busybox and similar reduced shells will NOT work, because this script +# requires all of these POSIX shell features: +# * functions; +# * expansions «$var», «${var}», «${var:-default}», «${var+SET}», +# «${var#prefix}», «${var%suffix}», and «$( cmd )»; +# * compound commands having a testable exit status, especially «case»; +# * various built-in commands including «command», «set», and «ulimit». +# +# Important for patching: +# +# (2) This script targets any POSIX shell, so it avoids extensions provided +# by Bash, Ksh, etc; in particular arrays are avoided. +# +# The "traditional" practice of packing multiple parameters into a +# space-separated string is a well documented source of bugs and security +# problems, so this is (mostly) avoided, by progressively accumulating +# options in "$@", and eventually passing that to Java. +# +# Where the inherited environment variables (DEFAULT_JVM_OPTS, JAVA_OPTS, +# and GRADLE_OPTS) rely on word-splitting, this is performed explicitly; +# see the in-line comments for details. +# +# There are tweaks for specific operating systems such as AIX, CygWin, +# Darwin, MinGW, and NonStop. +# +# (3) This script is generated from the Groovy template +# https://github.com/gradle/gradle/blob/master/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt +# within the Gradle project. +# +# You can find Gradle at https://github.com/gradle/gradle/. +# +############################################################################## + +# Attempt to set APP_HOME + +# Resolve links: $0 may be a link +app_path=$0 + +# Need this for daisy-chained symlinks. +while + APP_HOME=${app_path%"${app_path##*/}"} # leaves a trailing /; empty if no leading path + [ -h "$app_path" ] +do + ls=$( ls -ld "$app_path" ) + link=${ls#*' -> '} + case $link in #( + /*) app_path=$link ;; #( + *) app_path=$APP_HOME$link ;; + esac +done + +APP_HOME=$( cd "${APP_HOME:-./}" && pwd -P ) || exit + +APP_NAME="Gradle" +APP_BASE_NAME=${0##*/} + +# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. +DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"' + +# Use the maximum available, or set MAX_FD != -1 to use that value. +MAX_FD=maximum + +warn () { + echo "$*" +} >&2 + +die () { + echo + echo "$*" + echo + exit 1 +} >&2 + +# OS specific support (must be 'true' or 'false'). +cygwin=false +msys=false +darwin=false +nonstop=false +case "$( uname )" in #( + CYGWIN* ) cygwin=true ;; #( + Darwin* ) darwin=true ;; #( + MSYS* | MINGW* ) msys=true ;; #( + NONSTOP* ) nonstop=true ;; +esac + +CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar + + +# Determine the Java command to use to start the JVM. +if [ -n "$JAVA_HOME" ] ; then + if [ -x "$JAVA_HOME/jre/sh/java" ] ; then + # IBM's JDK on AIX uses strange locations for the executables + JAVACMD=$JAVA_HOME/jre/sh/java + else + JAVACMD=$JAVA_HOME/bin/java + fi + if [ ! -x "$JAVACMD" ] ; then + die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME + +Please set the JAVA_HOME variable in your environment to match the +location of your Java installation." + fi +else + JAVACMD=java + which java >/dev/null 2>&1 || die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. + +Please set the JAVA_HOME variable in your environment to match the +location of your Java installation." +fi + +# Increase the maximum file descriptors if we can. +if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then + case $MAX_FD in #( + max*) + MAX_FD=$( ulimit -H -n ) || + warn "Could not query maximum file descriptor limit" + esac + case $MAX_FD in #( + '' | soft) :;; #( + *) + ulimit -n "$MAX_FD" || + warn "Could not set maximum file descriptor limit to $MAX_FD" + esac +fi + +# Collect all arguments for the java command, stacking in reverse order: +# * args from the command line +# * the main class name +# * -classpath +# * -D...appname settings +# * --module-path (only if needed) +# * DEFAULT_JVM_OPTS, JAVA_OPTS, and GRADLE_OPTS environment variables. + +# For Cygwin or MSYS, switch paths to Windows format before running java +if "$cygwin" || "$msys" ; then + APP_HOME=$( cygpath --path --mixed "$APP_HOME" ) + CLASSPATH=$( cygpath --path --mixed "$CLASSPATH" ) + + JAVACMD=$( cygpath --unix "$JAVACMD" ) + + # Now convert the arguments - kludge to limit ourselves to /bin/sh + for arg do + if + case $arg in #( + -*) false ;; # don't mess with options #( + /?*) t=${arg#/} t=/${t%%/*} # looks like a POSIX filepath + [ -e "$t" ] ;; #( + *) false ;; + esac + then + arg=$( cygpath --path --ignore --mixed "$arg" ) + fi + # Roll the args list around exactly as many times as the number of + # args, so each arg winds up back in the position where it started, but + # possibly modified. + # + # NB: a `for` loop captures its iteration list before it begins, so + # changing the positional parameters here affects neither the number of + # iterations, nor the values presented in `arg`. + shift # remove old arg + set -- "$@" "$arg" # push replacement arg + done +fi + +# Collect all arguments for the java command; +# * $DEFAULT_JVM_OPTS, $JAVA_OPTS, and $GRADLE_OPTS can contain fragments of +# shell script including quotes and variable substitutions, so put them in +# double quotes to make sure that they get re-expanded; and +# * put everything else in single quotes, so that it's not re-expanded. + +set -- \ + "-Dorg.gradle.appname=$APP_BASE_NAME" \ + -classpath "$CLASSPATH" \ + org.gradle.wrapper.GradleWrapperMain \ + "$@" + +# Use "xargs" to parse quoted args. +# +# With -n1 it outputs one arg per line, with the quotes and backslashes removed. +# +# In Bash we could simply go: +# +# readarray ARGS < <( xargs -n1 <<<"$var" ) && +# set -- "${ARGS[@]}" "$@" +# +# but POSIX shell has neither arrays nor command substitution, so instead we +# post-process each arg (as a line of input to sed) to backslash-escape any +# character that might be a shell metacharacter, then use eval to reverse +# that process (while maintaining the separation between arguments), and wrap +# the whole thing up as a single "set" statement. +# +# This will of course break if any of these variables contains a newline or +# an unmatched quote. +# + +eval "set -- $( + printf '%s\n' "$DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS" | + xargs -n1 | + sed ' s~[^-[:alnum:]+,./:=@_]~\\&~g; ' | + tr '\n' ' ' + )" '"$@"' + +exec "$JAVACMD" "$@" diff --git a/Task_2_1_2/gradlew.bat b/Task_2_1_2/gradlew.bat new file mode 100644 index 0000000..107acd3 --- /dev/null +++ b/Task_2_1_2/gradlew.bat @@ -0,0 +1,89 @@ +@rem +@rem Copyright 2015 the original author or authors. +@rem +@rem Licensed under the Apache License, Version 2.0 (the "License"); +@rem you may not use this file except in compliance with the License. +@rem You may obtain a copy of the License at +@rem +@rem https://www.apache.org/licenses/LICENSE-2.0 +@rem +@rem Unless required by applicable law or agreed to in writing, software +@rem distributed under the License is distributed on an "AS IS" BASIS, +@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +@rem See the License for the specific language governing permissions and +@rem limitations under the License. +@rem + +@if "%DEBUG%" == "" @echo off +@rem ########################################################################## +@rem +@rem Gradle startup script for Windows +@rem +@rem ########################################################################## + +@rem Set local scope for the variables with windows NT shell +if "%OS%"=="Windows_NT" setlocal + +set DIRNAME=%~dp0 +if "%DIRNAME%" == "" set DIRNAME=. +set APP_BASE_NAME=%~n0 +set APP_HOME=%DIRNAME% + +@rem Resolve any "." and ".." in APP_HOME to make it shorter. +for %%i in ("%APP_HOME%") do set APP_HOME=%%~fi + +@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. +set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m" + +@rem Find java.exe +if defined JAVA_HOME goto findJavaFromJavaHome + +set JAVA_EXE=java.exe +%JAVA_EXE% -version >NUL 2>&1 +if "%ERRORLEVEL%" == "0" goto execute + +echo. +echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. +echo. +echo Please set the JAVA_HOME variable in your environment to match the +echo location of your Java installation. + +goto fail + +:findJavaFromJavaHome +set JAVA_HOME=%JAVA_HOME:"=% +set JAVA_EXE=%JAVA_HOME%/bin/java.exe + +if exist "%JAVA_EXE%" goto execute + +echo. +echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME% +echo. +echo Please set the JAVA_HOME variable in your environment to match the +echo location of your Java installation. + +goto fail + +:execute +@rem Setup the command line + +set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar + + +@rem Execute Gradle +"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %* + +:end +@rem End local scope for the variables with windows NT shell +if "%ERRORLEVEL%"=="0" goto mainEnd + +:fail +rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of +rem the _cmd.exe /c_ return code! +if not "" == "%GRADLE_EXIT_CONSOLE%" exit 1 +exit /b 1 + +:mainEnd +if "%OS%"=="Windows_NT" endlocal + +:omega diff --git a/Task_2_1_2/settings.gradle b/Task_2_1_2/settings.gradle new file mode 100644 index 0000000..891b549 --- /dev/null +++ b/Task_2_1_2/settings.gradle @@ -0,0 +1 @@ +rootProject.name = 'Task_2_1_2' \ No newline at end of file diff --git a/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/Master.java b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/Master.java new file mode 100644 index 0000000..acf2887 --- /dev/null +++ b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/Master.java @@ -0,0 +1,217 @@ +package ru.nsu.nmashkin.task212; + +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.net.DatagramPacket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.MulticastSocket; +import java.net.NetworkInterface; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketException; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Master of Slaves. + */ +public class Master { + private final int port; + private final BlockingQueue taskQueue = new LinkedBlockingQueue<>(); + private final AtomicBoolean globalResult = new AtomicBoolean(false); + private final AtomicBoolean isDone = new AtomicBoolean(false); + private final List workerHandlers = new CopyOnWriteArrayList<>(); + private final Object lock = new Object(); + private volatile MulticastSocket discoverySocket; + private int tasksLeft; + + private static final String MULTICAST_ADDRESS = "230.0.0.1"; + private static final int MULTICAST_PORT = 4446; + + /** + * Make a master. + * + * @param port to listen to. + */ + public Master(int port) { + this.port = port; + } + + /** + * Accept slaves, give them work, return result. + */ + public boolean execute(int[] numbers, int chunkSize, int expectedWorkers) { + int taskId = 0; + for (int i = 0; i < numbers.length; i += chunkSize) { + int end = Math.min(i + chunkSize, numbers.length); + int[] chunk = Arrays.copyOfRange(numbers, i, end); + taskQueue.add(new Task(taskId++, chunk)); + } + + tasksLeft = taskQueue.size(); + + System.out.println("Master is up and expecting " + expectedWorkers + " slaves..."); + + Thread discoveryThread = new Thread(this::searchForSlaves); + discoveryThread.setDaemon(true); + discoveryThread.start(); + + try (ServerSocket serverSocket = new ServerSocket(port)) { + Socket[] workerSockets = new Socket[expectedWorkers]; + for (int i = 0; i < expectedWorkers; i++) { + try { + workerSockets[i] = serverSocket.accept(); + System.out.println("Slave detected: " + + workerSockets[i].getRemoteSocketAddress()); + } catch (IOException e) { + System.out.println("Not enough slaves: " + e.getMessage()); + break; + } + } + + System.out.println("All slaves are ready, " + taskQueue.size() + " works to be done"); + + for (var sock : workerSockets) { + if (sock != null) { + Thread handler = new Thread(new WorkerHandler(sock)); + workerHandlers.add(handler); + handler.start(); + } + } + + synchronized (lock) { + while (tasksLeft > 0 && !globalResult.get()) { + lock.wait(); + } + } + + } catch (IOException | InterruptedException e) { + System.err.println("Error: " + e.getMessage()); + } finally { + isDone.set(true); + for (Thread t : workerHandlers) { + t.interrupt(); + } + + if (discoverySocket != null && !discoverySocket.isClosed()) { + discoverySocket.close(); + } + } + + return globalResult.get(); + } + + private void searchForSlaves() { + try { + discoverySocket = new MulticastSocket(MULTICAST_PORT); + InetAddress group = InetAddress.getByName(MULTICAST_ADDRESS); + NetworkInterface netIf = NetworkInterface.getByInetAddress(InetAddress.getLocalHost()); + if (netIf == null) { + netIf = NetworkInterface.getNetworkInterfaces().nextElement(); + } + discoverySocket.joinGroup(new InetSocketAddress(group, MULTICAST_PORT), netIf); + + byte[] msgBytes = ("MASTER_START:" + port).getBytes(StandardCharsets.UTF_8); + DatagramPacket notification = new DatagramPacket(msgBytes, msgBytes.length, + group, MULTICAST_PORT); + discoverySocket.send(notification); + + byte[] buffer = new byte[1024]; + while (!isDone.get()) { + DatagramPacket packet = new DatagramPacket(buffer, buffer.length); + discoverySocket.receive(packet); + String message = new String(packet.getData(), 0, + packet.getLength(), StandardCharsets.UTF_8); + + if ("SLAVE_READY".equals(message)) { + byte[] responseBytes = ("MASTER_INFO:" + port).getBytes(StandardCharsets.UTF_8); + DatagramPacket response = new DatagramPacket( + responseBytes, responseBytes.length, + group, MULTICAST_PORT + ); + discoverySocket.send(response); + } + } + } catch (SocketException e) { + System.out.println("Search ended"); + } catch (Exception e) { + System.err.println("We're screwed: " + e.getMessage()); + } finally { + if (discoverySocket != null && !discoverySocket.isClosed()) { + discoverySocket.close(); + } + } + } + + private class WorkerHandler implements Runnable { + private final Socket socket; + + public WorkerHandler(Socket socket) { + this.socket = socket; + } + + @Override + public void run() { + Task currentTask = null; + try (Socket s = socket; + ObjectOutputStream out = new ObjectOutputStream(s.getOutputStream()); + ObjectInputStream in = new ObjectInputStream(s.getInputStream())) { + + while (!isDone.get() && !globalResult.get()) { + currentTask = taskQueue.poll(1, TimeUnit.SECONDS); + if (currentTask == null) { + if (taskQueue.isEmpty()) { + break; + } + continue; + } + + out.writeObject(currentTask); + out.flush(); + + TaskResult result = (TaskResult) in.readObject(); + + synchronized (lock) { + if (result.hasNonPrime()) { + globalResult.set(true); + } + System.out.println("Slave has completed work #" + result.taskId() + + " with result " + result.hasNonPrime()); + tasksLeft--; + + if (result.hasNonPrime() || tasksLeft == 0) { + lock.notifyAll(); + } + } + + if (!s.isClosed()) { + currentTask = null; + } + } + + if (!s.isClosed()) { + out.writeObject(null); + out.flush(); + } + + } catch (Exception e) { + System.err.println("Slave is dead: " + socket.getRemoteSocketAddress()); + } finally { + System.out.println(currentTask); + if (currentTask != null) { + System.out.println("Sun is still up in the sky, returning task to queue: " + + currentTask.taskId()); + taskQueue.add(currentTask); + } + } + } + } +} \ No newline at end of file diff --git a/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/MultiThreaded.java b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/MultiThreaded.java new file mode 100644 index 0000000..551ceee --- /dev/null +++ b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/MultiThreaded.java @@ -0,0 +1,50 @@ +package ru.nsu.nmashkin.task212; + +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Multithreaded implementation for finding a non prime. + */ +public class MultiThreaded { + /** + * Main method. + * + * @param numbers array of numbers + * @return whether there is a non prime among numbers + */ + public static boolean hasNonPrime(int[] numbers, int threadCount) { + AtomicBoolean result = new AtomicBoolean(false); + Thread[] threads = new Thread[threadCount]; + int chunkSize = (int) Math.ceil((double) numbers.length / threadCount); + + for (int i = 0; i < threadCount; i++) { + final int start = i * chunkSize; + final int end = Math.min(start + chunkSize, numbers.length); + + threads[i] = new Thread(() -> { + for (int j = start; j < end; j++) { + if (result.get()) { + break; + } + if (!PrimeChecker.isPrime(numbers[j])) { + result.set(true); + break; + } + } + }); + threads[i].start(); + } + + for (Thread thread : threads) { + try { + thread.join(); + } catch (InterruptedException e) { + System.err.println("MultiThreaded.hasNonPrime(): Could not join thread"); + } + } + + return result.get(); + + } + +} \ No newline at end of file diff --git a/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/PrimeChecker.java b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/PrimeChecker.java new file mode 100644 index 0000000..b738dec --- /dev/null +++ b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/PrimeChecker.java @@ -0,0 +1,28 @@ +package ru.nsu.nmashkin.task212; + +/** + * Check if the number is prime. + */ +public class PrimeChecker { + /** + * Check if the number is prime. + * + * @param n number + * @return is n prime + */ + public static boolean isPrime(int n) { + if (n <= 1) { + return false; + } + if (n == 2) { + return true; + } + + for (int i = 2; i * i <= n; i++) { + if (n % i == 0) { + return false; + } + } + return true; + } +} \ No newline at end of file diff --git a/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/Slave.java b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/Slave.java new file mode 100644 index 0000000..80641a3 --- /dev/null +++ b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/Slave.java @@ -0,0 +1,127 @@ +package ru.nsu.nmashkin.task212; + +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.net.ConnectException; +import java.net.DatagramPacket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.MulticastSocket; +import java.net.NetworkInterface; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.nio.charset.StandardCharsets; + +/** + * Slave for 300$. + */ +public record Slave(int threadCount) { + private static final String MULTICAST_ADDRESS = "230.0.0.1"; + private static final int MULTICAST_PORT = 4446; + + /** + * Start the work day. + */ + public void start() { + System.out.println("Slave ready. Initiating Master discovery..."); + + while (true) { + InetSocketAddress masterAddress = discoverMaster(); + if (masterAddress == null) { + System.out.println("Discovery failed. Retrying..."); + continue; + } + + try { + System.out.println("Connecting to discovered Master at " + masterAddress); + + try (Socket socket = new Socket(masterAddress.getAddress(), + masterAddress.getPort()); + ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream()); + ObjectInputStream in = new ObjectInputStream(socket.getInputStream())) { + + System.out.println("Master detected"); + + while (true) { + Object input = in.readObject(); + if (input == null) { + System.out.println("Work is over"); + return; + } + + Task task = (Task) input; + System.out.println("Work received: #" + task.taskId() + + ", size: " + task.numbers().length); + + boolean result = MultiThreaded.hasNonPrime(task.numbers(), threadCount); + + TaskResult taskResult = new TaskResult(task.taskId(), result); + out.writeObject(taskResult); + out.flush(); + System.out.println("Work #" + task.taskId() + + " is done, result: " + result); + } + } + } catch (ConnectException e) { + System.err.println("Master found but unavailable (Connection refused)."); + } catch (IOException | ClassNotFoundException e) { + System.err.println("Connection with Master broken: " + e.getMessage()); + System.out.println("Work is over"); + return; + } + + try { + Thread.sleep(3000); + } catch (InterruptedException ie) { + System.out.println("Slave interrupted, stopping loop."); + Thread.currentThread().interrupt(); + break; + } + } + } + + private InetSocketAddress discoverMaster() { + try (MulticastSocket multicastSocket = new MulticastSocket(MULTICAST_PORT)) { + InetAddress group = InetAddress.getByName(MULTICAST_ADDRESS); + NetworkInterface netIf = NetworkInterface.getByInetAddress(InetAddress.getLocalHost()); + if (netIf == null) { + netIf = NetworkInterface.getNetworkInterfaces().nextElement(); + } + multicastSocket.joinGroup(new InetSocketAddress(group, MULTICAST_PORT), netIf); + + byte[] msgBytes = "SLAVE_READY".getBytes(StandardCharsets.UTF_8); + DatagramPacket readyPacket = new DatagramPacket(msgBytes, msgBytes.length, + group, MULTICAST_PORT); + multicastSocket.send(readyPacket); + multicastSocket.setSoTimeout(3000); + + byte[] buffer = new byte[1024]; + while (true) { + try { + DatagramPacket packet = new DatagramPacket(buffer, buffer.length); + multicastSocket.receive(packet); + String response = new String(packet.getData(), 0, + packet.getLength(), StandardCharsets.UTF_8); + + if (response.startsWith("MASTER_INFO:") + || response.startsWith("MASTER_START:")) { + int port = Integer.parseInt(response.split(":")[1]); + InetAddress masterIp = packet.getAddress(); + + if (masterIp.isLoopbackAddress() || masterIp.isAnyLocalAddress()) { + masterIp = InetAddress.getLocalHost(); + } + return new InetSocketAddress(masterIp, port); + } + } catch (SocketTimeoutException e) { + System.out.println("No Master answered. Re-broadcasting/waiting..."); + multicastSocket.send(readyPacket); + } + } + } catch (Exception e) { + System.err.println("Discovery exception: " + e.getMessage()); + return null; + } + } +} \ No newline at end of file diff --git a/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/Task.java b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/Task.java new file mode 100644 index 0000000..b5fd0db --- /dev/null +++ b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/Task.java @@ -0,0 +1,15 @@ +package ru.nsu.nmashkin.task212; + +import java.io.Serial; +import java.io.Serializable; + +/** + * Slave's work. + * + * @param taskId . + * @param numbers . + */ +public record Task(int taskId, int[] numbers) implements Serializable { + @Serial + private static final long serialVersionUID = 1L; +} \ No newline at end of file diff --git a/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/TaskResult.java b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/TaskResult.java new file mode 100644 index 0000000..57aa70b --- /dev/null +++ b/Task_2_1_2/src/main/java/ru/nsu/nmashkin/task212/TaskResult.java @@ -0,0 +1,15 @@ +package ru.nsu.nmashkin.task212; + +import java.io.Serial; +import java.io.Serializable; + +/** + * Slave's work report. + * + * @param taskId . + * @param hasNonPrime . + */ +public record TaskResult(int taskId, boolean hasNonPrime) implements Serializable { + @Serial + private static final long serialVersionUID = 1L; +} \ No newline at end of file diff --git a/Task_2_1_2/src/test/java/ru/nsu/nmashkin/task212/EndToEndTest.java b/Task_2_1_2/src/test/java/ru/nsu/nmashkin/task212/EndToEndTest.java new file mode 100644 index 0000000..29f9b28 --- /dev/null +++ b/Task_2_1_2/src/test/java/ru/nsu/nmashkin/task212/EndToEndTest.java @@ -0,0 +1,45 @@ +package ru.nsu.nmashkin.task212; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; + +class EndToEndTest { + + @Test + public void e2e() { + int port = 8004; + int[] numbers = {11, 13, 17, 19, 23, 29}; + Master master = new Master(port); + + Thread slaveThread = new Thread(() -> { + Slave slave = new Slave(2); + slave.start(); + }); + slaveThread.start(); + + boolean result = master.execute(numbers, 2, 1); + assertFalse(result); + + slaveThread.interrupt(); + } + + @Test + public void e2e_allPrime() { + int port = 8005; + int[] numbers = {11, 13, 15, 17, 19}; + Master master = new Master(port); + + Thread slaveThread = new Thread(() -> { + Slave slave = new Slave(2); + slave.start(); + }); + slaveThread.start(); + + boolean result = master.execute(numbers, 2, 1); + assertTrue(result); + + slaveThread.interrupt(); + } +} \ No newline at end of file diff --git a/Task_2_1_2/src/test/java/ru/nsu/nmashkin/task212/MasterTest.java b/Task_2_1_2/src/test/java/ru/nsu/nmashkin/task212/MasterTest.java new file mode 100644 index 0000000..1a055da --- /dev/null +++ b/Task_2_1_2/src/test/java/ru/nsu/nmashkin/task212/MasterTest.java @@ -0,0 +1,230 @@ +package ru.nsu.nmashkin.task212; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.net.DatagramPacket; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.MulticastSocket; +import java.net.NetworkInterface; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.jupiter.api.Test; + +class MasterTest { + + @Test + public void execute_allPrime() throws Exception { + int port = 8011; + int[] numbers = {2, 3, 5, 7, 11, 13, 17, 19}; + int chunkSize = 3; + int expectedWorkers = 1; + + Master master = new Master(port); + + Thread slaveSimulator = new Thread(() -> { + try { + Thread.sleep(200); + try (Socket socket = new Socket("localhost", port); + ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream()); + ObjectInputStream in = new ObjectInputStream(socket.getInputStream())) { + + while (true) { + Object obj = in.readObject(); + if (obj == null) { + break; + } + Task task = (Task) obj; + out.writeObject(new TaskResult(task.taskId(), false)); + out.flush(); + } + } + } catch (Exception e) { + e.printStackTrace(); + } + }); + slaveSimulator.start(); + + boolean result = master.execute(numbers, chunkSize, expectedWorkers); + assertFalse(result); + slaveSimulator.join(); + } + + @Test + public void execute() throws Exception { + int port = 8012; + int[] numbers = {2, 3, 4, 5, 7}; + int chunkSize = 2; + int expectedWorkers = 1; + + Master master = new Master(port); + + Thread slaveSimulator = new Thread(() -> { + try { + Thread.sleep(200); + try (Socket socket = new Socket("localhost", port); + ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream()); + ObjectInputStream in = new ObjectInputStream(socket.getInputStream())) { + + while (true) { + Object obj = in.readObject(); + if (obj == null) { + break; + } + Task task = (Task) obj; + boolean hasNonPrime = false; + for (int n : task.numbers()) { + if (n == 4) { + hasNonPrime = true; + break; + } + } + out.writeObject(new TaskResult(task.taskId(), hasNonPrime)); + out.flush(); + } + } + } catch (Exception e) { + e.printStackTrace(); + } + }); + slaveSimulator.start(); + + boolean result = master.execute(numbers, chunkSize, expectedWorkers); + assertTrue(result); + slaveSimulator.join(); + } + + @Test + public void masterRespondsToSlaveReady() throws Exception { + int port = 8030; + Master master = new Master(port); + AtomicBoolean responseReceived = new AtomicBoolean(false); + + Thread slaveSimulator = new Thread(() -> { + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + + try { + try (MulticastSocket multicastSocket = new MulticastSocket(4446)) { + InetAddress group = InetAddress.getByName("230.0.0.1"); + NetworkInterface netIf = NetworkInterface.getByInetAddress( + InetAddress.getLocalHost()); + if (netIf == null) { + netIf = NetworkInterface.getNetworkInterfaces().nextElement(); + } + multicastSocket.joinGroup(new InetSocketAddress(group, 4446), netIf); + multicastSocket.setSoTimeout(3000); + + byte[] msgBytes = "SLAVE_READY".getBytes(StandardCharsets.UTF_8); + DatagramPacket readyPacket = new DatagramPacket(msgBytes, msgBytes.length, + group, 4446); + multicastSocket.send(readyPacket); + + byte[] buffer = new byte[1024]; + while (true) { + try { + DatagramPacket packet = new DatagramPacket(buffer, buffer.length); + multicastSocket.receive(packet); + String response = new String(packet.getData(), 0, + packet.getLength(), StandardCharsets.UTF_8); + + System.err.println(response); + if (response.startsWith("MASTER_INFO:")) { + responseReceived.set(true); + break; + } + } catch (SocketTimeoutException e) { + System.out.println("No Master answered. Re-broadcasting/waiting..."); + multicastSocket.send(readyPacket); // Повторяем запрос + } + } + } catch (Exception e) { + System.err.println("Discovery exception: " + e.getMessage()); + return; + } + + try (Socket socket = new Socket("localhost", port); + ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream()); + ObjectInputStream in = new ObjectInputStream(socket.getInputStream())) { + + Task task = (Task) in.readObject(); + out.writeObject(new TaskResult(task.taskId(), false)); + out.flush(); + in.readObject(); + } + } catch (Exception e) { + e.printStackTrace(); + } + }); + + slaveSimulator.start(); + + master.execute(new int[]{1}, 1, 1); + + slaveSimulator.join(5000); + assertTrue(responseReceived.get()); + } + + + @Test + public void workerCrash_returnsTaskToQueue() throws Exception { + int port = 8040; + int[] numbers = {2, 3, 5, 7}; + Master master = new Master(port); + + Thread badSlave = new Thread(() -> { + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + + try (Socket socket = new Socket("localhost", port); + ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream()); + ObjectInputStream in = new ObjectInputStream(socket.getInputStream())) { + in.readObject(); + } catch (Exception e) { + e.printStackTrace(); + } + }); + + AtomicBoolean success = new AtomicBoolean(false); + Thread goodSlave = new Thread(() -> { + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + try (Socket socket = new Socket("localhost", port); + ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream()); + ObjectInputStream in = new ObjectInputStream(socket.getInputStream())) { + + for (int i = 0; i < 2; i++) { + Task task = (Task) in.readObject(); + out.writeObject(new TaskResult(task.taskId(), false)); + out.flush(); + } + success.set(true); + } catch (Exception e) { + e.printStackTrace(); + } + }); + + badSlave.start(); + goodSlave.start(); + + boolean result = master.execute(numbers, 2, 2); + + assertFalse(result); + goodSlave.join(2000); + assertTrue(success.get()); + } +} \ No newline at end of file diff --git a/Task_2_1_2/src/test/java/ru/nsu/nmashkin/task212/SlaveTest.java b/Task_2_1_2/src/test/java/ru/nsu/nmashkin/task212/SlaveTest.java new file mode 100644 index 0000000..53181f8 --- /dev/null +++ b/Task_2_1_2/src/test/java/ru/nsu/nmashkin/task212/SlaveTest.java @@ -0,0 +1,140 @@ +package ru.nsu.nmashkin.task212; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.jupiter.api.Test; + +class SlaveTest { + + private void masterSim(int port, CountDownLatch stopLatch) { + Thread advertiser = new Thread(() -> { + try (DatagramSocket udpSocket = new DatagramSocket()) { + InetAddress group = InetAddress.getByName("230.0.0.1"); + byte[] msgBytes = ("MASTER_START:" + port).getBytes(StandardCharsets.UTF_8); + DatagramPacket packet = new DatagramPacket(msgBytes, msgBytes.length, group, 4446); + + while (stopLatch.getCount() > 0) { + udpSocket.send(packet); + Thread.sleep(500); + } + } catch (Exception e) { + e.printStackTrace(); + } + }); + advertiser.setDaemon(true); + advertiser.start(); + } + + @Test + public void start() throws Exception { + int port = 8003; + AtomicBoolean resultReceived = new AtomicBoolean(false); + CountDownLatch stopAdvertiser = new CountDownLatch(1); + + Thread masterSimulator = new Thread(() -> { + try (ServerSocket server = new ServerSocket(port)) { + masterSim(port, stopAdvertiser); + + try (Socket socket = server.accept(); + ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream()); + ObjectInputStream in = new ObjectInputStream(socket.getInputStream())) { + + stopAdvertiser.countDown(); + + out.writeObject(new Task(99, new int[]{2, 4, 6})); + out.flush(); + + TaskResult result = (TaskResult) in.readObject(); + if (result != null && result.taskId() == 99) { + resultReceived.set(result.hasNonPrime()); + } + + out.writeObject(null); + out.flush(); + } + } catch (Exception e) { + e.printStackTrace(); + } finally { + stopAdvertiser.countDown(); + } + }); + masterSimulator.start(); + + Thread.sleep(200); + + Slave slave = new Slave(2); + slave.start(); + + masterSimulator.join(); + assertTrue(resultReceived.get()); + } + + @Test + public void start_reconnect() throws Exception { + int port = 8025; + AtomicBoolean taskProcessedAfterReconnect = new AtomicBoolean(false); + CountDownLatch masterStartedLatch = new CountDownLatch(1); + CountDownLatch taskDoneLatch = new CountDownLatch(1); + CountDownLatch stopAdvertiser = new CountDownLatch(1); + + Thread slaveThread = new Thread(() -> { + Slave slave = new Slave(2); + slave.start(); + }); + + slaveThread.start(); + Thread.sleep(5000); + + Thread masterDelayedThread = new Thread(() -> { + masterStartedLatch.countDown(); + try (ServerSocket server = new ServerSocket(port)) { + masterSim(port, stopAdvertiser); + + try (Socket socket = server.accept(); + ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream()); + ObjectInputStream in = new ObjectInputStream(socket.getInputStream())) { + + stopAdvertiser.countDown(); + + out.writeObject(new Task(42, new int[]{4, 6, 8})); + out.flush(); + + TaskResult result = (TaskResult) in.readObject(); + if (result != null && result.taskId() == 42) { + taskProcessedAfterReconnect.set(true); + } + + out.writeObject(null); + out.flush(); + taskDoneLatch.countDown(); + + } + } catch (Exception e) { + e.printStackTrace(); + } finally { + stopAdvertiser.countDown(); + } + }); + + masterDelayedThread.start(); + + assertTrue(masterStartedLatch.await(2, TimeUnit.SECONDS)); + assertTrue(taskDoneLatch.await(5, TimeUnit.SECONDS)); + + slaveThread.interrupt(); + slaveThread.join(1000); + + assertTrue(taskProcessedAfterReconnect.get()); + } +} \ No newline at end of file