[WIP] Espresso Parallel Implementation

cbrew/espresso/src/main/java/org/jbrew/cbrew/espresso/EspressoOperation.java → cbrew/espresso/src/main/java/org/jbrew/cbrew/espresso/IntOperation.java

@@ -3,7 +3,7 @@
 import java.util.Arrays;
 import java.util.stream.IntStream;
 
-public abstract class EspressoOperation {
+public abstract class IntOperation extends NumericOperation<Integer>{
 
 	public final static int DEFAULT_NUM_THREADS = 8;
 
@@ -13,9 +13,14 @@ public abstract class EspressoOperation {
 	 * @param arr - the <code>int</code> array upon which the operation will occur.
 	 * @return an <code>int</code> with the results of the operation.
 	 */
+	public final Integer performOperation(Integer[] arr) {
+		return performSequential(mapToIntArray(arr));
+	}
+
 	public final int performOperation(int[] arr) {
-		return performSequential(arr);
+		return performSequential((arr));
 	}
+
 
 	/**
 	 * The variable argument implementation of {@link #performOperation(int[])}.
@@ -62,6 +67,10 @@ public final int performOperation(int[] arr, boolean parallel) {
 	public final int performOperation(int[] arr, int numThreads) {
 		return performParallel(arr, numThreads);
 	}
+
+	private final int[] mapToIntArray(Integer[] arr) {
+		return Arrays.stream(arr).parallel().mapToInt(Integer::intValue).toArray();
+	}
 
 	protected abstract int performSequential(int[] arr);
 

cbrew/espresso/src/main/java/org/jbrew/cbrew/espresso/SumOperation.java → cbrew/espresso/src/main/java/org/jbrew/cbrew/espresso/IntSumOperation.java

@@ -2,12 +2,12 @@
 
 import org.jbrew.core.annotations.UnderDevelopmentInvestigation;
 
-public class SumOperation extends EspressoOperation {
+public class IntSumOperation extends IntOperation {
 
 	private final static int DEFAULT_NUM_THREADS = 5;
 
 	static {
-		System.loadLibrary("agg_summer");
+		System.loadLibrary("summer");
 	}
 
 	private native int sumSequential(int size, int[] arr);

cbrew/espresso/src/main/java/org/jbrew/cbrew/espresso/NumericOperation.java

@@ -0,0 +1,27 @@
+package org.jbrew.cbrew.espresso;
+
+public abstract class NumericOperation<T extends Number> {
+
+	/**
+	 * Performs the appropriate action on the given array of type <code>T</code>.
+	 * @param arr - the <code>T</code> array upon which the operation will occur.
+	 * @return a <code>T</code> with the results of the operation.
+	 */
+	public abstract T performOperation(T[] arr);
+
+	/**
+	 * The variable argument implementation of {@link #performOperation(Number[])}.
+	 * 
+	 * @param arr - a <code>T</code> array, set of <code>T</code> arrays,
+	 *            integers, or a heterogeneous mix of any of the following, provided
+	 *            that they are of type <code>T</code>.
+	 * @return an <code>T</code> with the results of the operation.
+	 */
+	//@SuppressWarnings("unchecked")
+	//public abstract T performOperation(T[]... arr);
+
+	//protected abstract void performSequential();
+
+	//protected abstract int performParallel(int[] arr, int numThreads);
+
+}

cbrew/espresso/src/test/java/org/jbrew/cbrew_tests/espresso/SumTest.java → cbrew/espresso/src/test/java/org/jbrew/cbrew_tests/espresso/IntSumOperationTest.java

@@ -6,18 +6,18 @@
 import java.util.List;
 import java.util.Random;
 
-import org.jbrew.cbrew.espresso.SumOperation;
+import org.jbrew.cbrew.espresso.IntSumOperation;
 import org.jbrew.core.annotations.Testing;
 import org.junit.Ignore;
 import org.junit.Test;
 
 @Testing
-public class SumTest {
+public class IntSumOperationTest {
 
 	@Test
 	public void sumRecBasicTest() {
 		int[] arr = {100, 50, 50};
-		SumOperation op = new SumOperation();
+		IntSumOperation op = new IntSumOperation();
 		int sum = op.performOperation(arr);
 		assertEquals(200, sum);
 	}
@@ -29,7 +29,7 @@ public void sumRecMedTest() {
 			arr[index] = new Random().nextInt();
 		}
 		int trueSum = findSum(arr);
-		SumOperation op = new SumOperation();
+		IntSumOperation op = new IntSumOperation();
 		int testSum = op.performOperation(arr);
 		System.out.println("Expecting: " + trueSum + ", Received: " + testSum);
 		assertEquals(trueSum, testSum);
@@ -42,7 +42,7 @@ public void sumStreamsMed2Test() {
 			arr[index] = new Random().nextInt();
 		}
 		int trueSum = findSum(arr);
-		SumOperation op = new SumOperation();
+		IntSumOperation op = new IntSumOperation();
 		int testSum = op.performOperation(arr);
 		System.out.println("Expecting: " + trueSum + ", Received: " + testSum);
 		assertEquals(trueSum, testSum);
@@ -55,7 +55,7 @@ public void sumStreamsMed3Test() {
 			arr[index] = new Random().nextInt();
 		}
 		int trueSum = findSum(arr);
-		SumOperation op = new SumOperation();
+		IntSumOperation op = new IntSumOperation();
 		int testSum = op.performOperation(arr);
 		System.out.println("Expecting: " + trueSum + ", Received: " + testSum);
 		assertEquals(trueSum, testSum);
@@ -68,7 +68,7 @@ public void sumStreamsLargeTest() {
 			arr[index] = new Random().nextInt();
 		}
 		int trueSum = findSumStreams(arr);
-		SumOperation op = new SumOperation();
+		IntSumOperation op = new IntSumOperation();
 		int testSum = op.performOperation(arr);
 		System.out.println("Expecting: " + trueSum + ", Received: " + testSum);
 		assertEquals(trueSum, testSum);
@@ -81,7 +81,7 @@ public void sumStreamsLarge2Test() {
 			arr[index] = new Random().nextInt();
 		}
 		int trueSum = findSumStreams(arr);
-		SumOperation op = new SumOperation();
+		IntSumOperation op = new IntSumOperation();
 		int testSum = op.performOperation(arr);
 		System.out.println("Expecting: " + trueSum + ", Received: " + testSum);
 		assertEquals(trueSum, testSum);
@@ -94,7 +94,7 @@ public void sumStreamsLarge3Test() {
 			arr[index] = new Random().nextInt();
 		}
 		int trueSum = findSumStreams(arr);
-		SumOperation op = new SumOperation();
+		IntSumOperation op = new IntSumOperation();
 		int testSum = op.performOperation(arr);
 		System.out.println("Expecting: " + trueSum + ", Received: " + testSum);
 		assertEquals(trueSum, testSum);
@@ -108,7 +108,7 @@ public void sumStreamsLarge4Test() {
 			arr[index] = new Random().nextInt();
 		}
 		int trueSum = findSumStreams(arr);
-		SumOperation op = new SumOperation();
+		IntSumOperation op = new IntSumOperation();
 		int testSum = op.performOperation(arr);
 		System.out.println("Expecting: " + trueSum + ", Received: " + testSum);
 		assertEquals(trueSum, testSum);
@@ -122,7 +122,7 @@ public void sumStreamsLarge5Test() {
 			arr[index] = new Random().nextInt();
 		}
 		int trueSum = findSumStreams(arr);
-		SumOperation op = new SumOperation();
+		IntSumOperation op = new IntSumOperation();
 		int testSum = op.performOperation(arr);
 		System.out.println("Expecting: " + trueSum + ", Received: " + testSum);
 		assertEquals(trueSum, testSum);
@@ -136,7 +136,7 @@ public void sumStreamsLarge6Test() {
 			arr[index] = new Random().nextInt();
 		}
 		int trueSum = findSumStreams(arr);
-		SumOperation op = new SumOperation();
+		IntSumOperation op = new IntSumOperation();
 		int testSum = op.performOperation(arr);
 		System.out.println("Expecting: " + trueSum + ", Received: " + testSum);
 		assertEquals(trueSum, testSum);

summer-devel.c

@@ -0,0 +1,73 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <pthread.h>
+
+#define THREADS 8
+
+int part = 0;
+int sum[THREADS] = { 0 };
+int arr[] = {1, 2, 3, 1, 1, 6, 7, 8, 9, 20, 11, 12, 13, 14, 15, 16};
+
+int* arrp;
+int* sizep;
+
+int sumSequential(int size, int arr[]){
+	int accumulator=0;
+	int *p;
+	for(p=&arr[0]; p<&arr[size]; p++){
+		accumulator += *p;
+	}
+	return accumulator;
+}
+
+void* helper(void *argv){
+	int thread_part = part++;
+	//each thread computes 1/4 of the 16 element array.
+	for(int index=thread_part * (*sizep / THREADS); index < (thread_part+1) * (*sizep / THREADS); index++)
+		sum[thread_part] += arrp[index];
+	return NULL;
+}
+
+int sumParallel(int size, int arr[]){
+	int* p = &arr[0];
+	puts("BASIC:: Printing pointer array vals:");
+	for(p; p<&arr[size]; p++){
+		printf("%d\n", *p);
+	}
+
+	puts("");
+
+	arrp = &arr[0];
+	sizep = &size;
+	puts("HARD:: Printing pointer array vals:");
+	for(int index=0; index<size; index++){
+		printf("%d\n", arrp[index]);
+	}
+
+	int total_sum=0;
+	//int *p;
+	pthread_t threads[THREADS];
+	//spawn 4 threads
+	for(int index=0; index<THREADS;index++)
+		pthread_create(&threads[index], NULL, helper, (void*)NULL);
+	//join the threads once they're done
+	for(int index=0; index<THREADS; index++){
+		pthread_join(threads[index], NULL);
+	}
+
+	//sum up the values placed in total_sum
+	for(int index=0; index<THREADS; index++)
+		total_sum += sum[index];
+
+	//free(&arrp);
+	//free(&sizep);
+	return total_sum;
+}
+
+int main(int numArgs, char **argc){
+	int sum1 = sumSequential(16, arr);
+	int sum2 = sumParallel(16, arr);
+	printf("Sum Sequential= %d\n", sum1);
+	printf("Sum Parallel= %d\n", sum2);
+	return 0;
+}