second Homework

.gitignore

+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### IntelliJ IDEA ###
+.idea/modules.xml
+.idea/jarRepositories.xml
+.idea/compiler.xml
+.idea/libraries/
+*.iws
+*.iml
+*.ipr
+
+### Eclipse ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/
+
+### Mac OS ###
+.DS_Store
\ No newline at end of file

.idea/.gitignore

+# Default ignored files
+/shelf/
+/workspace.xml

.idea/encodings.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Encoding">
+    <file url="file://$PROJECT_DIR$/src/main/java" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/src/main/resources" charset="UTF-8" />
+  </component>
+</project>
\ No newline at end of file

.idea/misc.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ExternalStorageConfigurationManager" enabled="true" />
+  <component name="MavenProjectsManager">
+    <option name="originalFiles">
+      <list>
+        <option value="$PROJECT_DIR$/pom.xml" />
+      </list>
+    </option>
+  </component>
+  <component name="ProjectRootManager" version="2" languageLevel="JDK_1_8" default="true" project-jdk-name="1.8" project-jdk-type="JavaSDK">
+    <output url="file://$PROJECT_DIR$/out" />
+  </component>
+</project>
\ No newline at end of file

.idea/vcs.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>
\ No newline at end of file

pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<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>
+
+    <groupId>org.example</groupId>
+    <artifactId>Ahmad_HW2</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.13.2</version>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>RELEASE</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+</project>
\ No newline at end of file

src/main/java/org/example/Main.java

+package org.example;
+
+import org.example.data.TestData;
+import org.example.parallelsummers.ParallelSummer0;
+import org.example.parallelsummers.ParallelSummer1;
+import org.example.parallelsummers.ParallelSummer2;
+import org.example.parallelsummers.ParallelSummer3;
+
+public class Main {
+    public static void main(String[] args) {
+
+        int size = 100000000;
+        int[] data = TestData.createData(size);
+        long result = SequentialSummer.sum(data);
+        int threadCount = 8;
+        System.out.println("Sequential summer:" + result);
+        System.out.println("Parallel summer 1: " + ParallelSummer0.sum(data, 8));
+        System.out.println("Parallel summer 2: " + ParallelSummer1.sum(data, 8));
+        System.out.println("Parallel summer 3: " + ParallelSummer2.sum(data, 8));
+        System.out.println("Parallel summer 4: " + ParallelSummer3.sum(data, 8));
+    }
+}
\ No newline at end of file

src/main/java/org/example/SequentialSummer.java

+package org.example;
+
+import static org.example.worker.Worker.isPrime;
+
+public class SequentialSummer {
+    public static long sum(int[] data) {
+        long sum = 0;
+        for (int i = 0; i < data.length; i++) {
+            if (isPrime(data[i])){
+                sum= sum+1;
+            }
+        }
+        return sum;
+    }
+}

src/main/java/org/example/data/TestData.java

+package org.example.data;
+
+public class TestData {
+    public static int[] createData(int size) {
+        int[] data =  new int[size];
+        for (int i = 0; i < data.length; i++) {
+            data[i] = i+1;
+        }
+        return data;
+    }
+}

src/main/java/org/example/parallelsummers/ParallelSummer0.java

+package org.example.parallelsummers;
+
+import org.example.summers.SummerThread;
+import org.example.worker.WorkPartitioner;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ParallelSummer0 {
+    public static long sum(int[] data, int threadCount) {
+
+        List<SummerThread> summers =  new ArrayList<>();
+        List<WorkPartitioner.Part> parts =  WorkPartitioner.partitions(data.length, threadCount);
+
+        for (WorkPartitioner.Part part : parts) {
+            summers.add(new SummerThread(data,part));
+        }
+
+        for (SummerThread summerThread : summers) {
+            summerThread.start();
+        }
+
+        for (SummerThread summerThread : summers) {
+            try {
+                summerThread.join();
+            } catch (InterruptedException e) {
+                System.err.println("Thread cannot join!");
+            }
+        }
+
+
+        int sum = 0;
+        for (SummerThread summerThread : summers) {
+            sum += summerThread.getSum();
+        }
+        return sum;
+    }
+
+
+}

src/main/java/org/example/parallelsummers/ParallelSummer1.java

+package org.example.parallelsummers;
+
+import org.example.summers.SummerRunnable;
+import org.example.worker.WorkPartitioner;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ParallelSummer1 {
+    public static long sum(int[] data, int threadCount) {
+
+        List<SummerRunnable> summers =  new ArrayList<>();
+        List<WorkPartitioner.Part> parts =  WorkPartitioner.partitions(data.length, threadCount);
+
+        for (WorkPartitioner.Part part : parts) {
+            summers.add(new SummerRunnable(data,part));
+        }
+
+        for (SummerRunnable SummerRunnable : summers) {
+            SummerRunnable.startThread();
+        }
+
+        for (SummerRunnable SummerRunnable : summers) {
+            try {
+                SummerRunnable.joinThread();
+            } catch (InterruptedException e) {
+                System.err.println("Thread cannot join!");
+            }
+        }
+
+        long sum = 0;
+        for (SummerRunnable summerRunnable : summers) {
+            sum += summerRunnable.getSum();
+        }
+
+        return sum;
+    }
+
+
+}

src/main/java/org/example/parallelsummers/ParallelSummer2.java

+package org.example.parallelsummers;
+
+import org.example.summers.SummerThread;
+import org.example.worker.WorkPartitioner;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class ParallelSummer2 {
+
+    public static long sum(int[] data, int threadCount) {
+
+        List<SummerThread> summers =  new ArrayList<>();
+        List<WorkPartitioner.Part> parts =  WorkPartitioner.partitions(data.length, threadCount);
+
+        for (WorkPartitioner.Part part : parts) {
+            summers.add(new SummerThread(data,part));
+        }
+
+        //creating a pool of 5 threads
+        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+
+        for (SummerThread summer: summers) {
+            executor.execute(summer);
+        }
+
+        executor.shutdown();
+
+        //Waiting for all the thread to finalize
+        while (!executor.isTerminated()) {
+            //System.out.println("Processing....");
+        }
+
+        return summers.stream().mapToLong(SummerThread::getSum).sum();
+    }
+
+}

src/main/java/org/example/parallelsummers/ParallelSummer3.java

+package org.example.parallelsummers;
+
+import org.example.summers.SummerCallable;
+import org.example.worker.WorkPartitioner;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+public class ParallelSummer3 {
+    public static long sum(int[] data, int threadCount) {
+
+        //creating a pool of #threadCount threads
+        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+
+        List<SummerCallable> summers =  new ArrayList<SummerCallable>();
+        List<WorkPartitioner.Part> parts =  WorkPartitioner.partitions(data.length, threadCount);
+
+        for (WorkPartitioner.Part part : parts) {
+            summers.add(new SummerCallable(data,part));
+        }
+
+        List<Future<Long>> results;
+        try {
+            results = executor.invokeAll(summers);
+        } catch (InterruptedException e) {
+            System.err.println("Cannot invoke the threads.");
+            return -1;
+        }
+
+        executor.shutdown();
+
+        long sum = 0;
+        for (Future<Long> future : results) {
+            try {
+                sum += future.get();
+            } catch (InterruptedException | ExecutionException e) {
+                System.err.println("Cannot get the results from threads.");
+                return -2;
+            }
+        }
+        return sum;
+    }
+}

src/main/java/org/example/parallelsummers/ParallelSummer4.java

+package org.example.parallelsummers;
+
+import org.example.summers.SummerCallable;
+import org.example.worker.WorkPartitioner;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+public class ParallelSummer4 {
+    public static long sum(int[] data, int threadCount) {
+
+        //creating a pool of 5 threads
+        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+        List<Future<Long>> results = new ArrayList<Future<Long>>();
+        List<WorkPartitioner.Part> parts =  WorkPartitioner.partitions(data.length, threadCount);
+
+        for (WorkPartitioner.Part part : parts) {
+            results.add(executor.submit(new SummerCallable(data, part)));
+        }
+
+        executor.shutdown();
+
+        while (!executor.isTerminated()) {
+            //System.out.println("Processing...");
+        }
+
+        long sum = 0;
+        for (Future<Long> future : results) {
+            try {
+                sum += future.get();
+            } catch (InterruptedException | ExecutionException e) {
+                System.err.println("Cannot get the results from threads.");
+                return -2;
+            }
+        }
+
+        return sum;
+    }
+
+}

src/main/java/org/example/summers/Summer.java

+package org.example.summers;
+
+
+import org.example.worker.WorkPartitioner.Part;
+import org.example.worker.Worker;
+
+public class Summer extends Worker{
+    protected long sum;
+    public Summer(int[] data, Part part) {
+        super(data, part);
+    }
+    public long getSum() {
+        return sum;
+    }
+}

src/main/java/org/example/summers/SummerCallable.java

+package org.example.summers;
+import java.util.concurrent.Callable;
+import org.example.worker.WorkPartitioner.Part;
+import org.example.worker.Worker;
+
+
+public class SummerCallable extends Summer implements Callable<Long>{
+    public SummerCallable(int[] data, Part part) {
+        super(data, part);
+    }
+
+    @Override
+    public Long call() throws Exception {
+        this.sum = 0;
+        for (int i = start; i < finish; i++) {
+            if (isPrime(data[i])){
+                sum= sum+1;
+            }
+        }
+        return sum;
+    }
+}

src/main/java/org/example/summers/SummerRunnable.java

+package org.example.summers;
+
+import java.util.concurrent.Callable;
+import org.example.worker.WorkPartitioner.Part;
+
+public class SummerRunnable extends Summer implements Runnable{
+
+    private Thread thread;
+
+    public SummerRunnable(int[] data, Part part) {
+        super(data, part);
+    }
+
+    @Override
+    public void run() {
+        sum = 0;
+        for (int i = start; i < finish; i++) {
+            if (isPrime(data[i])){
+                sum= sum+1;
+            }
+        }
+    }
+
+    public void startThread() {
+        thread =  new Thread(this);
+        thread.start();
+    }
+
+    public void joinThread() throws InterruptedException {
+        thread.join();
+    }
+
+}

src/main/java/org/example/summers/SummerThread.java

+package org.example.summers;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.example.worker.WorkPartitioner.Part;
+
+import static org.example.worker.Worker.isPrime;
+
+public class SummerThread extends Thread{
+    private int start;
+    private int finish;
+    private int[] data;
+    private long sum;
+
+    public SummerThread(int[] data, int start, int finish) {
+        this.start = start;
+        this.finish = finish;
+        this.data = data;
+    }
+
+    public SummerThread(int[] data, Part part) {
+        this(data, part.getStart(), part.getFinish());
+    }
+
+    @Override
+    public void run() {
+        sum = 0;
+        for (int i = start; i < finish; i++) {
+            if (isPrime(data[i])){
+                sum= sum+1;
+            }
+        }
+    }
+
+    public long getSum() {
+        return sum;
+    }
+
+}

src/main/java/org/example/worker/WorkPartitioner.java

+package org.example.worker;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class WorkPartitioner {
+    public static List<Part> partitions(int size, int workerCount){
+        List<Part> parts =  new ArrayList<>();
+        int part = (int) Math.ceil( (double)size / workerCount);
+        for (int i = 0; i <workerCount ; i++) {
+            int start = i * part;
+            int finish = Math.min((i+1) * part, size);
+            parts.add(new Part(start, finish));
+        }
+        return parts;
+    }
+
+    public static class Part {
+        private final int start;
+        private final int finish;
+        public Part(int start, int finish) {
+            this.start = start;
+            this.finish = finish;
+        }
+        public int getStart() {
+            return start;
+        }
+        public int getFinish() {
+            return finish;
+        }
+    }
+}

src/main/java/org/example/worker/Worker.java

+package org.example.worker;
+import org.example.worker.WorkPartitioner.Part;
+
+public class Worker {
+
+    protected final int start;
+    protected final int finish;
+    protected final int[] data;
+
+    public Worker(int[] data, int start, int finish) {
+        this.start = start;
+        this.finish = finish;
+        this.data = data;
+    }
+
+    public Worker(int[] data, Part part) {
+        this(data, part.getStart(), part.getFinish());
+    }
+
+    public int getStart() {
+        return start;
+    }
+
+    public int getFinish() {
+        return finish;
+    }
+
+    public int[] getData() {
+        return data;
+    }
+
+    public static boolean isPrime(int number) {
+        if (number <= 1) {
+            return false;
+        }
+
+        for (int i = 2; i <= Math.sqrt(number); i++) {
+            if (number % i == 0) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+}

src/test/java/ParallelSummersTest.java

+import junit.framework.TestCase;
+import org.example.SequentialSummer;
+import org.example.data.TestData;
+import org.example.parallelsummers.*;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import static org.junit.Assert.assertEquals;
+
+public class ParallelSummersTest extends TestCase {
+    private static int size;
+    private static int[] data;
+    private static long result;
+
+
+    public static void initialize() {
+        size = 1000000;
+        data = TestData.createData(size);
+        result = SequentialSummer.sum(data);
+    }
+
+    private static void testAll(int threadCount) {
+        initialize();
+        assertEquals(result, ParallelSummer0.sum(data,threadCount));
+        assertEquals(result, ParallelSummer1.sum(data,threadCount));
+        assertEquals(result, ParallelSummer2.sum(data,threadCount));
+        assertEquals(result, ParallelSummer3.sum(data,threadCount));
+        assertEquals(result, ParallelSummer4.sum(data,threadCount));
+    }
+
+
+   public void testSingleThread() {
+        testAll(1);
+    }
+
+
+    public void testTwoThread() {
+        testAll(2);
+    }
+
+
+    public void testFourThread() {
+        testAll(4);
+    }
+
+
+    public void testEightThread() {
+        testAll(8);
+    }
+
+
+    public void testSixteenThread() {
+        testAll(16);
+    }
+
+    public void testParallel()
+    {
+        int size = 10000000;
+        int[] data = TestData.createData(size);
+        long startTime1 = System.currentTimeMillis();
+        System.out.println("Parallel summer 1: " + ParallelSummer0.sum(data, 4));
+        long endTime1 = System.currentTimeMillis();
+        long executionTime1 = endTime1 - startTime1;
+        double executionTimeSeconds1 = (double) executionTime1 / 1000.0;
+        System.out.println("Execution Time: " + executionTimeSeconds1 + " seconds");
+
+        long startTime2 = System.currentTimeMillis();
+        System.out.println("Parallel summer 2: " + ParallelSummer1.sum(data, 4));
+        long endTime2 = System.currentTimeMillis();
+        long executionTime2 = endTime2 - startTime2;
+        double executionTimeSeconds2 = (double) executionTime2 / 1000.0;
+        System.out.println("Execution Time: " + executionTimeSeconds2 + " seconds");
+
+        long startTime3 = System.currentTimeMillis();
+        System.out.println("Parallel summer 3: " + ParallelSummer2.sum(data, 4));
+        long endTime3 = System.currentTimeMillis();
+        long executionTime3 = endTime3 - startTime3;
+        double executionTimeSeconds3 = (double) executionTime3 / 1000.0;
+        System.out.println("Execution Time: " + executionTimeSeconds3 + " seconds");
+
+        long startTime4 = System.currentTimeMillis();
+        System.out.println("Parallel summer 4: " + ParallelSummer3.sum(data, 4));
+        long endTime4 = System.currentTimeMillis();
+        long executionTime4 = endTime4 - startTime4;
+        double executionTimeSeconds4 = (double) executionTime4 / 1000.0;
+        System.out.println("Execution Time: " + executionTimeSeconds4 + " seconds");
+
+        long startTime5 = System.currentTimeMillis();
+        System.out.println("Parallel summer 5: " + ParallelSummer4.sum(data, 4));
+        long endTime5 = System.currentTimeMillis();
+        long executionTime5 = endTime5 - startTime5;
+        double executionTimeSeconds5 = (double) executionTime5 / 1000.0;
+        System.out.println("Execution Time: " + executionTimeSeconds5 + " seconds");
+    }
+}