Threads and Executors

credit for this assignment: Finn Voichick and Dennis Cosgrove

Motivation

The X10 family of programming languages, including Habanero and our own modest V5, provide a less cumbersome way to write parallel programs with async and finish.

Finish has particularly nice semantics: keeping track of and joining all of its descendant tasks. In this lab we will remove the training wheels for a moment to get some experience with some core Java parallel features Thread and ExecutorService.

Tracking all of the descendant tasks provides us with an opportunity to use a thread safe data structure (ConcurrentLinkedQueue) as well as deal with the complications of weakly consistent iterators.

Background

Java 8 Concurrency Tutorial: Threads and Executors

Code to Implement

(Optional But Recommended) Join All Warm Up

ConcurrentLinkedQueue is a thread safe data structure. Multiple threads can add and remove from it without fear. The order they will be in might not be deterministic, but it won't lose anybody. We will end up using a ConcurrentLinkedQueue to track all of the spawned Futures in the #XQuicksort section of this lab.

Sadly, ConcurrentLinkedQueue's documentation reports that:

"Iterators are weakly consistent, returning elements reflecting the state of the queue at some point at or since the creation of the iterator."

This means that if all of the items aren't in the queue when you start iterating through them, then you are not guaranteed to get updated when new items get added.

Therefore, using the standard iterating for loop from ThreadsRightNow will NOT work for ThreadsEventually:

/* package-private */ static int joinAllInQueueViaIteration(Queue<Thread> queue) throws InterruptedException {
	int count = 0;
	for (Thread thread : queue) {
		thread.join();
		count++;
	}
	return count;
}

Think about how you can make sure that all of threads are joined.

Drain the queue by repeatedly polling until it is empty. Be sure to actually join the threads.

class:	ThreadsEventually.java
methods:	joinAllInQueueViaPoll
package:	tnx.warmup.joinall
source folder:	student/src/main/java

method: private static int joinAllInQueueViaPoll(Queue<Thread> queue) (sequential implementation only)

NOTE: this method should return the number of threads joined.

Completing this optional warm up will help you when you implement #XQuicksort.

Testing Your Solution

Correctness

class:	JoinAllTestSuite.java
package:	tnx.warmup.joinall
source folder:	testing/src/test/java

Threads

While we strive to make CSE 231 generally applicable across libraries and languages, it would be madness to have a parallel programming class in Java and not have students know how to create a Thread, start it, and join it. In this section of the lab, you will do just that.

box link: Thread Start and Join

Code To Use

interface ThreadFactory

newThread

class Thread

constructor

start

join

As always, the wiki's reference page can be of help.

SimpleThreadFactory

class:	SimpleThreadFactory.java
methods:	newThread
package:	tnx.lab.thread
source folder:	student/src/main/java

method: Thread newThread(Runnable target) (sequential implementation only)

Create and return a new thread with the target Runnable parameter you are passed.

Do *NOT* start this thread.

Certainly, do *NOT* run this thread.

Do not pass Go. Do not collect $200.

To repeat: just create a new Thread with the target Runnable and return it.

TAgeSum

class:	TAgeSum.java
methods:	sumUpperLowerSplit
package:	tnx.lab.thread
source folder:	student/src/main/java

method: int sumUpperLowerSplit(int[] ages, ThreadFactory threadFactory) (parallel implementation required)

You will need use the passed in ThreadFactory to create a new thread or two (at your preference), start any threads you create, and join them.

Think about where you need to start and join any Threads to ensure both correctness and an appropriate amount of parallelism.

Executors

Code To Use

interface ExecutorService

submit(task)

invokeAll(tasks)

interface Future

get()

NucleobaseCounting

countRangeSequential(chromosome,targetNucleobase,min,maxExclusive)

Slices

createNSlices(data, numSlices).

As always, the wiki's reference page can be of help.

XNucleobaseCount

This part of the assignment should be very familiar as it is to a large degree implementing the Nucleobase Counting assignment with Executors instead of Habanero. It also adds a divide and conquer implementation of nucleobase counting.

class:	XNucleobaseCount.java
methods:	countLowerUpperSplit countNWaySplit countDivideAndConquer countDivideAndConquerKernel
package:	tnx.lab.executor
source folder:	student/src/main/java

Tip:Use countRangeSequential() from Nucleobase_Counting

lower upper split

method: int countLowerUpperSplit(ExecutorService executor, byte[] chromosome, Nucleobase nucleobase) (parallel implementation required)

NOTE: the tests will enforce that you use submit and get.

n-way split

method: int countNWaySplit(ExecutorService executor, byte[] chromosome, Nucleobase nucleobase, int numTasks) (parallel implementation required)

Please feel free to use your own Slices createNSlices(byte[] data, int numSlices). Then use the Java for each loop to iterate over the slices to create your tasks.

NOTE: the tests will enforce that you use invokeAll.

divide-and-conquer

method: int countDivideAndConquer(ExecutorService executor, byte[] chromosome, Nucleobase nucleobase, int threshold) (parallel implementation required)

This method should just get things going by invoking the countDivideAndConquerKernel on the entire chromosome array.

method: int countDivideAndConquerKernel(ExecutorService executor, byte[] chromosome, Nucleobase nucleobase, int min, int maxExclusive, int threshold) (parallel implementation required)

NOTE: the tests will enforce that you use submit and get.

NOTE: When you get down below the threshold and convert from parallel to sequential execution, do NOT feel compelled to build a sequential divide and conquer. Just invoke countRangeSequential on the remaining range. It is not like divide and conquer gives you a performance benefit in counting like it would for sorting.

XQuicksort

Quicksort is an oldie but a goodie. First developed in 1959 and published in 1961 it is still the go to sorting algorithm today. The JDK8 implementation of Arrays.sort(array) is a DualPivotQuicksort.

Quicksort is also amenable to parallelism. Once the partitioning is done, both sides of the pivot can be sorted completely independently in parallel. This lends itself very nicely to the X10 family of languages as you can freely async as you divide and conquer, then join all of the tasks by wrapping it all in a single call to finish.

In this assignment you will mimic this behavior by submitting tasks to an executor, tracking the returned futures in in a ConcurrentLinkedQueue, then invoking get on all of those futures to mimic the single finish.

Warning: ConcurrentLinkQueue's iterators are weakly consistent. Do the Join All Warm Up to gain experience with handling this issue.

RiceX Lecture on Quicksort

Warning:Unlike in Sarkar's and McDowell’s videos which use inclusive maximums, CSE 231 consistently uses exclusive maximums to avoid having to subtract 1 all of the time.

Code To Use

interface Partitioner

partitionRange(data,min,maxExclusive)

class PivotLocation

getLeftSidesUpperExclusive()

getRightSidesLowerInclusive()

Note: Investigate SequentialPartitioner.

XSequentialQuicksorter

class:	XSequentialQuicksorter.java
methods:	sortRange
package:	tnx.lab.executor
source folder:	student/src/main/java

method: public void sortRange(T[] data, Comparator<T> comparator, int min, int maxExclusive) (sequential implementation only)

Warning:Do NOT implement your own partition. Call Partitioner partitionRange method. It will do this work for you.

Warning:Do NOT invoke partitionRange twice. Invoke partitionRange once and catch the return value in a variable. Then use the two methods on the single instance of PivotLocation.

XParallelQuicksorter

class:	XParallelQuicksorter.java
methods:	kernel sortRange
package:	tnx.lab.executor
source folder:	student/src/main/java

method: private void kernel(Queue<Future<?>> futures, T[] data, Comparator<T> comparator, int min, int maxExclusive) (parallel implementation required)

Tip: Be sure to test the isParallelPredicate to determine if the current range length is worthy of parallel processing or if you should fall back to the sequentialSorter.

method: public void sortRange(T[] data, Comparator<T> comparator, int min, int maxExclusive) (sequential implementation only)

Tip: Make sure to use a thread safe data structure like ConcurrentLinkedQueue and NOT an unsafe data structure like LinkedList.

Tip: return null; from your lambdas to invoke the overloaded Callable version of submit(task) to deal with the checked exceptions.

Warning:Do NOT implement your own partition. Call Partitioner partitionRange method. It will do this work for you.

Warning:Do NOT invoke partitionRange twice. Invoke partitionRange once and catch the return value in a variable. Then use the two methods on the single instance of PivotLocation.

(Optional) Parallel Partition Challenge

The partitioning step can also be done in parallel with scan. While not particularly practical, it can get the CPL down to ${\mathcal {O}}(\lg ^{k}{}n)$.

For details on how to complete this challenge, check out: Quicksort_Parallel_Partitioner

Testing Your Solution

Correctness

There is a top-level test suite comprised of sub test suites which can be invoked separately when you want to focus on one part of the assignment.

top level

class:	ThreadsAndExecutorsTestSuite.java
package:	tnx.lab
source folder:	testing/src/test/java

sub

class:	ThreadsTestSuite.java
package:	tnx.lab.thread
source folder:	testing/src/test/java

class:	NucleobaseExecutorTestSuite.java
package:	tnx.lab.executor
source folder:	testing/src/test/java

class:	QuicksortExecutorTestSuite.java
package:	tnx.lab.executor
source folder:	testing/src/test/java

Less Time Consuming Suite

class:	QuicksortExecutorNonWeaklyConsistentIteratorTestSuite.java
package:	tnx.lab.executor
source folder:	testing/src/test/java

Performance

class:	NucleobaseCountTiming.java
package:	tnx.lab.executor
source folder:	src/main/java

class:	QuicksortTiming.java
package:	tnx.lab.executor
source folder:	src/main/java

Rubric

As always, please make sure to cite your work appropriately.

Total points: 100

SimpleThreadFactory subtotal: 5

• Correct newThread (5)

TAgeSum subtotal: 10

• Correct sumUpperLowerSplit (10)

XNucleobaseCount subtotal: 40

• Correct count2WaySplit (10)
• Correct countNWaySplit (15)
• Correct countDivideAndConquer and countDivideAndConquerKernel (15)

XQuicksort subtotal: 35

• Correct sequentialQuicksort and sequentialQuicksortKernel (10)
• Correct parallelQuicksort and parallelQuicksortKernel (25)

Whole project:

• Clarity and efficiency (10)

Pledge, Acknowledgments, Citations

file:

lab-threads-and-executors-pledge-acknowledgments-citations.txt

More info about the Honor Pledge